Apache Kafka connector (DataStage)

Prerequisite

Create the connection. For instructions, see Connecting to a data source in DataStage and the Apache Kafka connection.

• Reading data from the topics in Kafka
• Writing data to topics in Kafka
• Schema registry
• Kafka client configuration properties
• Troubleshooting

Reading data from the topics in Kafka

You can configure the Apache Kafka connector to read messages from a Kafka data source.

Configuring the Apache Kafka connector as a source

1. From the job design canvas, double-click the Apache Kafka connector.
2. On the Stage tab, click Properties.
3. Select Use DataStage properties.
4. Enter the values for the required properties:
   • Topic name - Kafka topic to read messages from.
   • Key serializer - The serializer for the key data types in Apache Kafka. The Key serializer value must be compatible with the key column (if defined) in the Output link. If you select either an Avro or an Avro as JSON key serializer, the Value serializer must be set to either Avro or Avro as JSON.
   • Value serializer - The serializer for the value data types in Apache Kafka. The Value serializer value must be compatible with the key column (if defined) in the Output link.
   • Consumer group - A string that uniquely identifies the group of consumer processes to which this consumer belongs. By setting the same group ID for multiple processes, you indicate that they are all part of the same consumer group.
   • Max poll records - The maximum number of records that are returned in a single call to poll.
   • Max messages - The maximum number of messages to consume or produce from or to the topic on a player-process basis. This value is a multiple of the Max poll records. This setting is not active if you select Continuous mode.
   • Reset policy - The value that indicates the policy if there is no initial offset in the Kafka server or if the current offset does not exist anymore on the server. For example, if the data is deleted, you can use the values earliest and latest to automatically reset the offset to the earliest of the latest offset.
5. Configure optional properties:
   • Continuous mode - The connector will fetch messages in a continuous mode by waiting for new incoming messages indefinitely.
     • Stop message - The message that stops the continuous-mode processing.
   • Transaction - The transaction that sends the "End of wave" marker settings:
     • Record count - The number of records per transaction. The value 0 means all available records.
     • Time interval - The time interval for a transaction. Use this property if you set the Record count to 0.
   • End of wave - Send an "End of wave" marker after each transaction.
   • End of data - Insert an "End of wave" marker for the final set of records when their number is less than the value specified for the transaction record count. If the transaction Record count value is 0 (all available records), there is only one transaction wave. In this case, set the End of data value to Yes so that the "End of wave" marker will be inserted for that transaction wave.
   • Logging level - Configures the Kafka client logging. See Controlling log output.
   • Client configuration properties - Sets additional Kafka client properties. See Kafka client configuration properties
6. Click the Output tab to define the columns for the connector. The columns must be compatible with the data type that is defined in the Key serializer and Value serializer properties. If you use an Avro or an Avro as JSON serializer, the column names must match the Avro record fields. Refer to Schema registry for details.
   • value - The column that contains the Kafka message values. It must be compatible with the Value serializer property.
   • key - The column that contains the Kafka message keys. It must be compatible with the Key serializer property.
   • partition - The column that contains the partition number that the Kafka message comes from. It must be compatible with Integer data type.
   • timestamp - The column that contains the timestamp for the Kafka message. It must be compatible with the Timestamp data type.
   • offset - The column that contains the offset for the Kafka message. It must be compatible with the BigInt data type.

Writing data to topics in Kafka

You can configure the Apache Kafka connector to write data to a Kafka data source.

Configuring the Apache Kafka connector as a target

1. From the job design canvas, double-click the Apache Kafka connector.
2. On the Stage tab, click Properties.
3. Select Use DataStage properties.
4. Enter the values for the required properties:
   • Topic name - The name of the topic into which the messages are to be written from the upstream stage.
   • Key serializer - The serializer for the key data types in Apache Kafka. The Key serializer value must be compatible with the key column (if defined) in the Input link. If you select either an Avro or an Avro as JSON key serializer, the Value serializer must be set to either Avro or to Avro as JSON.
   • Value serializer - The serializer for the value data types in Apache Kafka. The Value serializer value must be compatible with the key column (if defined) in the Input link.
5. Configure optional properties:
   • Logging level - Configures the Kafka client logging. See Controlling log output.
   • Client configuration properties - Sets additional Kafka client properties. See Kafka client configuration properties.
6. Click the Input tab to define the columns for the connector. The columns must be compatible with data type that is defined in the Key serializer and Value serializer properties. If you use an Avro or an Avro as JSON serializer, the column names must match the Avro record fields. Refer to Schema registry for details.
   • value - The column that contains the Kafka message values. It must be compatible with the Value serializer property.
   • key - The column that contains the Kafka message keys. It must be compatible with the Key serializer property.

Schema registry

The schema registry gives you the ability to use more complex Kafka messages. You can use the schema registry for general DataStage column definitions within job designs.

By default, messages are returned as a value within a single column that you define in the Apache Kafka connector. When you use the Avro and Avro as JSON formats and the schema registry, you enable decomposition of complex Kafka messages into the DataStage columns.

When you write to a topic, the connector creates a schema that is based on the column definitions.

When the connector reads from a Kafka topic, the connector takes a schema from the schema registry. Based on that schema, the Avro record fields are extracted into columns that are defined as connector output.

{
  "fields": [
    {
      "name": "id",
      "type": "int"
    },
    {
      "name": "name",
      "type": "string"
    },
    {
      "name": "salary",
      "type": "double"
    }
  ],
  "name": "KafkaConnectorSchema",
  "type": "record"
}

To use the schema registry, use the following configuration:

• You must set either the Key serializer or the Value serializer configuration property to Avro or to Avro as JSON.
• After you set the Key serializer to Avro or Avro as JSON, you cannot configure the Value serializer to a value other than Avro or Avro as JSON.
• When you set the Key serializer configuration property to Avro, the column definition for the link must conform to the Avro schema definition.
• Key columns must either be selected with the Key checkbox or their names must start with a key_ prefix. Use only one method. You cannot use a Key checkbox for one column and a key_ prefix for another column.

Kafka client configuration properties

Use the Client configuration properties (under Advanced Kafka configuration on the Stage tab) when you need to set a Kafka client configuration property, but the Apache Kafka connector does not have a configuration dialog for it.

Enter the property with the format key=value with entries defined on each line. Use the # character for comments. For a description of the rules for the client configuration properties, refer to the Java Properties definition.

Troubleshooting