> ## Documentation Index > Fetch the complete documentation index at: https://docs.activeviam.com/llms.txt > Use this file to discover all available pages before exploring further. # Kafka topic > Configuration reference for `KafkaTopicDescription`, covering the broker topic, consumer properties, target stores, and performance tuning via `MessagingTopicConfigProperties` and `TxControllerTaskConfigProperties`. ### 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` | 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](/configuration/custom-field). Columns without a matching datastore column or parser are skipped. | | `target-stores` | | `Set` | 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` | | 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` | A channel is created using the topic's name as the name of the target. | [Channels](/configuration/channel). | | `restrict-to-sources` | | `Set` | | Sources *to* which the topic is restricted. See [source to topic matching](/functional/source-to-topic-matching). | | `restrict-from-sources` | | `Set` | | Sources *from* which the topic is restricted. See [source to topic matching](/functional/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 | Required | 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 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 | Default | 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` | | Offset Key-Value pairs. | #### YAML example ```yaml theme={"languages":{"custom":["/engine/python-sdk/0.9/languages/pycon.tmLanguage.json"]}} 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` | A channel is created using the topic's name as the name of the target. | [Channels](/configuration/channel). | | `restrictToSources` | | `Set` | | Sources *to* which the topic is restricted. See [source to topic matching](/functional/source-to-topic-matching). | | `restrictFromSources` | | `Set` | | Sources *from* which the topic is restricted. See [source to topic matching](/functional/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` | 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](/configuration/custom-field). Columns without a matching datastore column or parser are skipped. | | `targetStores` | | `Set` | 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 | Required | 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` 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 | Default | 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` | | Offset Key-Value pairs. | #### Java example ```java theme={"languages":{"custom":["/engine/python-sdk/0.9/languages/pycon.tmLanguage.json"]}} @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; } ```