Kafka Receiver

Apache Kafka is an open-source distributed event-streaming data platform. The Kafka Receiver enables a STEP platform integrated with Apache Kafka to use a Kafka topic. For more information about Apache Kafka, search the web.

The Kafka Receiver option is only available in an inbound integration endpoint (IIEP).

The Kafka Receiver uses background processes to move messages read from one partition on topics between STEP and the Kafka queue. To use Kafka to consume small messages of individual objects without the overhead of individual background processes, refer to the Kafka Streaming Receiver topic.

Prerequisites

Before setting up Kafka Receiver, read the Considerations for Setting Up Kafka Receiver topic.
Prior to configuration, clicking the Bootstrap Server(s) dropdown parameter displays the required server name. Provide a selection for the dropdown parameter via the sharedconfig.properties file on the STEP application server using the case-sensitive Kafka.Server configuration property. If connecting to a cluster, use a comma-separated list.

The following example shows two Kafka server configurations where the first server is a cluster:
```
Kafka.Server.1=mybroker1a:9092,mybroker1b:9092
Kafka.Server.2=mybroker2a:9094
```
Prior to configuration, clicking the Topic dropdown parameter displays the available topics. Provide a selection for the dropdown parameter via the sharedconfig.properties file on the STEP application server using the case-sensitive Kafka.Topic configuration property. For example:
```
Kafka.Topic.1=my-topic
```
OPTIONAL: Prior to configuration, clicking the Consumer Group ID dropdown parameter displays the available group IDs. Use this setting if you want to overwrite the auto-generated group ID. Provide a selection for the dropdown parameter via the sharedconfig.properties file on the STEP application server using the case-sensitive Kafka.GroupId configuration property. For example:
```
Kafka.GroupId.1=my-group-id
```
Prior to configuration, clicking the Keystore Location dropdown parameter displays the required property name. Provide a selection for the dropdown parameter via the sharedconfig.properties file on the STEP application server using the case-sensitive Kafka.SSLKeyStoreLocation property. For example:
```
Kafka.SSLKeyStoreLocation.1=[/[path]/key_store.jks].
```
Prior to configuration, clicking the Truststore Location dropdown parameter displays the required property name. Provide a selection for the dropdown parameter via the sharedconfig.properties file on the STEP application server using the case-sensitive Kafka.SSLTrustStoreLocation property. For example:
```
Kafka.SSLTrustStoreLocation.1=[/[path]/trust_store.jks].
```
SASL / OAuth 2.0 can be configured for STEP with Kafka using the ExtraDriverOptions property to authenticate securely via bearer tokens. A business function allows integration with the Kafka Receiver by handling the OAuth authentication and returning a HashMap containing the Bearer Token and other details.

Configure the following case-sensitive sharedconfig.properties entries within the value of the ExtraDriverOptions as shown in the example below, which can be copied:
- Kafka.Receiver.[IIEP_ID].ExtraDriverOptions - indicate the IIEP ID.
- stibo.authentication.function - set the JavaScript business function ID. Refer to the Business Function Example for Generating an OAuth Bearer Token section in the Kafka Delivery Method topic.
- security.protocol - indicate either SASL_PLAINTEXT (while testing) or SASL_SSL (in production) for the SSL transport layer.
For example, in the property definition below, ‘MyInboundKafka’ is the ID of the IIEP, ‘HydraAuthFunction’ is the ID of the business function, and ‘SASL_PLAINTEXT’ is the testing value for the security protocol. The rest of the property values are configured as displayed:
Copy
```
Kafka.Receiver.MyInboundKafka.ExtraDriverOptions=sasl.mechanism=OAUTHBEARER,sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required;,sasl.login.callback.handler.class=com.stibo.integration.kafka.domain.impl.auth.OAuthFunctionCallbackHandler,stibo.authentication.function=HydraAuthFunction,security.protocol=SASL_PLAINTEXT    
```

Use the 'Copy' option to add this property for your system and then edit it for your environment.

Configuration

While STEPXML is recommended for the messaging format, other formats supported for integration with STEP are allowed. The descriptions below have been written for STEPXML.

After completing the prerequisite steps, edit the receiver of the IIEP. Use the following parameters to configure the IIEP. For information on a parameter, hover over the parameter field to display help text.

For the Receiver, choose Kafka Receiver.
In Consumer Group ID, select the ID to be used to overwrite the auto-generated ID.
For the Bootstrap Server(s), select the server(s) where the Kafka broker instances used by the endpoint are running.
In Topic, select the topic used by this endpoint.
For Use Kafka To Manage Offset, select an option:
- Yes - use the Kafka offset management mechanism instead of the STEP-owned one. This makes it easy to know when Kafka can purge messages. For more information, refer to Kafka Topic Offset on the web.
- No - (the default) to have the offset managed completely in STEP with no commits done to Kafka.
Important: When switching from having STEP manage the offset to having Kafka do it, upon invoking the IIEP, all messages that Kafka has available for a given topic will be re-fetched. To prevent this, you can manually tell the Kafka side what the offset should be for your 'consumer group.' The offset is managed individually for each consumer group, and each Kafka IIEP in STEP is its own consumer group.

Use the kafka-consumer-groups.sh script to list the Kafka consumer groups and identify the one that has the name of your STEP system and the ID of your integration endpoint in its ID. Use the same script and the ID of the consumer group for your STEP IIEP to set the desired offset for your consumer group (this can only be done after the STEP IIEP has been deactivated for at least 15 minutes).
For Max Events To Read Per Poll, enter the maximum number of events to read out of Kafka per invocation.

Note: Setting this too high can require allocating additional memory in the system.
In Decompress Message Content, optionally select a decompression option:
- None - message content will remain compressed.
- LZ4 - lossless data compression algorithm. Search the web for more information.
If required, check the Use SSL checkbox to activate the additional parameters:
- In Keystore Location, if SSL encryption is required, select an SSL encrypted connection to Kafka. Otherwise, leave this parameter blank.
- In Keystore Password, enter the password for keystore if required.
- In Keystore PrivateKey Password, enter the password of the private key in the keystore file, if needed.
- In Truststore Location, if SSL encryption is required, select an SSL encrypted connection to the Kafka. Otherwise, leave this parameter blank.
- In Truststore Password, enter the password for truststore if required.
In Topic Offset, disable the IIEP to edit the value. The Kafka offset tracks the sequential order that messages are received by Kafka topics. Enter the offset from which STEP should read messages.

Note: When 'Use Kafka to Manage Offset' is set to 'Yes' as described above, the Topic Offset parameter is not applicable and removed from the dialog. This is because commits happen after background processes are started for polled messages, and as these messages are consumed, STEP communicates to Kafka that the message has been consumed. Kafka then increments the offset.
Click the Next button to continue with the IIEP - Configure Endpoint and subsequent steps.

Integration Endpoint Log

The Kafka Receiver integration endpoint log includes the following entries:

When the Kafka broker is running but no messages are retrieved, the number of events, the Topic, and the Topic Offset are reported.
When the Kafka broker is not running, the IIEP fails, and a minimum number of entries are reported since no messages are being processed.