groupId in RetryTopicConfiguration

[hariraogotit]

Team

The below note from https://docs.spring.io/spring-kafka/reference/html/#using-retrytopicconfiguration-beans is not clear to me

The retry topics' and dlt’s consumers will be assigned to a consumer group with a group id that is the combination of the one with you provide in the groupId parameter of the @KafkaListener annotation with the topic’s suffix. If you don’t provide any they’ll all belong to the same group, and rebalance on a retry topic will cause an unnecessary rebalance on the main topic.

In general

1. Say we have a topic named - order-topic with 2 partitions
2. An order-service with a group id order-service-123
3. At max 2 instances of order-services can consume from the above topic
4. Say this as retryTopicSuffix as -order-retry and dltSuffix as order-dlt

Similarly

1. An order-somethingelse-service with a group id order-somthingelse-service-123
2. At max 2 instances of order-somethingelse-service can consume from the same above topic
3. Say this as retryTopicSuffix as -order-somthingelse-retry and dltSuffix as order-somthingelse-dlt

If we introduce say Non-blocking retry on order-service and order-somethingelse-service and say NO group id is passed in any of the KafkaListners, what does it mean by

If you don’t provide any they’ll all belong to the same group, and rebalance on a retry topic will cause an unnecessary rebalance on the main topic. ?

Is this recommending to introduce a new group id per KafkaListner so that the dlt and retry topic belong to a group id?
If one of the instance of the service go down anyway the main topic, retry and dlt topic message consumption needs to be rebalanced (lets assume all the topics - main, retry and dlt all have same no of partitions)

In general 2 instances of a service with a group id consuming messages from 2 partitioned topic when one instance goes down, rebalance occurs and the active instance will consume from both the partitions.

So why is this new layer of group id in KafkaListner and what is it supposed to prevent?

Kindly clarify.

[garyrussell]

It's pretty straightforward.

@RetryableTopic
@KafkaListener(topics = "topic1")
void listen1(String in) {
	System.out.println(in);
}

@RetryableTopic
@KafkaListener(topics = "topic2")
void listen2(String in) {
	System.out.println(in);
}

With the default configuration, these listeners, and consumer group id of foo, we get

foo: partitions assigned: [topic2-retry-1-0]
foo: partitions assigned: [topic2-0]
foo: partitions assigned: [topic2-dlt-0]
foo: partitions assigned: [topic1-retry-0-0]
foo: partitions assigned: [topic1-0]
foo: partitions assigned: [topic2-retry-0-0]
foo: partitions assigned: [topic1-retry-1-0]
foo: partitions assigned: [topic1-dlt-0]

All 8 consumers are in the same group. This means if there is a rebalance on topic2 (and its retry containers), it will cause an unnecessary rebalance on topic 1.

with

@RetryableTopic
@KafkaListener(id = "one", topics = "topic1")
void listen1(String in) {
	System.out.println(in);
}

@RetryableTopic
@KafkaListener(id = "two", topics = "topic2")
void listen2(String in) {
	System.out.println(in);
}

we get

two: partitions assigned: [topic2-0]
two-retry-0: partitions assigned: [topic2-retry-0-0]
two-retry-1: partitions assigned: [topic2-retry-1-0]
two-dlt: partitions assigned: [topic2-dlt-0]
one-retry-1: partitions assigned: [topic1-retry-1-0]
one: partitions assigned: [topic1-0]
one-dlt: partitions assigned: [topic1-dlt-0]
one-retry-0: partitions assigned: [topic1-retry-0-0]

with each consumer in its own group.

By default, the id is used as the group id.

[garyrussell]

But, let's say you use the endpoint registry to stop one of the retry containers (for some reason), it will cause a rebalance on all containers.

[hariraogotit]

Thanks @garyrussell

So this means irrespective of using non blocking retries its better to have

group id in each @KafkaListner rather than using one group id per service?

[garyrussell]

Yes, exactly; it is also good practice to give each listener an id so you can control its lifecycle (stop/start/pause/resume) via the KafkaListenerEndpointRegistry; if you don't give it an id, it gets a system generated id.

If you want a different group.id to the id, then use the groupId property. See

	/**
	 * The unique identifier of the container for this listener.
	 * <p>If none is specified an auto-generated id is used.
	 * <p>Note: When provided, this value will override the group id property
	 * in the consumer factory configuration, unless {@link #idIsGroup()}
	 * is set to false or {@link #groupId()} is provided.
	 * <p>SpEL {@code #{...}} and property place holders {@code ${...}} are supported.
	 * @return the {@code id} for the container managing for this endpoint.
	 * @see org.springframework.kafka.config.KafkaListenerEndpointRegistry#getListenerContainer(String)
	 */
	String id() default "";
...
	/**
	 * Override the {@code group.id} property for the consumer factory with this value
	 * for this listener only.
	 * <p>SpEL {@code #{...}} and property place holders {@code ${...}} are supported.
	 * @return the group id.
	 * @since 1.3
	 */
	String groupId() default "";

	/**
	 * When {@link #groupId() groupId} is not provided, use the {@link #id() id} (if
	 * provided) as the {@code group.id} property for the consumer. Set to false, to use
	 * the {@code group.id} from the consumer factory.
	 * @return false to disable.
	 * @since 1.3
	 */
	boolean idIsGroup() default true;

[VitorJuliani]

@garyrussell Is there any way to override this groupId + suffix behavior? I have 2 consumers that consume messages from the same topic but execute different processes. I would like to add a group ID to avoid the rebalancing you mentioned, but when I edit both the group ID and the suffix, I end up with a very long consumer group name and duplicated information.

[artembilan]

@VitorJuliani ,

Gary is not with the team any more.
While he might answer your question, but better to ask a new one with much more info.
Not clear if your context is relevant somehow to a subject of this old closed discussion.

Thank you for understanding!