Skip to main content

​
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.
KeyRequiredTypeDefaultDescription
nameYStringSQL statement. Can accept parameters.
messaging-broker-topicYStringThe Kafka Topic the KafkaConsumer should be subscribed to. This will match the Producer’s Topic.
columnsList<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-storesSet<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-descriptionMessagingTopicConfigPropertiesDefault MessagingTopicConfigPropertiesPerformance configurations of this Topic.
messaging-consumer-propertiesYMap<String, String>The configuration Properties to be passed into the KafkaConsumer.
tx-controller-task-descriptionTxControllerTaskConfigPropertiesDefault TxControllerTaskConfigPropertiesConfiguration of the TxControllerTask that this Topic will use.
channelsSet<Channel>A channel is created using the topic’s name as the name of the target.Channels.
restrict-to-sourcesSet<String>Sources to which the topic is restricted. See source to topic matching.
restrict-from-sourcesSet<String>Sources from which the topic is restricted. See source to topic matching.

​
MessagingTopicConfigProperties

Defines some performance configuration properties for the message consumer.
KeyRequiredTypeDefaultDescription
num-consumersInteger1Number of KafkaConsumers to open.
publish-queue-capacityInteger100,000,000The maximum size of the BlockingQueue shared between the KafkaMessageConsumer and TxSizeController.

​
TxControllerTaskConfigProperties

Defines some performance configuration properties for the message consumer.
KeyRequiredTypeDefaultDescription
num-publishersInteger4Number of parallel TxSizeController.PublishTask to run. Each will publish tuples into the IMessageChannel.
publish-chunk-sizeInteger100,000Maximum number of records to publish in a single TxSizeController.PublishTask.
max-tx-drain-sizeInteger400,000Threshold to flush and commit a Datastore Transaction. Counter is reset on each Transaction.
max-tx-drain-time-msInteger3000Threshold 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:
KeyRequiredTypeDefaultDescription
bootstrap-serversYStringURL of your Kafka Producer.
key-deserializerYStringClasspath of the Deserializer to use to parse the keys of the messages from the Kafka Producer.
value-deserializerYStringClasspath of the Deserializer to use to parse the values of the messages from the Kafka Producer.
group-idYStringGroup Identifier of the Kafka Producer.
auto-offset-resetYStringKafka Consumer offset handling behavior.
offsetsMap<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.
KeyRequiredTypeDefaultDescription
nameYStringSQL statement. Can accept parameters.
channelsSet<Channel>A channel is created using the topic’s name as the name of the target.Channels.
restrictToSourcesSet<String>Sources to which the topic is restricted. See source to topic matching.
restrictFromSourcesSet<String>Sources from which the topic is restricted. See source to topic matching.
messagingBrokerTopicYStringThe Kafka Topic the KafkaConsumer should be subscribed to. This will match the Producer’s Topic.
columnsList<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.
targetStoresSet<String>The Topic’s name.Stores to write tuples into. If null, then it is assumed the name is the target store.
messagingTopicConfigDescriptionKafkaTopicConfigDescriptionDefault MessagingTopicConfigPropertiesPerformance configurations of this Topic.
txControllerTaskDescriptionTxControllerTaskDescriptionDefault TxControllerTaskConfigPropertiesConfiguration of the TxControllerTask that this Topic will use.
messagingConsumerPropertiesYPropertiesThe configuration Properties to be passed into the KafkaConsumer.

​
KafkaTopicConfigDescription

Defines some performance configuration properties for the message consumer.
KeyRequiredTypeDefaultDescription
numConsumersInteger1Number of KafkaConsumers to open.
publishQueueCapacityInteger100,000,000The maximum size of the BlockingQueue shared between the KafkaMessageConsumer and TxSizeController.

​
TxControllerTaskDescription

Defines some performance configuration properties for the message consumer.
KeyRequiredTypeDefaultDescription
numPublishersInteger4Number of parallel TxSizeController.PublishTask to run. Each will publish tuples into the IMessageChannel.
publishChunkSizeInteger100,000Maximum number of records to publish in a single TxSizeController.PublishTask.
maxTxDrainSizeInteger400,000Threshold to flush and commit a Datastore Transaction. Counter is reset on each Transaction.
maxTxDrainTimeMsInteger3000Threshold to flush and commit a Datastore Transaction. Clock is reset on each Transaction.

​
Properties 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:
KeyRequiredTypeDefaultDescription
bootstrap-serversYStringURL of your Kafka Producer.
key-deserializerYStringClasspath of the Deserializer to use to parse the keys of the messages from the Kafka Producer.
value-deserializerYStringClasspath of the Deserializer to use to parse the values of the messages from the Kafka Producer.
group-idYStringGroup Identifier of the Kafka Producer.
auto-offset-resetYStringKafka Consumer offset handling behavior.
offsetsMap<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;
}