You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by da...@apache.org on 2023/11/21 10:52:06 UTC
(kafka) branch trunk updated: MINOR: Mark ConsumerGroupHeartbeat API (v1), OffsetCommit API (v9) and OffsetFetch API (v9) as stable (KIP-848) (#14801)
This is an automated email from the ASF dual-hosted git repository.
dajac pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 7826d5fc8ab MINOR: Mark ConsumerGroupHeartbeat API (v1), OffsetCommit API (v9) and OffsetFetch API (v9) as stable (KIP-848) (#14801)
7826d5fc8ab is described below
commit 7826d5fc8ab695a5ad927338469ddc01b435a298
Author: David Jacot <dj...@confluent.io>
AuthorDate: Tue Nov 21 11:51:59 2023 +0100
MINOR: Mark ConsumerGroupHeartbeat API (v1), OffsetCommit API (v9) and OffsetFetch API (v9) as stable (KIP-848) (#14801)
We plan to ship an early access of KIP-848 in AK 3.7. Therefore, we need to mark the ConsumerGroupHeartbeat API (v1), OffsetCommit API (v9) and OffsetFetch API (v9) as stable.
Reviewers: Andrew Schofield <as...@confluent.io>, Justine Olshan <jo...@confluent.io>
---
.../consumer/internals/MembershipManagerImpl.java | 4 ++--
.../message/ConsumerGroupHeartbeatRequest.json | 4 ----
.../common/message/OffsetCommitRequest.json | 3 ---
.../common/message/OffsetFetchRequest.json | 4 ----
.../server/ConsumerGroupHeartbeatRequestTest.scala | 27 +++++-----------------
.../kafka/server/DeleteGroupsRequestTest.scala | 1 -
.../unit/kafka/server/ListGroupsRequestTest.scala | 1 -
.../kafka/server/OffsetCommitRequestTest.scala | 1 -
.../kafka/server/OffsetDeleteRequestTest.scala | 1 -
.../unit/kafka/server/OffsetFetchRequestTest.scala | 7 ------
10 files changed, 8 insertions(+), 45 deletions(-)
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/MembershipManagerImpl.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/MembershipManagerImpl.java
index 4a98d23a62f..cdfa9881e8e 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/MembershipManagerImpl.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/MembershipManagerImpl.java
@@ -129,7 +129,7 @@ public class MembershipManagerImpl implements MembershipManager, ClusterResource
* Member ID assigned by the server to the member, received in a heartbeat response when
* joining the group specified in {@link #groupId}
*/
- private String memberId;
+ private String memberId = "";
/**
* Current epoch of the member. It will be set to 0 by the member, and provided to the server
@@ -137,7 +137,7 @@ public class MembershipManagerImpl implements MembershipManager, ClusterResource
* incremented as the member reconciles and acknowledges the assignments it receives. It will
* be reset to 0 if the member gets fenced.
*/
- private int memberEpoch;
+ private int memberEpoch = 0;
/**
* Current state of this member as part of the consumer group, as defined in {@link MemberState}
diff --git a/clients/src/main/resources/common/message/ConsumerGroupHeartbeatRequest.json b/clients/src/main/resources/common/message/ConsumerGroupHeartbeatRequest.json
index e86a79fcd68..95c26421ce2 100644
--- a/clients/src/main/resources/common/message/ConsumerGroupHeartbeatRequest.json
+++ b/clients/src/main/resources/common/message/ConsumerGroupHeartbeatRequest.json
@@ -18,10 +18,6 @@
"type": "request",
"listeners": ["zkBroker", "broker"],
"name": "ConsumerGroupHeartbeatRequest",
- // The ConsumerGroupHeartbeat API is added as part of KIP-848 and is still
- // under development. Hence, the API is not exposed by default by brokers
- // unless explicitly enabled.
- "latestVersionUnstable": true,
"validVersions": "0",
"flexibleVersions": "0+",
"fields": [
diff --git a/clients/src/main/resources/common/message/OffsetCommitRequest.json b/clients/src/main/resources/common/message/OffsetCommitRequest.json
index c11b56c9544..27edbba5272 100644
--- a/clients/src/main/resources/common/message/OffsetCommitRequest.json
+++ b/clients/src/main/resources/common/message/OffsetCommitRequest.json
@@ -34,9 +34,6 @@
//
// Version 9 is the first version that can be used with the new consumer group protocol (KIP-848). The
// request is the same as version 8.
- // Version 9 is added as part of KIP-848 and is still under development. Hence, the last version of the
- // API is not exposed by default by brokers unless explicitly enabled.
- "latestVersionUnstable": true,
"validVersions": "0-9",
"flexibleVersions": "8+",
"fields": [
diff --git a/clients/src/main/resources/common/message/OffsetFetchRequest.json b/clients/src/main/resources/common/message/OffsetFetchRequest.json
index 33dab5c957c..85a5c2d399b 100644
--- a/clients/src/main/resources/common/message/OffsetFetchRequest.json
+++ b/clients/src/main/resources/common/message/OffsetFetchRequest.json
@@ -36,10 +36,6 @@
//
// Version 9 is the first version that can be used with the new consumer group protocol (KIP-848). It adds
// the MemberId and MemberEpoch fields. Those are filled in and validated when the new consumer protocol is used.
- //
- // Version 9 is added as part of KIP-848 and is still under development. Hence, the last version of the
- // API is not exposed by default by brokers unless explicitly enabled.
- "latestVersionUnstable": true,
"validVersions": "0-9",
"flexibleVersions": "6+",
"fields": [
diff --git a/core/src/test/scala/unit/kafka/server/ConsumerGroupHeartbeatRequestTest.scala b/core/src/test/scala/unit/kafka/server/ConsumerGroupHeartbeatRequestTest.scala
index 42bfb97bb25..9c26ef52a66 100644
--- a/core/src/test/scala/unit/kafka/server/ConsumerGroupHeartbeatRequestTest.scala
+++ b/core/src/test/scala/unit/kafka/server/ConsumerGroupHeartbeatRequestTest.scala
@@ -24,12 +24,11 @@ import kafka.utils.TestUtils
import org.apache.kafka.common.message.{ConsumerGroupHeartbeatRequestData, ConsumerGroupHeartbeatResponseData}
import org.apache.kafka.common.protocol.Errors
import org.apache.kafka.common.requests.{ConsumerGroupHeartbeatRequest, ConsumerGroupHeartbeatResponse}
-import org.junit.jupiter.api.Assertions.{assertEquals, assertNotNull, assertThrows}
+import org.junit.jupiter.api.Assertions.{assertEquals, assertNotNull}
import org.junit.jupiter.api.Tag
import org.junit.jupiter.api.Timeout
import org.junit.jupiter.api.extension.ExtendWith
-import java.io.EOFException
import java.util.stream.Collectors
import scala.jdk.CollectionConverters._
@@ -39,20 +38,10 @@ import scala.jdk.CollectionConverters._
@Tag("integration")
class ConsumerGroupHeartbeatRequestTest(cluster: ClusterInstance) {
- @ClusterTest
- def testConsumerGroupHeartbeatIsDisabledByDefault(): Unit = {
- val consumerGroupHeartbeatRequest = new ConsumerGroupHeartbeatRequest.Builder(
- new ConsumerGroupHeartbeatRequestData(),
- true
- ).build()
- assertThrows(classOf[EOFException], () => connectAndReceive(consumerGroupHeartbeatRequest))
- }
-
- @ClusterTest(serverProperties = Array(new ClusterConfigProperty(key = "unstable.api.versions.enable", value = "true")))
+ @ClusterTest()
def testConsumerGroupHeartbeatIsAccessibleWhenEnabled(): Unit = {
val consumerGroupHeartbeatRequest = new ConsumerGroupHeartbeatRequest.Builder(
- new ConsumerGroupHeartbeatRequestData(),
- true
+ new ConsumerGroupHeartbeatRequestData()
).build()
val consumerGroupHeartbeatResponse = connectAndReceive(consumerGroupHeartbeatRequest)
@@ -61,7 +50,6 @@ class ConsumerGroupHeartbeatRequestTest(cluster: ClusterInstance) {
}
@ClusterTest(clusterType = Type.KRAFT, serverProperties = Array(
- new ClusterConfigProperty(key = "unstable.api.versions.enable", value = "true"),
new ClusterConfigProperty(key = "group.coordinator.new.enable", value = "true"),
new ClusterConfigProperty(key = "offsets.topic.num.partitions", value = "1"),
new ClusterConfigProperty(key = "offsets.topic.replication.factor", value = "1")
@@ -86,8 +74,7 @@ class ConsumerGroupHeartbeatRequestTest(cluster: ClusterInstance) {
.setMemberEpoch(0)
.setRebalanceTimeoutMs(5 * 60 * 1000)
.setSubscribedTopicNames(List("foo").asJava)
- .setTopicPartitions(List.empty.asJava),
- true
+ .setTopicPartitions(List.empty.asJava)
).build()
// Send the request until receiving a successful response. There is a delay
@@ -115,8 +102,7 @@ class ConsumerGroupHeartbeatRequestTest(cluster: ClusterInstance) {
new ConsumerGroupHeartbeatRequestData()
.setGroupId("grp")
.setMemberId(consumerGroupHeartbeatResponse.data.memberId)
- .setMemberEpoch(consumerGroupHeartbeatResponse.data.memberEpoch),
- true
+ .setMemberEpoch(consumerGroupHeartbeatResponse.data.memberEpoch)
).build()
// This is the expected assignment.
@@ -142,8 +128,7 @@ class ConsumerGroupHeartbeatRequestTest(cluster: ClusterInstance) {
new ConsumerGroupHeartbeatRequestData()
.setGroupId("grp")
.setMemberId(consumerGroupHeartbeatResponse.data.memberId)
- .setMemberEpoch(-1),
- true
+ .setMemberEpoch(-1)
).build()
consumerGroupHeartbeatResponse = connectAndReceive(consumerGroupHeartbeatRequest)
diff --git a/core/src/test/scala/unit/kafka/server/DeleteGroupsRequestTest.scala b/core/src/test/scala/unit/kafka/server/DeleteGroupsRequestTest.scala
index 805d014c66a..2cc3b09c0dc 100644
--- a/core/src/test/scala/unit/kafka/server/DeleteGroupsRequestTest.scala
+++ b/core/src/test/scala/unit/kafka/server/DeleteGroupsRequestTest.scala
@@ -32,7 +32,6 @@ import org.junit.jupiter.api.extension.ExtendWith
@Tag("integration")
class DeleteGroupsRequestTest(cluster: ClusterInstance) extends GroupCoordinatorBaseRequestTest(cluster) {
@ClusterTest(serverProperties = Array(
- new ClusterConfigProperty(key = "unstable.api.versions.enable", value = "true"),
new ClusterConfigProperty(key = "group.coordinator.new.enable", value = "true"),
new ClusterConfigProperty(key = "group.consumer.max.session.timeout.ms", value = "600000"),
new ClusterConfigProperty(key = "group.consumer.session.timeout.ms", value = "600000"),
diff --git a/core/src/test/scala/unit/kafka/server/ListGroupsRequestTest.scala b/core/src/test/scala/unit/kafka/server/ListGroupsRequestTest.scala
index ad99e93ffdd..eff20274051 100644
--- a/core/src/test/scala/unit/kafka/server/ListGroupsRequestTest.scala
+++ b/core/src/test/scala/unit/kafka/server/ListGroupsRequestTest.scala
@@ -33,7 +33,6 @@ import org.junit.jupiter.api.extension.ExtendWith
@Tag("integration")
class ListGroupsRequestTest(cluster: ClusterInstance) extends GroupCoordinatorBaseRequestTest(cluster) {
@ClusterTest(serverProperties = Array(
- new ClusterConfigProperty(key = "unstable.api.versions.enable", value = "true"),
new ClusterConfigProperty(key = "group.coordinator.new.enable", value = "true"),
new ClusterConfigProperty(key = "group.consumer.max.session.timeout.ms", value = "600000"),
new ClusterConfigProperty(key = "group.consumer.session.timeout.ms", value = "600000"),
diff --git a/core/src/test/scala/unit/kafka/server/OffsetCommitRequestTest.scala b/core/src/test/scala/unit/kafka/server/OffsetCommitRequestTest.scala
index 1437aa63418..2ff776aa42a 100644
--- a/core/src/test/scala/unit/kafka/server/OffsetCommitRequestTest.scala
+++ b/core/src/test/scala/unit/kafka/server/OffsetCommitRequestTest.scala
@@ -31,7 +31,6 @@ import org.junit.jupiter.api.extension.ExtendWith
class OffsetCommitRequestTest(cluster: ClusterInstance) extends GroupCoordinatorBaseRequestTest(cluster) {
@ClusterTest(serverProperties = Array(
- new ClusterConfigProperty(key = "unstable.api.versions.enable", value = "true"),
new ClusterConfigProperty(key = "group.coordinator.new.enable", value = "true"),
new ClusterConfigProperty(key = "group.consumer.max.session.timeout.ms", value = "600000"),
new ClusterConfigProperty(key = "group.consumer.session.timeout.ms", value = "600000"),
diff --git a/core/src/test/scala/unit/kafka/server/OffsetDeleteRequestTest.scala b/core/src/test/scala/unit/kafka/server/OffsetDeleteRequestTest.scala
index 99fc4151b50..719697da131 100644
--- a/core/src/test/scala/unit/kafka/server/OffsetDeleteRequestTest.scala
+++ b/core/src/test/scala/unit/kafka/server/OffsetDeleteRequestTest.scala
@@ -30,7 +30,6 @@ import org.junit.jupiter.api.extension.ExtendWith
@Tag("integration")
class OffsetDeleteRequestTest(cluster: ClusterInstance) extends GroupCoordinatorBaseRequestTest(cluster) {
@ClusterTest(serverProperties = Array(
- new ClusterConfigProperty(key = "unstable.api.versions.enable", value = "true"),
new ClusterConfigProperty(key = "group.coordinator.new.enable", value = "true"),
new ClusterConfigProperty(key = "group.consumer.max.session.timeout.ms", value = "600000"),
new ClusterConfigProperty(key = "group.consumer.session.timeout.ms", value = "600000"),
diff --git a/core/src/test/scala/unit/kafka/server/OffsetFetchRequestTest.scala b/core/src/test/scala/unit/kafka/server/OffsetFetchRequestTest.scala
index 6adf4910c81..828f8ab0aaf 100644
--- a/core/src/test/scala/unit/kafka/server/OffsetFetchRequestTest.scala
+++ b/core/src/test/scala/unit/kafka/server/OffsetFetchRequestTest.scala
@@ -39,7 +39,6 @@ import scala.jdk.CollectionConverters._
class OffsetFetchRequestTest(cluster: ClusterInstance) extends GroupCoordinatorBaseRequestTest(cluster) {
@ClusterTest(serverProperties = Array(
- new ClusterConfigProperty(key = "unstable.api.versions.enable", value = "true"),
new ClusterConfigProperty(key = "group.coordinator.new.enable", value = "true"),
new ClusterConfigProperty(key = "group.consumer.max.session.timeout.ms", value = "600000"),
new ClusterConfigProperty(key = "group.consumer.session.timeout.ms", value = "600000"),
@@ -73,7 +72,6 @@ class OffsetFetchRequestTest(cluster: ClusterInstance) extends GroupCoordinatorB
}
@ClusterTest(serverProperties = Array(
- new ClusterConfigProperty(key = "unstable.api.versions.enable", value = "true"),
new ClusterConfigProperty(key = "group.coordinator.new.enable", value = "true"),
new ClusterConfigProperty(key = "group.consumer.max.session.timeout.ms", value = "600000"),
new ClusterConfigProperty(key = "group.consumer.session.timeout.ms", value = "600000"),
@@ -85,7 +83,6 @@ class OffsetFetchRequestTest(cluster: ClusterInstance) extends GroupCoordinatorB
}
@ClusterTest(serverProperties = Array(
- new ClusterConfigProperty(key = "unstable.api.versions.enable", value = "true"),
new ClusterConfigProperty(key = "group.coordinator.new.enable", value = "true"),
new ClusterConfigProperty(key = "group.consumer.max.session.timeout.ms", value = "600000"),
new ClusterConfigProperty(key = "group.consumer.session.timeout.ms", value = "600000"),
@@ -97,7 +94,6 @@ class OffsetFetchRequestTest(cluster: ClusterInstance) extends GroupCoordinatorB
}
@ClusterTest(clusterType = Type.ALL, serverProperties = Array(
- new ClusterConfigProperty(key = "unstable.api.versions.enable", value = "false"),
new ClusterConfigProperty(key = "group.coordinator.new.enable", value = "false"),
new ClusterConfigProperty(key = "group.consumer.max.session.timeout.ms", value = "600000"),
new ClusterConfigProperty(key = "group.consumer.session.timeout.ms", value = "600000"),
@@ -109,7 +105,6 @@ class OffsetFetchRequestTest(cluster: ClusterInstance) extends GroupCoordinatorB
}
@ClusterTest(serverProperties = Array(
- new ClusterConfigProperty(key = "unstable.api.versions.enable", value = "true"),
new ClusterConfigProperty(key = "group.coordinator.new.enable", value = "true"),
new ClusterConfigProperty(key = "group.consumer.max.session.timeout.ms", value = "600000"),
new ClusterConfigProperty(key = "group.consumer.session.timeout.ms", value = "600000"),
@@ -121,7 +116,6 @@ class OffsetFetchRequestTest(cluster: ClusterInstance) extends GroupCoordinatorB
}
@ClusterTest(serverProperties = Array(
- new ClusterConfigProperty(key = "unstable.api.versions.enable", value = "true"),
new ClusterConfigProperty(key = "group.coordinator.new.enable", value = "true"),
new ClusterConfigProperty(key = "group.consumer.max.session.timeout.ms", value = "600000"),
new ClusterConfigProperty(key = "group.consumer.session.timeout.ms", value = "600000"),
@@ -133,7 +127,6 @@ class OffsetFetchRequestTest(cluster: ClusterInstance) extends GroupCoordinatorB
}
@ClusterTest(clusterType = Type.ALL, serverProperties = Array(
- new ClusterConfigProperty(key = "unstable.api.versions.enable", value = "false"),
new ClusterConfigProperty(key = "group.coordinator.new.enable", value = "false"),
new ClusterConfigProperty(key = "group.consumer.max.session.timeout.ms", value = "600000"),
new ClusterConfigProperty(key = "group.consumer.session.timeout.ms", value = "600000"),