Newly created groupId will delete automatically upon application shutdown.

[cmabdullah]

A unique consumer group can be created for a Kafka listener by setting uniqueGroupId=true on the @KafkaListener annotation. This automatically generates a unique groupId each time the application starts. However, once the application shuts down, the generated groupId remains in Kafka and is not automatically deleted.

package org.cm.kafka;

import io.micronaut.configuration.kafka.annotation.KafkaListener;
import io.micronaut.configuration.kafka.annotation.Topic;

@KafkaListener(groupId = "message-app", uniqueGroupId = true)
public class KafkaMessageConsumerClient {
	@Topic("user-online")
	void consumeEvent(String message) {
		System.out.println("event from Kafka: " + message);
	}
}

This is the groupId list remains on kafka

Multiple consumer groups support is added on this.
where a flag is added (uniqueGroupId) in the annotation KafkaListener, if uniqueGroupId is true, Then the app will create a unique groupID.

However, the generated groupId persists in Kafka even after the application shuts down.

We are currently developing a whiteboard application where each service have multiple instances, and any instance can broadcasts events, and all other instances need to receive these event. This follows a pub/sub pattern.
As this feature is critical for our use case, we want to ensure better manageability by avoiding unnecessary groupId entries in Kafka.

To address this, I have implemented a quick fix that still relies on annotations to delete newly created groupId will deleted automatically upon application shutdown.

I successfully extended the @KafkaListener annotation and tested the new feature using plain Kafka.

My initial idea is, I have added uniqueGroupIdDeleteOnShutdown field on @KafkaListener, If uniqueGroupIdDeleteOnShutdown is set to true alongside uniqueGroupId, the unique group ID will be automatically deleted during the application shutdown process, ensuring better manageability of Kafka group IDs.

    @Override
    public void onApplicationEvent(ApplicationShutdownEvent event) {
        LOG.trace("Application shutdown initiated. Preparing to delete registered Kafka unique consumer groups.");
        List<String> uniqueGroupIdsDeleteOnShutdown = registerConsumerForGroupDeletion.values()
            .stream()
            .filter(consumerState -> consumerState.kafkaConsumer.groupMetadata() != null)
            .map(consumerState -> consumerState.kafkaConsumer.groupMetadata().groupId())
            .toList();
        if (!uniqueGroupIdsDeleteOnShutdown.isEmpty()) {
            LOG.trace("Closing {} consumers and attempting to delete the following consumer groups: {}",
                uniqueGroupIdsDeleteOnShutdown.size(), uniqueGroupIdsDeleteOnShutdown);
            closeConsumers();
            adminClient.deleteConsumerGroups(uniqueGroupIdsDeleteOnShutdown)
                .all().whenComplete((voidResult, throwable) -> {
                    if (throwable == null) {
                        LOG.trace("Successfully deleted the following consumer groups: {}", uniqueGroupIdsDeleteOnShutdown);
                    } else {
                        LOG.error("Failed to delete the following consumer groups: {}. Error: {}",
                            uniqueGroupIdsDeleteOnShutdown, throwable.getMessage(), throwable);
                    }
                });
        } else {
            LOG.trace("No unique consumer groups are registered for deletion.");
        }
    }

[CLAassistant]

All committers have signed the CLA.

[graemerocher]

can you add a test please?

[iNviNho]

This is a nice addition (I haven't checked the code yet) but please bear in mind that not all applications will be granted access to delete consumer group.

If this is meant to go through, I would either make this deletion by default disabled or make the log level only warning or info as this is not critical functionality that has to work.

Byt the way you can always decrease retention on internal topic that you use to store committed offset and therefore such unused consumer groups will eventually disappear.

[graemerocher]

it should absolutely be opt-in

[cmabdullah]

@graemerocher,
Thank you for the feedback!
I have added an integration test as requested and also provided a video demonstration at the given link.

Additionally, I’ve made the deletion functionality disabled by default, ensuring it aligns with the concerns raised.

I’m also sharing the build log for reference. Please let me know if any further refinements are needed. Looking forward to your thoughts!