Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

GH-3254: Group protocol in ConsumerProperties #3255

Closed
wants to merge 1 commit into from
Closed

GH-3254: Group protocol in ConsumerProperties #3255

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
26 changes: 25 additions & 1 deletion spring-kafka/src/main/java/org/springframework/kafka/listener/ConsumerProperties.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
/*
* Copyright 2019-2022 the original author or authors.
* Copyright 2019-2024 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
Expand Down Expand Up @@ -35,6 +35,7 @@
* Common consumer properties.
*
* @author Gary Russell
* @author Soby Chacko
* @since 2.3
*
*/
Expand Down Expand Up @@ -116,6 +117,12 @@ public class ConsumerProperties {

private boolean checkDeserExWhenValueNull;

/**
* Container group protocol that will be set on the Kafka Consumer.
* Default to classic consumer protocol.
*/
private String groupProtocol;
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Isn't there some enum to rely on?
Or we might introduce one as an inner in this ConsumerProperties class...


/**
* Create properties for a container that will subscribe to the specified topics.
* @param topics the topics.
Expand Down Expand Up @@ -501,6 +508,23 @@ public void setCheckDeserExWhenValueNull(boolean checkDeserExWhenValueNull) {
this.checkDeserExWhenValueNull = checkDeserExWhenValueNull;
}

/**
* Return the container's group protocol.
* @return Container group protocol set via the group.protocol Consumer property
*/
public String getGroupProtocol() {
return this.groupProtocol;
}

/**
* Set the container group protocol.
*
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

No blank lines in method Javadocs.
The @since is supposed to be there on the public API.
And why is this property is not set on the target KafkaConsumer in this PR?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

In this PR, we are not really using this property directly - but rather giving an opportunity to set it via the ConsumerProperties. This property needs to be eventually exposed from Spring Boot auto-configuration, so this PR is sort of the ground work. As this new consumer-group evolves, I would imagine that we may need to make more changes in the framework. At this point, we simply pass it down to the KafkaConsumer as it is if it is set. Otherwise, the KafkaConumser uses the defaults. https://github.com/spring-projects/spring-kafka/blob/main/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java#L862

Do you still see a need to set this somewhere in the framework?

* @param groupProtocol group protocol via the group.protocol Consumer property
*/
public void setGroupProtocol(String groupProtocol) {
this.groupProtocol = groupProtocol;
}

@Override
public String toString() {
return "ConsumerProperties ["
Expand Down