consumer group specific offset seeking for AbstractConsumerSeekAware

[akemalsaglam]

Expected Behavior

We want to be able to seek offset for specific consumer group by using AbstractConsumerSeekAware.

Current Behavior

regarding to below implementation it is clear that we can seek offset for all assigned partitions in a topic regardless of different consumer group ids.

    @Override
    public void onPartitionsAssigned(Map<TopicPartition, Long> assignments, ConsumerSeekCallback callback) {
        super.onPartitionsAssigned(assignments, callback);
    }

    @Override
    public void seekToTimestamp(long time) {
        getSeekCallbacks().forEach((tp, callback) -> {
            callback.seekToTimestamp(tp.topic(), tp.partition(), time);
        });
    }

Context
For our use case there might be more than one group instance which is assigned same partition in a topic. Below example might be useful to describe our case:

1. we have one topic let's name it: product.feed.fullexport, it has 12 partitions.
2. 10 different micro-services with different group ids are listening same topic and each has 12 concurrent consumers.
3. when we want to seek offset by using above ConsumerSeekCallback implementation for one of the micro-service, it affects all assigned partitions and listening consumer instances regardless to expected group id.

Is there any way to seek offset in a partition but only for specific group id?

[artembilan]

I don't see a behavior you are describing:

@SpringBootApplication
public class KafkaGh2302Application {

	public static void main(String[] args) {
		SpringApplication.run(KafkaGh2302Application.class, args);
	}


	@Bean
	public NewTopic topic() {
		return new NewTopic("seekExample", 3, (short) 1);
	}

	@Component
	public static class Listener extends AbstractConsumerSeekAware {

		@KafkaListener(id = "seekExample", topics = "seekExample", concurrency = "3")
		public void listen(String payload) {
			System.out.println("Listener received: " + payload);
		}

		public void seekToStart() {
			getSeekCallbacks().forEach((tp, callback) -> callback.seekToBeginning(tp.topic(), tp.partition()));
		}

	}

	@Component
	public static class Listener2 extends AbstractConsumerSeekAware {

		@KafkaListener(id = "seekExample2", topics = "seekExample", concurrency = "3")
		public void listen(String payload) {
			System.out.println("Listener2 received: " + payload);
		}

		public void seekToStart() {
			getSeekCallbacks().forEach((tp, callback) -> callback.seekToBeginning(tp.topic(), tp.partition()));
		}

	}

}

As you see I have two @KafkaListener classes with different ids which is, essentially, pointing to different consumer groups.

In the unit test I do:

@SpringBootTest
@EmbeddedKafka(bootstrapServersProperty = "spring.kafka.bootstrap-servers")
@DirtiesContext
class KafkaGh2302ApplicationTests {

	@Autowired
	KafkaGh2302Application.Listener listener;

	@Autowired
	KafkaTemplate<String, String> template;

	@Test
	void contextLoads() throws InterruptedException {

		for (int i = 0; i < 10; i++) {
			this.template.send("seekExample", i % 3, "some_key", "test#" + i);
		}

		Thread.sleep(1000);

		this.listener.seekToStart();

		Thread.sleep(10000);
	}

}

So, after sending some data, I just call seekToStart() on one of the listener services.
The output is like this:

Listener2 received: test#2
Listener received: test#1
Listener2 received: test#0
Listener2 received: test#5
Listener received: test#2
Listener received: test#4
Listener received: test#5
Listener received: test#7
Listener2 received: test#1
Listener2 received: test#8
Listener received: test#0
Listener2 received: test#3
Listener2 received: test#4
Listener received: test#8
Listener2 received: test#6
Listener received: test#3
Listener2 received: test#9
Listener2 received: test#7
Listener received: test#6
Listener received: test#9
Listener received: test#0
Listener received: test#3
Listener received: test#6
Listener received: test#9
Listener received: test#2
Listener received: test#5
Listener received: test#1
Listener received: test#4
Listener received: test#7
Listener received: test#8

This confirms that seeking really happens only in one consumer group and it does not effect other groups on the same topic.

[garyrussell]

Correct; seeks only affect the current group.

[bky373]

@artembilan @garyrussell
Hello, I've found that the current behavior can occur when using multiple listeners with different group IDs in the same class.

Current Behavior - described above

it is clear that we can seek offset for all assigned partitions in a topic regardless of different consumer group ids.

Test

I have modified the code provided above slightly.

• KafkaGh2302Application

...
  // Existing code
  @KafkaListener(id = "seekExample", topics = "seekExample", concurrency = "3")
  public void listen(String payload) {
      ...
  }

  // Added code in the same class Listener
  @KafkaListener(id = "seekExample3", topics = "seekExample", concurrency = "3")
  public void listen3(String payload) {
      System.out.println("Listener3 received: " + payload);
  }
...

• KafkaGh2302ApplicationTests

...

  // Existing code
  for (int i = 0; i < 30; i++) { // Change: 10 -> 30 for both two listeners(seekExample and seekExample3) to be able to seek offsets
    this.template.send("seekExample", i % 3, "some_key", "test#" + i);
  }

...

• Result
  • Since callbacks are registered per thread in a single listener class that implements AbstractConsumerSeekAware, seeking offsets is performed regardless of the consumer group ID. (In this case, Listener(id: seekExample) and Listener 3(id: seekExample3))

...
Listener received: test#0
Listener3 received: test#3
Listener2 received: test#16
Listener2 received: test#19
Listener2 received: test#22
Listener received: test#29
Listener2 received: test#25
Listener2 received: test#28
Listener received: test#28
Listener received: test#21
Listener received: test#24
Listener received: test#27
Listener2 received: test#23
Listener2 received: test#26
Listener3 received: test#16
========
Listener received: test#0
Listener received: test#3
Listener received: test#24
Listener received: test#27
Listener received: test#2
Listener3 received: test#1
Listener received: test#5
Listener3 received: test#4
Listener received: test#8
Listener3 received: test#7
Listener received: test#11
Listener3 received: test#10
...

IMHO, If we want to seek offset for a specific consumer group only, we can use the following methods:

• Use separate classes for consumers with different group IDs.
• Add filtering conditions based on the consumer group ID included in the thread name to ensure callbacks are registered or executed selectively (in registerSeekCallback() or onPartitionsAssigned()).
  • However, I don't want to recommend this as thread names can be changed arbitrarily and may not always include the consumer group ID.

If you have any other solutions for seeking offsets based on a specific consumer group ID, please let me know. I would appreciate hearing them. Thank you!

[garyrussell]

I am no longer involved with the project, but what you are suggesting is incorrect.

Each listener method is invoked by a different listener container and, therefore, on different threads.

So, if there is a problem, it is not related to any thread-based state.

[sobychacko]

@bky373 Thanks for reporting. As @garyrussell pointed out, this looks like it is non-thread-state related, but it looks like some bug (Thanks, Gary, for chiming in!! :) ). We will look at this today. Do you have any sample application for us to reproduce? (that would be easier). Otherwise, we can look into creating one since you provided some snippets.

[bky373]

@garyrussell

I am no longer involved with the project

Oh I didn't know that! Thanks for your comment!! 🙇

Each listener method is invoked by a different listener container and, therefore, on different threads.

You are totally right. Threads are different. I didn't mean to say that it's an issue with thread state.
If my suggestion seemed like it was due to a thread state issue, I'm afraid I wrote it wrong.

I just wanted to report that in a class that have listeners with different consumer group IDs and implements AbstractConsumerSeekAware, it's difficult to find the offset by specifying the consumer group ID. (This might not be the problem).

[bky373]

@bky373 Thanks for reporting. As @garyrussell pointed out, this looks like it is non-thread-state related, but it looks like some bug (Thanks, Gary, for chiming in!! :) ). We will look at this today. Do you have any sample application for us to reproduce? (that would be easier). Otherwise, we can look into creating one since you provided some snippets.

@sobychacko

Sure! the code is so simple so I'll leave it in the comments here.
Thank you for your time and reply!!

@SpringBootApplication
public class KafkaGh2302Application {

    public static void main(String[] args) {
        SpringApplication.run(KafkaGh2302Application.class, args);
    }

    @Bean
    public NewTopic topic() {
        return new NewTopic("seekExample", 3, (short) 1);
    }

    @Component
    public static class Listener extends AbstractConsumerSeekAware {

        @KafkaListener(id = "seekExample", topics = "seekExample", concurrency = "3")
        public void listen(String payload) {
            System.out.println("Listener received: " + payload);
        }

        @KafkaListener(id = "seekExample3", topics = "seekExample", concurrency = "3")
        public void listen3(String payload) {
            System.out.println("Listener3 received: " + payload);
        }

        public void seekToStart() {
            getSeekCallbacks().forEach((tp, callback) -> callback.seekToBeginning(tp.topic(), tp.partition()));
        }
    }

    @Component
    public static class Listener2 extends AbstractConsumerSeekAware {

        @KafkaListener(id = "seekExample2", topics = "seekExample", concurrency = "3")
        public void listen(String payload) {
            System.out.println("Listener2 received: " + payload);
        }

        public void seekToStart() {
            getSeekCallbacks().forEach((tp, callback) -> callback.seekToBeginning(tp.topic(), tp.partition()));
        }
    }
}

@SpringBootTest
@EmbeddedKafka(bootstrapServersProperty = "spring.kafka.bootstrap-servers")
@DirtiesContext
class KafkaGh2302ApplicationTest {

    @Autowired
    KafkaGh2302Application.Listener listener;

    @Autowired
    KafkaTemplate<String, String> template;

    @Test
    void contextLoads() throws InterruptedException {

        for (int i = 0; i < 50; i++) {
            this.template.send("seekExample", i % 3, "some_key", "test#" + i);
        }

        Thread.sleep(1000);
        System.out.println("====================================");
        this.listener.seekToStart();
        Thread.sleep(10000);
    }
}

[sobychacko]

I think the best course of action is to have a single consumer (KafkaListener) per class that extends AbstractConsumerSeekAware. If there are multiple listeners, the callbacks are applied against all the listeners in that particular class. We will see if we can come up with a solution, such as making an API level change to better accommodate group id's.

[garyrussell]

@sobychacko FYI, the thread's associated group is available in KafkaUtils, if that helps.

spring-kafka/spring-kafka/src/main/java/org/springframework/kafka/support/KafkaUtils.java

Lines 110 to 117 in 4a5a849

	/**
	* Get the group id for the consumer bound to this thread.
	* @return the group id.
	* @since 2.3
	*/
	public static String getConsumerGroupId() {
	return KafkaUtils.GROUP_IDS.get(Thread.currentThread());
	}