Configuring Kafka endpoints
- 1 Configuration properties
- 1.1 Generic Kafka parameters
- 1.1.1 kafka.[kafka_properties]
- 1.2 Generic Kafka Producer parameters
- 1.3 Kafka Producer parameters
- 1.3.1 kafka.producer.[ClientName].client.id
- 1.3.2 kafka.producer.[ClientName].topic
- 1.3.3 kafka.producer.[ClientName].async.sending
- 1.3.4 kafka.producer.[ClientName].key.serializer
- 1.3.5 kafka.producer.[ClientName].value.serializer
- 1.3.6 kafka.producer.[ClientName].groups
- 1.3.7 kafka.producer.[ClientName].startOnload
- 1.3.8 kafka.producer.[ClientName].startTime
- 1.3.9 kafka.producer.[ClientName].stopTime
- 1.3.10 kafka.producer.[ClientName].scheduleTimeZone
- 1.3.11 kafka.producer.[ClientName].[kafka_properties]
- 1.4 Generic Kafka Consumer parameters
- 1.5 Kafka Consumer parameters
- 1.5.1 kafka.consumer.[ClientName].client.id
- 1.5.2 kafka.consumer.[ClientName].topics
- 1.5.3 kafka.consumer.[ClientName].topics.regexp
- 1.5.4 kafka.consumer.[ClientName].
- 1.5.5 kafka.consumer.[ClientName].key.deserializer
- 1.5.6 kafka.consumer.[ClientName].value.deserializer
- 1.5.7 kafka.consumer.[ClientName].groups
- 1.5.8 kafka.consumer.[ClientName].startOnload
- 1.5.9 kafka.consumer.[ClientName].startTime
- 1.5.10 kafka.consumer.[ClientName].stopTime
- 1.5.11 kafka.consumer.[ClientName].scheduleTimeZone
- 1.5.12 kafka.consumer.[ClientName].[kafka_properties]
- 1.1 Generic Kafka parameters
- 2 Sample configuration
The Kafka endpoint is a FEJ transport adapter, which provides connectivity to the Kafka streaming platform. Its implementation is based on the FIX Antenna Java Kafka adapter and it uses Kafka Producer and Consumer API internally.
Kafka endpoint parameters are described in the conf/kafka-adaptor.properties file. The configuration should include a list of Kafka endpoints and link them to the corresponding Kafka topics.
Configuration properties
The Kafka endpoints are configured by means of the following properties:
Property Name | Description | Required | Default Value |
|---|---|---|---|
kafka.clients | Comma-delimited list of Kafka endpoints. | Y | |
Generic Kafka parametersThese parameters are applied to all producer and consumer endpoints | |||
kafka.[kafka_properties] | This section may include multiple options, which are applied to all producer and consumer endpoints. As a [kafka_property] may be used any key from Kafka Producer Configs or Kafka Consumer Config. For example, it's a good section to define a bootstrap server for all endpoints:
| N | |
Generic Kafka Producer parameters | |||
kafka.producer.key.serializer | The serializer class for a record key that implements the | N | |
kafka.producer.value.serializer | The serializer class for a record value that implements the | N | |
kafka.producer.[kafka_properties] | This section may include multiple options, which are applied to all producer endpoints. As a [kafka_property] may be used any key from Kafka Producer Configs. | N | |
Kafka Producer parameters | |||
kafka.producer.[ClientName].client.id | The ID string to pass to the server when making requests. The purpose of this is to be able to track the source of requests beyond just IP/port by allowing a logical application name to be included in server-side request logging. | Y | |
kafka.producer.[ClientName].topic | The topic for publishing by the given Kafka producer endpoint. | Y | |
kafka.producer.[ClientName].async.sending | Defines the sending mode for this producer endpoint. If the value is true, the endpoint sends data asynchronously. If the value is false, it waits for confirmation from the Kafka broker for every sent message. | N | false |
kafka.producer.[ClientName].key.serializer | The serializer class for a record key that implements the | Y, if the | |
kafka.producer.[ClientName].value.serializer | The serializer class for a record value that implements the | Y, if the | |
kafka.producer.[ClientName].groups | The list of groups for routing | N | |
kafka.producer.[ClientName].startOnload | Specifies whether a producer should be started during the FIXEdge Java server initialization. | N | true |
kafka.producer.[ClientName].startTime | A cron expression that defines a producer's start time. | N | |
kafka.producer.[ClientName].stopTime | A cron expression that defines a producer's stop time. | N | |
kafka.producer.[ClientName].scheduleTimeZone | A time zone for the start and stop times | N | |
kafka.producer.[ClientName].[kafka_properties] | This section may include multiple options, which are applied to all producer endpoints. As a [kafka_property] may be used any key from Kafka Producer Configs. | N | |
Generic Kafka Consumer parameters | |||
kafka.consumer.key.deserializer | The deserializer class for a record key that implements the | N | |
kafka.consumer.value.deserializer | The deserializer class for a record value that implements the | N | |
kafka.consumer.[kafka_properties] | This section may include multiple options, which are applied to all producer endpoints. As a [kafka_property] may be used any key from Kafka Consumer Configs. | N | |
Kafka Consumer parameters | |||
kafka.consumer.[ClientName].client.id | The ID string to pass to the server when making requests. The purpose of this is to be able to track the source of requests beyond just IP/port by allowing a logical application name to be included in server-side request logging. | Y | |
kafka.consumer.[ClientName].topics | The list of topics to subscribe to | Y, if the | |
kafka.consumer.[ClientName].topics.regexp | The pattern to subscribe to | Y, if the | |
kafka.consumer.[ClientName]. | The time, in milliseconds, which consumer spends waiting in the polling procedure if data is not available in the buffer. If 0, returns immediately with any records that are available currently in the buffer, else returns empty. Must not be negative. | N | 200 |
kafka.consumer.[ClientName].key.deserializer | The deserializer class for a record key that implements the | Y, if the | |
kafka.consumer.[ClientName].value.deserializer | The deserializer class for a record value that implements the | Y, if the | |
kafka.consumer.[ClientName].groups | The list of groups for routing | N | |
kafka.consumer.[ClientName].startOnload | Specifies whether a consumer should be started during FIXEdge Java server initialization. | N | true |
kafka.consumer.[ClientName].startTime | A cron expression that defines a consumer's start time. | N | |
kafka.consumer.[ClientName].stopTime | A cron expression that defines a consumer's stop time. | N | |
kafka.consumer.[ClientName].scheduleTimeZone | A time zone for the start and stop times | N | |
kafka.consumer.[ClientName].[kafka_properties] | This section may include multiple options, which are applied to all producer endpoints. As a [kafka_property] may be used any key from Kafka Consumer Configs. | N | |
For the full list of Kafka properties refer to the Kafka official documentation.
NOTE: All defined Kafka endpoints are started automatically during FIXEdge Java server initialization by default if there is a defined schedule for them (starting from FEJ 1.8.0). The schedule may be defined by startTime/stopTime properties or within a separate schedule configuration. Refer to the Scheduler section for more details.
NOTE: All changes in the properties file are applied only after the FIXEdge Java server restart.
Sample configuration
The sample of the simplest config with Kafka producer and consumer endpoints:
# list of kafka endpoints (consumers and producer)
kafka.clients = KProducer, KConsumer
# main properties to specify producer's id and topic
kafka.producer.KProducer.client.id = KProducer
kafka.producer.KProducer.topic = PTopic1
# Type is class. Serializer class for key that implements the com.epam.fej.kafka.KafkaEndpointKeySerializer interface.
kafka.producer.key.serializer = com.epam.fej.kafka.FIXMessageEventSerializer
# Type is class. Serializer class for value that implements the com.epam.fej.kafka.KafkaEndpointValueSerializer interface.
kafka.producer.value.serializer = com.epam.fej.kafka.FIXMessageEventSerializer
# main properties to specify consumer's id and topic
kafka.consumer.KConsumer.client.id = KConsumer
kafka.consumer.KConsumer.topics = KTopic2
# Type is class. Deserializer class for key that implements the com.epam.fej.kafka.KafkaEndpointKeyDeserializer interface.
kafka.consumer.key.deserializer = org.apache.kafka.common.serialization.StringDeserializer
# Type is class. Deserializer class for value that implements the com.epam.fej.kafka.KafkaEndpointValueDeserializer interface.
kafka.consumer.value.deserializer = com.epam.fej.kafka.FIXMessageEventDeserializer