Kafka Topic

Overview

The page outlines the parameters that should be included in your listener configuration when setting up a real time sync with a Kafka stream source. This process currently supports JSON string or AVRO serialized format. To listen in on multiple topics, you will need to configure multiple listener configs.

Configuring the sync

All of the confiuration parameters and variables for a Kafka real-time sync are the same as those outlined in the Kafka batch sync documentation. The only difference is that you must also set up your listener configuration as per the next section.

The Listener configuration

info

In Cinchy v5.7+, configuring the listener can be done directly in the Connections UI, however for multiple listener requirements you must still add additional configurations in the Listener Config table.

To listen in on multiple topics, you will need to configure multiple listener configs.

To set up an Stream Source, you must set up a Listener Configuration. The below table describes the parameters and their relevant descriptions.

Listener Config

The following column parameters can be found in the Listener Config table:

Parameter	Description	Example
Name	Mandatory. Provide a name for your listener config.	real-time sync
Event Connector Type	Mandatory. Select your Connector type from the drop-down menu.
Topic	Mandatory. This field is expecting a JSON formatted value specific to the connector type you are configuring.	See the Topic tab.
Connection Attributes	Mandatory. This field is expecting a JSON formatted value specific to the connector type you are configuring.	See the Connection Attributes tab.
Status	Mandatory. Set to "Enabled" to activate the listener. Leave on "Disabled" until you are ready to start syncing.	Enabled/Disabled
Running Status	Read-only. Shows the current state of the listener as Starting, Running, or Failed. This is automatically managed by the system. For more information, see the Listener status section.	Running
Active	Managed by User/System. Indicates whether the listener is set to retry after failure ("Yes") or has stopped attempting to sync and requires user intervention ("No").	Yes/No
Data Sync Config	Mandatory. This drop-down will list all the data syncs on your platform. Select the one that you want to use for your real-time sync.	CDC Data Sync
Subscription Expires On	Salesforce Stream Sources only. This field is a timestamp that's auto populated when it has successfully subscribed to a topic.
Message	Auto-populated. This field reports errors that occur during the sync.
Auto Offset Reset	Earliest, Latest, None. Determines where to start reading events if there is no last message ID or if it's invalid. Can be adjusted post-configuration. Learn more	Earliest ,Latest, None

Topic

The below table can be used to help create your Topic JSON needed to set up a real-time sync.

Parameter	Description	Example
topicName	Mandatory. This is the Kafka topic name to listen messages on.
messageFormat	Optional. Put "AVRO" if your messages are serialized in AVRO, otherwise leave blank.

Example Topic JSON

{
  "topicName": "<(mandatory) kafka topic name to listen messages on>",
  "messageFormat": "<(optional) Put AVRO if the messages are serialized in AVRO, otherwise leave blank>"
}

Connection Attributes

The below table can be used to help create your Connection Attributes JSON needed to set up a real-time sync.

Parameter	Description
bootstrapServers	List the Kafka bootstrap servers in a comma-separated list. This should be in the form of host:port
saslMechanism	One of PLAIN, SCRAM-SHA-256, SCRAM-SHA-512, or OAuthBearer. SCRAM-SHA-256 must be formatted as: SCRAMSHA256 SCRAM-SHA-512 must be formatted as: SCRAMSHA512 Omit this key entirely for a connection whose only credential is a client certificate (see Broker TLS).
saslPassword	The password for your chosen SASL mechanism
saslUsername	The username for your chosen SASL mechanism.
url	This is required if your data follows a schema when serialized in AVRO. It is a comma-separated list of URLs for schema registry instances that are used to register or lookup schemas.
basicAuthCredentialsSource	Specifies the Kafka configuration property "schema.registry.basic.auth.credentials.source" that provides the basic authentication credentials. This can be "UserInfo" | "SaslInherit"
basicAuthUserInfo	Basic Auth credentials specified in the form of username:password
sslKeystorePassword	This is the client keystore (PKCS#12) password.
securityProtocol	Kafka supports cluster encryption and authentication, which can encrypt data-in-transit between your applications and Kafka. Use this field to specify which protocol will be used for communication between client and server. Cinchy currently supports the following options: Plaintext, Ssl, SaslPlaintext, or SaslSsl. Plaintext: Unauthenticated, non-encrypted. Ssl: TLS-based encryption, with a client certificate as the only credential (mutual TLS). SaslPlaintext: SASL-based authentication, non-encrypted. SaslSsl: SASL-based authentication, TLS-based encryption. If no parameter is specified, this will default to Plaintext.

{
  "bootstrapServers": "<(mandatory) kafka bootstrap servers in a comma-separated list in the form of host:port>",
  "saslMechanism": "<PLAIN | SCRAMSHA256 | SCRAMSHA512 | OAuthBearer>",
  "saslPassword": "",
  "saslUsername": "",
  "schemaRegistrySettings": {
    "url": "<(optional) required if your data follows a schema when serialized in Avro. A comma-separated list of URLs for schema registry instances that are used to register or lookup schemas>",
    "basicAuthCredentialsSource": "<(optional) the Kafka configuration property schema.registry.basic.auth.credentials.source that provides the basic authentication credentials, either UserInfo or SaslInherit>",
    "basicAuthUserInfo": "<(optional) Basic Auth credentials specified in the form of username:password>",
    "sslKeystorePassword": "<(optional) the client keystore (PKCS#12) password>"
  },
  "securityProtocol": "<Plaintext | Ssl | SaslPlaintext | SaslSsl>"
}

Broker TLS

Supply TLS material as PEM content rather than as files on the listener host. Set securityProtocol to Ssl for a connection whose only credential is the client certificate, or to SaslSsl to combine TLS with a SASL mechanism.

Parameter	Description
sslCaPem	The CA chain that signs the broker's certificate, as PEM content. Needed when the broker's certificate is not signed by a public CA.
sslCertificatePem	The client certificate this listener presents, as PEM content. Required for mutual TLS.
sslKeyPem	The private key for that client certificate, as PEM content.
sslKeyPassword	The passphrase for the private key, if the key is encrypted.

Example: mutual TLS, with the certificate as the only credential

{
  "bootstrapServers": "broker.example.com:9093",
  "securityProtocol": "Ssl",
  "sslCaPem": "-----BEGIN CERTIFICATE-----\n...\n-----END CERTIFICATE-----",
  "sslCertificatePem": "-----BEGIN CERTIFICATE-----\n...\n-----END CERTIFICATE-----",
  "sslKeyPem": "-----BEGIN PRIVATE KEY-----\n...\n-----END PRIVATE KEY-----"
}

Note that saslMechanism is absent. Setting it to a mechanism you do not intend to use, or to a value the listener cannot read, is rejected when the configuration loads.

OAUTHBEARER

Set saslMechanism to OAuthBearer and securityProtocol to SaslSsl, then choose how the listener obtains its token with saslOauthbearerMethod:

Value	Meaning
Default	librdkafka builds an unsecured token from saslOauthbearerConfig, rather than Cinchy fetching one. See Default method.
Oidc	Cinchy fetches a token from saslOauthbearerTokenEndpointUrl, using either a client secret or a signed JWT client assertion.

Default method

Default selects librdkafka's built-in unsecured token handler, which assembles a token from saslOauthbearerConfig without contacting a provider. Nothing signs that token, so a broker accepts it only where it has been configured to trust unsecured JWTs. Use this for development and testing against a local broker, and Oidc everywhere else.

Parameter	Description
saslOauthbearerConfig	The token's contents, as space-separated name=value pairs. The names recognized are principal, principalClaimName (defaults to sub), scope, scopeClaimName (defaults to scope) and lifeSeconds (defaults to 3600). A scope takes a comma-separated list. An extension_NAME=value pair travels to the broker as a SASL extension.
enableSaslOauthbearerUnsecureJwt	Mandatory for Default. Set it to true to turn the handler on. It defaults to false, which leaves the client with no way to produce a token, and authentication fails.

Example: unsecured token, against a development broker

{
  "bootstrapServers": "localhost:9092",
  "securityProtocol": "SaslPlaintext",
  "saslMechanism": "OAuthBearer",
  "saslOauthbearerMethod": "Default",
  "enableSaslOauthbearerUnsecureJwt": true,
  "saslOauthbearerConfig": "principal=cinchy-listener scope=kafka lifeSeconds=600"
}

Token endpoint

Parameter	Description
saslOauthbearerTokenEndpointUrl	Mandatory for Oidc. The token endpoint. Must be HTTPS, except on a loopback address.
saslOauthbearerClientId	Mandatory for Oidc. The client ID registered with your identity provider.
saslOauthbearerClientSecret	The client secret, for providers that issue one. Leave it out when using a client assertion.
saslOauthbearerScope	The scope requested from the token endpoint, if your provider requires one.
saslOauthbearerExtensions	SASL extensions sent to the broker with the token, as key=value pairs separated by commas. A value cannot contain a comma.

Signed JWT client assertion

Provide a private key instead of a client secret and Cinchy signs an RFC 7523 section 2.2 client assertion (private_key_jwt) for each token request. This is what providers mean when they say they do not issue client secrets.

Parameter	Description
saslOauthbearerAssertionPrivateKeyPem	The signing key as PEM content. Its presence, with the default grant, selects the client assertion flow.
saslOauthbearerAssertionPrivateKeyFile	A path to the signing key on the listener host, as an alternative to PEM content.
saslOauthbearerAssertionPrivateKeyPassphrase	The passphrase for the signing key, if the key is encrypted.
saslOauthbearerAssertionAlgorithm	RS256 (default), RS384, RS512, PS256, PS384, PS512, ES256, ES384 or ES512. Financial-grade providers reject RS256 and require PS256 or ES256.
saslOauthbearerGrantType	ClientCredentials (default) sends the assertion as client authentication. JwtBearer sends it as the authorization grant itself (RFC 7523 section 2.1), which few providers accept.

Identifying your key to the provider

A provider that holds more than one registered public key needs a hint to know which one verifies the assertion. Providers commonly begin to require this at the first key rotation, not at first registration.

Parameter	Description
saslOauthbearerAssertionKeyId	Emitted as the assertion's kid header. Populate it only when the provider publishes that key ID, since a kid a provider cannot resolve fails outright.
saslOauthbearerAssertionCertificatePem	The certificate matching the signing key, as PEM content or a file path. Its thumbprint is emitted in the header named below.
saslOauthbearerAssertionThumbprintHeader	x5t (SHA-1, the default) or x5t#S256 (SHA-256). RFC 7515 ties the digest to the header name, so this one setting picks both.
saslOauthbearerAssertionX5tThumbprint	A pre-computed thumbprint, for providers that publish the thumbprint rather than the certificate. Takes precedence over the certificate above.

Provider-specific extras

Parameter	Description
saslOauthbearerAssertionClaimIss	The assertion's iss claim. Defaults to the client ID, which is what OIDC requires.
saslOauthbearerAssertionClaimSub	The assertion's sub claim. Defaults to the client ID.
saslOauthbearerAssertionClaimAud	The assertion's aud claim. Defaults to the token endpoint URL. Some providers require their issuer identifier instead.
saslOauthbearerAssertionClaimExpSeconds	How long the assertion is valid, in seconds. Defaults to 300.
saslOauthbearerAssertionClaimNbfSeconds	Subtracted from the signing time to produce an nbf claim. Omit it and no nbf is sent.
saslOauthbearerAssertionClaimJtiInclude	Whether to include a unique jti claim. Defaults to true, and providers that detect replay require it.
saslOauthbearerAssertionAdditionalClaims	Extra claims merged into the assertion, as a JSON object. A claim named here overrides the standard claim of the same name.
saslOauthbearerTokenRequestParameters	Extra form parameters on the token request, as a JSON object. Some providers require an audience or resource parameter.
saslOauthbearerTokenRequestHeaders	Extra HTTP headers on the token request, as a JSON object. Values can be stored encrypted.

Token endpoint behind mutual TLS

Some providers require a client certificate to reach the token endpoint. That certificate is separate from the broker connection.

Parameter	Description
saslOauthbearerTokenEndpointCertificatePem	The client certificate for the token endpoint, as PEM content or a file path.
saslOauthbearerTokenEndpointCertificateKeyPem	Its private key, as PEM content or a file path.
saslOauthbearerTokenEndpointCertificateKeyPassphrase	The passphrase for that key, if it is encrypted.

Example: OAUTHBEARER with a signed JWT client assertion

{
  "bootstrapServers": "broker.example.com:9095",
  "securityProtocol": "SaslSsl",
  "saslMechanism": "OAuthBearer",
  "saslOauthbearerMethod": "Oidc",
  "saslOauthbearerTokenEndpointUrl": "https://idp.example.com/oauth2/token",
  "saslOauthbearerClientId": "cinchy-listener",
  "saslOauthbearerAssertionPrivateKeyPem": "-----BEGIN PRIVATE KEY-----\n...\n-----END PRIVATE KEY-----",
  "saslOauthbearerAssertionAlgorithm": "PS256",
  "saslOauthbearerAssertionKeyId": "cinchy-signing-key-1",
  "saslOauthbearerAssertionClaimJtiInclude": true,
  "saslOauthbearerScope": "kafka",
  "sslCaPem": "-----BEGIN CERTIFICATE-----\n...\n-----END CERTIFICATE-----"
}

The minimum for a standards-conformant provider is bootstrapServers, securityProtocol, saslMechanism, saslOauthbearerMethod, saslOauthbearerTokenEndpointUrl, saslOauthbearerClientId and one of saslOauthbearerClientSecret or saslOauthbearerAssertionPrivateKeyPem. Everything else in the tables above exists for providers that ask for more than the specification requires.

Storing values encrypted

Each of the values listed below can be pasted as plaintext or as an encrypted string. The listener detects which it has and decrypts when the configuration loads, so no key or flag accompanies the value in the JSON.

To encrypt a value, run the CLI's encrypt command against the instance that will run the listener, then paste the string it returns in place of the plaintext. Your Cinchy instance performs the encryption rather than the CLI, so a string encrypted on one environment cannot be decrypted on another.

These values can be encrypted:

Category	Keys
Connection	bootstrapServers
SASL credentials	saslUsername, saslPassword, saslOauthbearerConfig, saslOauthbearerClientSecret
Client assertion	saslOauthbearerAssertionPrivateKeyPem, saslOauthbearerAssertionPrivateKeyPassphrase
Token endpoint TLS	saslOauthbearerTokenEndpointCertificateKeyPem, saslOauthbearerTokenEndpointCertificateKeyPassphrase, and the values inside saslOauthbearerTokenRequestHeaders
Broker TLS	sslCaPem, sslCertificatePem, sslKeyPem, sslKeyPassword
Kerberos	saslKerberosPrincipal, saslKerberosServiceName