Kafka topic

Configuration properties

Properties for a Kafka Messaging Topic are defined by name in the dlc.messaging.kafka.topics namespace, and are picked up as configurations for the DLC.

`KafkaTopicProperties`

The Kafka Message Topic defines the topic itself along with the configuration of the Message Consumer and the Transaction Controller Task.

Key	Required	Type	Default	Description
`name`	Y	`String`		SQL statement. Can accept parameters.
`messaging-broker-topic`	Y	`String`		The Kafka Topic the `KafkaConsumer` should be subscribed to. This will match the Producer’s Topic.
`columns`		`List<String>`	Columns will implicitly come from the store which is being published to.	The order of columns from the Kafka Producer. If a column name matches a datastore column, its parser will be implied. For columns that do not match a datastore column, you must provide a parser in the channel as a Custom Field. Columns without a matching datastore column or parser are skipped.
`target-stores`		`Set<String>`	The Topic’s `name`.	Stores to write tuples into. If null, then it is assumed the `name` is the target store.
`messaging-topic-config-description`		`MessagingTopicConfigProperties`	Default `MessagingTopicConfigProperties`	Performance configurations of this Topic.
`messaging-consumer-properties`	Y	`Map<String, String>`		The configuration Properties to be passed into the `KafkaConsumer`.
`tx-controller-task-description`		`TxControllerTaskConfigProperties`	Default `TxControllerTaskConfigProperties`	Configuration of the `TxControllerTask` that this Topic will use.
`channels`		`Set<Channel>`	A channel is created using the topic’s name as the name of the target.	Channels.
`restrict-to-sources`		`Set<String>`		Sources to which the topic is restricted. See source to topic matching.
`restrict-from-sources`		`Set<String>`		Sources from which the topic is restricted. See source to topic matching.

`MessagingTopicConfigProperties`

Defines some performance configuration properties for the message consumer.

Key	Required	Type	Default	Description
`num-consumers`		`Integer`	`1`	Number of `KafkaConsumer`s to open.
`publish-queue-capacity`		`Integer`	`100,000,000`	The maximum size of the `BlockingQueue` shared between the `KafkaMessageConsumer` and `TxSizeController`.

`TxControllerTaskConfigProperties`

Defines some performance configuration properties for the message consumer.

Key	Type	Default	Description
`num-publishers`	`Integer`	`4`	Number of parallel `TxSizeController.PublishTask` to run. Each will publish tuples into the `IMessageChannel`.
`publish-chunk-size`	`Integer`	`100,000`	Maximum number of records to publish in a single `TxSizeController.PublishTask`.
`max-tx-drain-size`	`Integer`	`400,000`	Threshold to flush and commit a Datastore Transaction. Counter is reset on each Transaction.
`max-tx-drain-time-ms`	`Integer`	`3000`	Threshold to flush and commit a Datastore Transaction. Clock is reset on each Transaction.

`Map<String, String> messagingConsumerProperties`

Defines properties required for the KafkaConsumer. These properties are passed directly to the KafkaConsumer. The individual properties required may vary depending on the Kafka Producer in use. Below are a few that may be required:

Key	Required	Type	Description
`bootstrap-servers`	Y	`String`	URL of your Kafka Producer.
`key-deserializer`	Y	`String`	Classpath of the Deserializer to use to parse the keys of the messages from the Kafka Producer.
`value-deserializer`	Y	`String`	Classpath of the Deserializer to use to parse the values of the messages from the Kafka Producer.
`group-id`	Y	`String`	Group Identifier of the Kafka Producer.
`auto-offset-reset`	Y	`String`	Kafka Consumer offset handling behavior.
`offsets`		`Map<String, String>`	Offset Key-Value pairs.

YAML example

dlc:
  messaging:
    kafka:
      topics:
        MyTradeTopic:
          messaging-broker-topic: Kafka-Source-Trades
          source-fields:
            - TradeId
            - AsOfDate
          target-stores:
            - TradesBaseStore
          tx-controller-task-description:
            num-publishers: 4
            publish-chunk-size: 100000
            max-tx-drain-size: 400000
            max-tx-drain-time-ms: 3000
          messaging-topic-config-description:
            num-consumers: 1
            publish-queue-capacity: 100000000
          messaging-consumer-properties:
            bootstrap:
              servers: host:port
            key:
              deserializer: com.activeviam.io.dlc.impl.description.source.messaging.StringTupleDeserializer
            value:
              deserializer: com.activeviam.io.dlc.impl.description.source.messaging.StringTupleDeserializer
            group:
              id: my-consumer-group-1
            auto:
              offset:
                reset: earliest

Java configuration

KafkaTopicDescription Spring Beans are picked up as configurations for the DLC.

`KafkaTopicDescription`

The Kafka Message Topic defines the topic itself along with the configuration of the Message Consumer and the Transaction Controller Task.

Key	Required	Type	Default	Description
`name`	Y	`String`		SQL statement. Can accept parameters.
`channels`		`Set<Channel>`	A channel is created using the topic’s name as the name of the target.	Channels.
`restrictToSources`		`Set<String>`		Sources to which the topic is restricted. See source to topic matching.
`restrictFromSources`		`Set<String>`		Sources from which the topic is restricted. See source to topic matching.
`messagingBrokerTopic`	Y	`String`		The Kafka Topic the `KafkaConsumer` should be subscribed to. This will match the Producer’s Topic.
`columns`		`List<String>`	Columns will implicitly come from the store which is being published to.	The order of columns from the Kafka Producer. If a column name matches a datastore column, its parser will be implied. For columns that do not match a datastore column, you must provide a parser in the channel as a Custom Field. Columns without a matching datastore column or parser are skipped.
`targetStores`		`Set<String>`	The Topic’s `name`.	Stores to write tuples into. If null, then it is assumed the `name` is the target store.
`messagingTopicConfigDescription`		`KafkaTopicConfigDescription`	Default `MessagingTopicConfigProperties`	Performance configurations of this Topic.
`txControllerTaskDescription`		`TxControllerTaskDescription`	Default `TxControllerTaskConfigProperties`	Configuration of the `TxControllerTask` that this Topic will use.
`messagingConsumerProperties`	Y	`Properties`		The configuration Properties to be passed into the `KafkaConsumer`.

`KafkaTopicConfigDescription`

Defines some performance configuration properties for the message consumer.

Key	Required	Type	Default	Description
`numConsumers`		`Integer`	`1`	Number of `KafkaConsumer`s to open.
`publishQueueCapacity`		`Integer`	`100,000,000`	The maximum size of the `BlockingQueue` shared between the `KafkaMessageConsumer` and `TxSizeController`.

`TxControllerTaskDescription`

Defines some performance configuration properties for the message consumer.

Key	Type	Default	Description
`numPublishers`	`Integer`	`4`	Number of parallel `TxSizeController.PublishTask` to run. Each will publish tuples into the `IMessageChannel`.
`publishChunkSize`	`Integer`	`100,000`	Maximum number of records to publish in a single `TxSizeController.PublishTask`.
`maxTxDrainSize`	`Integer`	`400,000`	Threshold to flush and commit a Datastore Transaction. Counter is reset on each Transaction.
`maxTxDrainTimeMs`	`Integer`	`3000`	Threshold to flush and commit a Datastore Transaction. Clock is reset on each Transaction.

`Properties messagingConsumerProperties`

Key	Required	Type	Description
`bootstrap-servers`	Y	`String`	URL of your Kafka Producer.
`key-deserializer`	Y	`String`	Classpath of the Deserializer to use to parse the keys of the messages from the Kafka Producer.
`value-deserializer`	Y	`String`	Classpath of the Deserializer to use to parse the values of the messages from the Kafka Producer.
`group-id`	Y	`String`	Group Identifier of the Kafka Producer.
`auto-offset-reset`	Y	`String`	Kafka Consumer offset handling behavior.
`offsets`		`Map<String, String>`	Offset Key-Value pairs.

Java example

		@Bean
KafkaTopicDescription kafkaTopicDescription() {
	return KafkaTopicDescription.builder(SIMPLE_BASE_STORE_NAME)
			.messagingBrokerTopic(SIMPLE_BASE_STORE_NAME)
			.columns(List.of("idKey", "price"))
			.targetStore(SIMPLE_BASE_STORE_NAME)
			.messagingTopicConfigDescription(messagingTopicConfigDescription())
			.messagingConsumerProperties(messagingConsumerProperties())
			.txControllerTaskDescription(transactionControllerTaskDescription())
			.build();
}

@Bean
KafkaTopicConfigDescription messagingTopicConfigDescription(){
	return KafkaTopicConfigDescription.builder()
			.numConsumers(1)
			.publishQueueCapacity(100_000_000)
			.build();
}

@Bean
TxControllerTaskDescription transactionControllerTaskDescription(){
	return TxControllerTaskDescription.builder()
			.numPublishers(4)
			.publishChunkSize(100_000)
			.maxTxDrainSize(400_000)
			.maxTxDrainTimeMs(3000)
			.build();
}

@Bean
Properties messagingConsumerProperties(){
	var messagingConsumerProperties = new Properties();
	messagingConsumerProperties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "host:PORT");
	messagingConsumerProperties.put(PROP_POLL_TIMEOUT_MS, "50");
	messagingConsumerProperties.put("key.deserializer", StringTupleDeserializer.class.getName());
	messagingConsumerProperties.put("value.deserializer", StringTupleDeserializer.class.getName());
	messagingConsumerProperties.put("group.id", "test-consumer-group");
	messagingConsumerProperties.put("auto.offset.reset", "earliest");
	return messagingConsumerProperties;
}

Operations

Request API

Configuration

Components

Functional

Extensions

Configuration properties

`KafkaTopicProperties`

`MessagingTopicConfigProperties`

`TxControllerTaskConfigProperties`

`Map<String, String> messagingConsumerProperties`

YAML example

Java configuration

`KafkaTopicDescription`

`KafkaTopicConfigDescription`

`TxControllerTaskDescription`

`Properties messagingConsumerProperties`

Java example

​Configuration properties

​KafkaTopicProperties

​MessagingTopicConfigProperties

​TxControllerTaskConfigProperties

​Map<String, String> messagingConsumerProperties

​YAML example

​Java configuration

​KafkaTopicDescription

​KafkaTopicConfigDescription

​TxControllerTaskDescription

​Properties messagingConsumerProperties

​Java example

Configuration properties

`KafkaTopicProperties`

`MessagingTopicConfigProperties`

`TxControllerTaskConfigProperties`

`Map<String, String> messagingConsumerProperties`

YAML example

Java configuration

`KafkaTopicDescription`

`KafkaTopicConfigDescription`

`TxControllerTaskDescription`

`Properties messagingConsumerProperties`

Java example