From b0af3d2227ce70ad6928370ba347cd19b9afc04d Mon Sep 17 00:00:00 2001
From: Soby Chacko <soby.chacko@broadcom.com>
Date: Fri, 17 May 2024 12:28:38 -0400
Subject: [PATCH] GH-3254: Group protocol in ConsumerProperties

Fixes: #3254

* Add group.protocol property to ConsumerProperties
---
 .../kafka/listener/ConsumerProperties.java    | 26 ++++++++++++++++++-
 1 file changed, 25 insertions(+), 1 deletion(-)

diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/ConsumerProperties.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/ConsumerProperties.java
index 097842cf4a..6faf3684de 100644
--- a/spring-kafka/src/main/java/org/springframework/kafka/listener/ConsumerProperties.java
+++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/ConsumerProperties.java
@@ -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.
@@ -35,6 +35,7 @@
  * Common consumer properties.
  *
  * @author Gary Russell
+ * @author Soby Chacko
  * @since 2.3
  *
  */
@@ -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;
+
 	/**
 	 * Create properties for a container that will subscribe to the specified topics.
 	 * @param topics the topics.
@@ -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.
+	 *
+	 * @param groupProtocol group protocol via the group.protocol Consumer property
+	 */
+	public void setGroupProtocol(String groupProtocol) {
+		this.groupProtocol = groupProtocol;
+	}
+
 	@Override
 	public String toString() {
 		return "ConsumerProperties ["