enhancement
The current implementation in TLKafkaProducer provides only a fraction of the possible producer configurations as typed configurations. In particular, settings for encryption, etc., are missing. However, since TLKafkaProducer was designed to be non-extensible (the method that actually instantiates the KafkaProducer hard-codes only very specific parameters from the configuration and cannot be overridden), this cannot simply be added later.
Requirement
The implementation should be modified so that—similar to the KafkaConsumerService —the configuration can be extended. Either the private API (getProperties() and newKafkaProducer()) would need to be exposed, or the configuration should be made extensible (similar to the ConsumerDispatcherConfiguration using annotations).
Implementation
- The missing Kafka properties have been added.
- Any properties can be configured using CommonClientConfig.getUntypedProperties(). These are treated by Top-Logic as a Map<String, String> and, in case of conflicts, override the explicitly defined properties. This allows Kafka to use these properties when the property type changes, thereby bypassing the now-obsolete typing. These properties are not parsed, and no validation is performed to check whether such a property exists in Kafka or whether the value is valid. Example:
{{{#!xml <producer class="com.top_logic.kafka.services.producer.TLKafkaProducer"
name="KBChange-Producer"
topic="%DATA_CHANGE_TOPIC_PRODUCER%"
>
<untyped-properties>
<property name="bootstrap.servers" value="%KAFKA_PRODUCER_SERVER%"/>
<property name="my.special.property" value="3.14.15"/>
</untyped-properties>
</producer> }}}
- Several relevant methods have been marked as protected and documented so that subclasses can override them.
Code Migration
The names of the following Kafka consumer/producer properties have been standardized. The name in Kafka itself is now always the authoritative one, so you no longer need to look up what a Kafka property is called in Top-Logic.
- acknowledgeLevel => acks
- compressionType => compression.type
- maxRequestSize => max.request.size
- request-timeout -> request.timeout.ms
- key.deserializer => key.deserializer.typed.config
- key-deserializer-class => key.deserializer
- key-serializer-class => key.serializer
- keySerializer => key.serializer.typed.config
- value-deserializer-class => value.deserializer
- value.deserializer => value.deserializer.typed.config
- value-serializer-class => value.serializer
- valueSerializer => value.serializer.typed.config
The names of some methods in CommonClientConfig and its subclasses have changed. However, these should not have been used in projects. Therefore, this is only relevant in theory.
ConsumerDispatcher.getProperties(...) has been replaced by getAllProperties().
Test
There is no specific test for the improved configurability. For Kafka in general, the tests are located in test.com.top_logic.kafka and test.com.top_logic.kafka.demo.