How to get the name of the topic outside the kafkalistener?

[hariraogotit]

Team,

I have a retry bean

@Bean
  public RetryTopicConfiguration rtKafkaConfig(
          KafkaTemplate<String, byte[]> rtKafkaTemplate) {
    return RetryTopicConfigurationBuilder
            .newInstance()
            .exponentialBackoff(initialInterval, multiplier, maxInterval)
            .maxAttempts(maxAttempts)
            .retryTopicSuffix(retryTopicSuffix)
            .dltSuffix(dltSuffix)
            .dltHandlerMethod("dltListener", "dlt")
            .useSingleTopicForSameIntervals()
            .create(rtKafkaTemplate);
  }

I need to log all the topic names in the listeners, in order to do this I added extra parameter in the listener as below

public void newHandler(
          ConsumerRecord<String, byte[]> record,
          @Header(KafkaHeaders.RECEIVED_TOPIC) String topicName)

Since I have lots of Kafka listeners it is time consuming to add the extra parameter.

Is there any other place where I can add @Header(KafkaHeaders.RECEIVED_TOPIC) String topicName and it will be applied to all my listeners? (i.e is there any other place outside kafkalistener where I can access the name of the topic and it will be applied to all my listeners just like the above retry bean is applied across all the listeners)

Kindly let me know. Any leads are much appreciated. Thanks

[artembilan]

You don't need that header if you deal with ConsumerRecord. This one has a dedicated property you can always consult:

    /**
     * Creates a record to be received from a specified topic and partition (provided for
     * compatibility with Kafka 0.9 before the message format supported timestamps and before
     * serialized metadata were exposed).
     *
     * @param topic The topic this record is received from
     * @param partition The partition of the topic this record is received from
     * @param offset The offset of this record in the corresponding Kafka partition
     * @param key The key of the record, if one exists (null is allowed)
     * @param value The record contents
     */
    public ConsumerRecord(String topic,
                          int partition,
                          long offset,
                          K key,
                          V value) {

On the other hand it is not clear how you are going to build those listener methods logic if you are not going to propagate the value via method argument.

When you talk about retry, you don't compare apples with apples: the retry (as well as listener instance) is created only once during configuration phase.
A received record is volatile object and its context can be extracted only if we say so via listener method signature.

[hariraogotit]

@artembilan Thanks a lot for the response.

I now understand the volatile object vs initialised in the beginning.

I was thinking Spring might have some kind of interceptor which when implemented will capture the topic name for eg I am planning to use aspects to log the topic name from ConsumerRecord like you said.

[artembilan]

Yes, you can add a RecordInterceptor to catch the listener call.
See AbstractMessageListenerContainer:

		/**
	 * Set an interceptor to be called before calling the record listener.
	 * Does not apply to batch listeners.
	 * @param recordInterceptor the interceptor.
	 * @since 2.2.7
	 * @see #setInterceptBeforeTx(boolean)
	 */
	public void setRecordInterceptor(RecordInterceptor<K, V> recordInterceptor) {

So, probably in the impl you can set some ThreadLocal value and get access to it downstream in your business logic if the listener method.

[hariraogotit]

ah! This is what I was looking for.

I see RecordInterceptor gives ConsumerRecord and KafkaConsumer. I am going to use the first parameter i.e ConsumerRecord as it has everything I want.

Once again thanks a lot @artembilan

[artembilan]

Good. So, in the success() you need to clean up that ThreadLocal value. If you are going to follow that approach.

[hariraogotit]

@artembilan So here is what I am going to do

I want to log saying X message is received from Y topic. I do not want to do this in each and every listener.

So in order to achieve this I will use setRecordInterceptor in ConcurrentKafkaListenerContainerFactory. I will pass a method in setRecordInterceptor which will log these.

So I believe I do not have to use the ThreadLocal, success etc. Correct me if this is not the case.