You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by ch...@apache.org on 2024/08/21 03:14:11 UTC
(kafka) branch trunk updated: KAFKA-16736 Remove `offsets.commit.required.acks` in 4.0 (#16938)
This is an automated email from the ASF dual-hosted git repository.
chia7712 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 944c1353a92 KAFKA-16736 Remove `offsets.commit.required.acks` in 4.0 (#16938)
944c1353a92 is described below
commit 944c1353a925858ea9bd9024a713cd7301f55133
Author: David Jacot <dj...@confluent.io>
AuthorDate: Wed Aug 21 05:14:04 2024 +0200
KAFKA-16736 Remove `offsets.commit.required.acks` in 4.0 (#16938)
Reviewers: Jeff Kim <je...@confluent.io>, Chia-Ping Tsai <ch...@gmail.com>
---
.../kafka/coordinator/group/GroupCoordinator.scala | 5 +----
.../coordinator/group/GroupMetadataManager.scala | 2 +-
core/src/main/scala/kafka/server/KafkaConfig.scala | 5 -----
.../group/GroupMetadataManagerTest.scala | 5 +----
.../scala/unit/kafka/server/KafkaConfigTest.scala | 1 -
.../coordinator/group/GroupCoordinatorConfig.java | 22 +---------------------
.../kafka/coordinator/group/OffsetConfig.java | 7 +------
.../group/GroupCoordinatorConfigTest.java | 8 --------
8 files changed, 5 insertions(+), 50 deletions(-)
diff --git a/core/src/main/scala/kafka/coordinator/group/GroupCoordinator.scala b/core/src/main/scala/kafka/coordinator/group/GroupCoordinator.scala
index 58a37e3333b..f592f3c9cea 100644
--- a/core/src/main/scala/kafka/coordinator/group/GroupCoordinator.scala
+++ b/core/src/main/scala/kafka/coordinator/group/GroupCoordinator.scala
@@ -36,7 +36,6 @@ import org.apache.kafka.coordinator.group.{Group, OffsetConfig}
import org.apache.kafka.server.record.BrokerCompressionType
import org.apache.kafka.storage.internals.log.VerificationGuard
-import scala.annotation.nowarn
import scala.collection.{Map, Seq, Set, immutable, mutable}
import scala.math.max
@@ -1768,7 +1767,6 @@ object GroupCoordinator {
GroupCoordinator(config, replicaManager, heartbeatPurgatory, rebalancePurgatory, time, metrics)
}
- @nowarn("cat=deprecation")
private[group] def offsetConfig(config: KafkaConfig) = new OffsetConfig(
config.groupCoordinatorConfig.offsetMetadataMaxSize,
config.groupCoordinatorConfig.offsetsLoadBufferSize,
@@ -1778,8 +1776,7 @@ object GroupCoordinator {
config.groupCoordinatorConfig.offsetsTopicSegmentBytes,
config.groupCoordinatorConfig.offsetsTopicReplicationFactor,
config.groupCoordinatorConfig.offsetTopicCompressionType,
- config.groupCoordinatorConfig.offsetCommitTimeoutMs,
- config.groupCoordinatorConfig.offsetCommitRequiredAcks
+ config.groupCoordinatorConfig.offsetCommitTimeoutMs
)
private[group] def apply(
diff --git a/core/src/main/scala/kafka/coordinator/group/GroupMetadataManager.scala b/core/src/main/scala/kafka/coordinator/group/GroupMetadataManager.scala
index 409761a8404..4386c226477 100644
--- a/core/src/main/scala/kafka/coordinator/group/GroupMetadataManager.scala
+++ b/core/src/main/scala/kafka/coordinator/group/GroupMetadataManager.scala
@@ -330,7 +330,7 @@ class GroupMetadataManager(brokerId: Int,
// call replica manager to append the group message
replicaManager.appendRecords(
timeout = config.offsetCommitTimeoutMs.toLong,
- requiredAcks = config.offsetCommitRequiredAcks,
+ requiredAcks = -1,
internalTopicsAllowed = true,
origin = AppendOrigin.COORDINATOR,
entriesPerPartition = records,
diff --git a/core/src/main/scala/kafka/server/KafkaConfig.scala b/core/src/main/scala/kafka/server/KafkaConfig.scala
index 39fbe283698..ab723e352bc 100755
--- a/core/src/main/scala/kafka/server/KafkaConfig.scala
+++ b/core/src/main/scala/kafka/server/KafkaConfig.scala
@@ -1067,11 +1067,6 @@ class KafkaConfig private(doLog: Boolean, val props: util.Map[_, _])
require(principalBuilderClass != null, s"${BrokerSecurityConfigs.PRINCIPAL_BUILDER_CLASS_CONFIG} must be non-null")
require(classOf[KafkaPrincipalSerde].isAssignableFrom(principalBuilderClass),
s"${BrokerSecurityConfigs.PRINCIPAL_BUILDER_CLASS_CONFIG} must implement KafkaPrincipalSerde")
-
-
- if (originals.containsKey(GroupCoordinatorConfig.OFFSET_COMMIT_REQUIRED_ACKS_CONFIG)) {
- warn(s"${GroupCoordinatorConfig.OFFSET_COMMIT_REQUIRED_ACKS_CONFIG} is deprecated and it will be removed in Apache Kafka 4.0.")
- }
}
/**
diff --git a/core/src/test/scala/unit/kafka/coordinator/group/GroupMetadataManagerTest.scala b/core/src/test/scala/unit/kafka/coordinator/group/GroupMetadataManagerTest.scala
index 97c7c2ee747..1ca368dcc6f 100644
--- a/core/src/test/scala/unit/kafka/coordinator/group/GroupMetadataManagerTest.scala
+++ b/core/src/test/scala/unit/kafka/coordinator/group/GroupMetadataManagerTest.scala
@@ -56,7 +56,6 @@ import org.mockito.{ArgumentCaptor, ArgumentMatchers}
import org.mockito.ArgumentMatchers.{any, anyInt, anyLong, anyShort}
import org.mockito.Mockito.{mock, reset, times, verify, when}
-import scala.annotation.nowarn
import scala.jdk.CollectionConverters._
import scala.collection.{immutable, _}
@@ -80,7 +79,6 @@ class GroupMetadataManagerTest {
val defaultRequireStable = false
val numOffsetsPartitions = 2
- @nowarn("cat=deprecation")
private val offsetConfig = {
val config = KafkaConfig.fromProps(TestUtils.createBrokerConfig(nodeId = 0, zkConnect = ""))
new OffsetConfig(config.groupCoordinatorConfig.offsetMetadataMaxSize,
@@ -91,8 +89,7 @@ class GroupMetadataManagerTest {
config.groupCoordinatorConfig.offsetsTopicSegmentBytes,
config.groupCoordinatorConfig.offsetsTopicReplicationFactor,
config.groupCoordinatorConfig.offsetTopicCompressionType,
- config.groupCoordinatorConfig.offsetCommitTimeoutMs,
- config.groupCoordinatorConfig.offsetCommitRequiredAcks)
+ config.groupCoordinatorConfig.offsetCommitTimeoutMs)
}
@BeforeEach
diff --git a/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala b/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala
index 4276dd9cb7d..5f7f3d5d820 100755
--- a/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala
+++ b/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala
@@ -982,7 +982,6 @@ class KafkaConfigTest {
case GroupCoordinatorConfig.OFFSETS_RETENTION_MINUTES_CONFIG => assertPropertyInvalid(baseProperties, name, "not_a_number", "0")
case GroupCoordinatorConfig.OFFSETS_RETENTION_CHECK_INTERVAL_MS_CONFIG => assertPropertyInvalid(baseProperties, name, "not_a_number", "0")
case GroupCoordinatorConfig.OFFSET_COMMIT_TIMEOUT_MS_CONFIG => assertPropertyInvalid(baseProperties, name, "not_a_number", "0")
- case GroupCoordinatorConfig.OFFSET_COMMIT_REQUIRED_ACKS_CONFIG => assertPropertyInvalid(baseProperties, name, "not_a_number", "-2")
case TransactionStateManagerConfigs.TRANSACTIONAL_ID_EXPIRATION_MS_CONFIG => assertPropertyInvalid(baseProperties, name, "not_a_number", "0", "-2")
case TransactionStateManagerConfigs.TRANSACTIONS_MAX_TIMEOUT_MS_CONFIG => assertPropertyInvalid(baseProperties, name, "not_a_number", "0", "-2")
case TransactionLogConfigs.TRANSACTIONS_TOPIC_MIN_ISR_CONFIG => assertPropertyInvalid(baseProperties, name, "not_a_number", "0", "-2")
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorConfig.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorConfig.java
index 7c7b964a0e1..87af61c1bb2 100644
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorConfig.java
+++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorConfig.java
@@ -204,11 +204,6 @@ public class GroupCoordinatorConfig {
public static final String OFFSET_COMMIT_TIMEOUT_MS_DOC = "Offset commit will be delayed until all replicas for the offsets topic receive the commit " +
"or this timeout is reached. This is similar to the producer request timeout.";
- @Deprecated
- public static final String OFFSET_COMMIT_REQUIRED_ACKS_CONFIG = "offsets.commit.required.acks";
- public static final short OFFSET_COMMIT_REQUIRED_ACKS_DEFAULT = -1;
- public static final String OFFSET_COMMIT_REQUIRED_ACKS_DOC = "DEPRECATED: The required acks before the commit can be accepted. In general, the default (-1) should not be overridden.";
-
public static final ConfigDef GROUP_COORDINATOR_CONFIG_DEF = new ConfigDef()
.define(GROUP_MIN_SESSION_TIMEOUT_MS_CONFIG, INT, GROUP_MIN_SESSION_TIMEOUT_MS_DEFAULT, MEDIUM, GROUP_MIN_SESSION_TIMEOUT_MS_DOC)
.define(GROUP_MAX_SESSION_TIMEOUT_MS_CONFIG, INT, GROUP_MAX_SESSION_TIMEOUT_MS_DEFAULT, MEDIUM, GROUP_MAX_SESSION_TIMEOUT_MS_DOC)
@@ -232,8 +227,7 @@ public class GroupCoordinatorConfig {
.define(OFFSETS_TOPIC_COMPRESSION_CODEC_CONFIG, INT, (int) OFFSETS_TOPIC_COMPRESSION_CODEC_DEFAULT.id, HIGH, OFFSETS_TOPIC_COMPRESSION_CODEC_DOC)
.define(OFFSETS_RETENTION_MINUTES_CONFIG, INT, OFFSETS_RETENTION_MINUTES_DEFAULT, atLeast(1), HIGH, OFFSETS_RETENTION_MINUTES_DOC)
.define(OFFSETS_RETENTION_CHECK_INTERVAL_MS_CONFIG, LONG, OFFSETS_RETENTION_CHECK_INTERVAL_MS_DEFAULT, atLeast(1), HIGH, OFFSETS_RETENTION_CHECK_INTERVAL_MS_DOC)
- .define(OFFSET_COMMIT_TIMEOUT_MS_CONFIG, INT, OFFSET_COMMIT_TIMEOUT_MS_DEFAULT, atLeast(1), HIGH, OFFSET_COMMIT_TIMEOUT_MS_DOC)
- .define(OFFSET_COMMIT_REQUIRED_ACKS_CONFIG, SHORT, OFFSET_COMMIT_REQUIRED_ACKS_DEFAULT, HIGH, OFFSET_COMMIT_REQUIRED_ACKS_DOC);
+ .define(OFFSET_COMMIT_TIMEOUT_MS_CONFIG, INT, OFFSET_COMMIT_TIMEOUT_MS_DEFAULT, atLeast(1), HIGH, OFFSET_COMMIT_TIMEOUT_MS_DOC);
public static final ConfigDef CONSUMER_GROUP_CONFIG_DEF = new ConfigDef()
.define(CONSUMER_GROUP_SESSION_TIMEOUT_MS_CONFIG, INT, CONSUMER_GROUP_SESSION_TIMEOUT_MS_DEFAULT, atLeast(1), MEDIUM, CONSUMER_GROUP_SESSION_TIMEOUT_MS_DOC)
.define(CONSUMER_GROUP_MIN_SESSION_TIMEOUT_MS_CONFIG, INT, CONSUMER_GROUP_MIN_SESSION_TIMEOUT_MS_DEFAULT, atLeast(1), MEDIUM, CONSUMER_GROUP_MIN_SESSION_TIMEOUT_MS_DOC)
@@ -278,7 +272,6 @@ public class GroupCoordinatorConfig {
private final int offsetsLoadBufferSize;
private final int offsetsTopicPartitions;
private final short offsetsTopicReplicationFactor;
- private final short offsetCommitRequiredAcks;
private final int consumerGroupMinSessionTimeoutMs;
private final int consumerGroupMaxSessionTimeoutMs;
private final int consumerGroupMinHeartbeatIntervalMs;
@@ -317,7 +310,6 @@ public class GroupCoordinatorConfig {
this.offsetsLoadBufferSize = config.getInt(GroupCoordinatorConfig.OFFSETS_LOAD_BUFFER_SIZE_CONFIG);
this.offsetsTopicPartitions = config.getInt(GroupCoordinatorConfig.OFFSETS_TOPIC_PARTITIONS_CONFIG);
this.offsetsTopicReplicationFactor = config.getShort(GroupCoordinatorConfig.OFFSETS_TOPIC_REPLICATION_FACTOR_CONFIG);
- this.offsetCommitRequiredAcks = config.getShort(GroupCoordinatorConfig.OFFSET_COMMIT_REQUIRED_ACKS_CONFIG);
this.consumerGroupMinSessionTimeoutMs = config.getInt(GroupCoordinatorConfig.CONSUMER_GROUP_MIN_SESSION_TIMEOUT_MS_CONFIG);
this.consumerGroupMaxSessionTimeoutMs = config.getInt(GroupCoordinatorConfig.CONSUMER_GROUP_MAX_SESSION_TIMEOUT_MS_CONFIG);
this.consumerGroupMinHeartbeatIntervalMs = config.getInt(GroupCoordinatorConfig.CONSUMER_GROUP_MIN_HEARTBEAT_INTERVAL_MS_CONFIG);
@@ -331,9 +323,6 @@ public class GroupCoordinatorConfig {
this.shareGroupMaxHeartbeatIntervalMs = config.getInt(GroupCoordinatorConfig.SHARE_GROUP_MAX_HEARTBEAT_INTERVAL_MS_CONFIG);
this.shareGroupMaxSize = config.getInt(GroupCoordinatorConfig.SHARE_GROUP_MAX_SIZE_CONFIG);
- require(offsetCommitRequiredAcks >= -1 && offsetCommitRequiredAcks <= offsetsTopicReplicationFactor,
- String.format("%s must be greater or equal to -1 and less or equal to %s", OFFSET_COMMIT_REQUIRED_ACKS_CONFIG, OFFSETS_TOPIC_REPLICATION_FACTOR_CONFIG));
-
// New group coordinator configs validation.
require(consumerGroupMaxHeartbeatIntervalMs >= consumerGroupMinHeartbeatIntervalMs,
String.format("%s must be greater than or equals to %s", CONSUMER_GROUP_MAX_HEARTBEAT_INTERVAL_MS_CONFIG, CONSUMER_GROUP_MIN_HEARTBEAT_INTERVAL_MS_CONFIG));
@@ -543,15 +532,6 @@ public class GroupCoordinatorConfig {
return offsetsTopicReplicationFactor;
}
- /**
- * DEPRECATED: The required acks before the commit can be accepted.
- * In general, the default (-1) should not be overridden.
- */
- @Deprecated // since 3.8
- public short offsetCommitRequiredAcks() {
- return offsetCommitRequiredAcks;
- }
-
/**
* The minimum allowed session timeout for registered consumers.
*/
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/OffsetConfig.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/OffsetConfig.java
index d77d2f57139..aaf1a415c8e 100644
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/OffsetConfig.java
+++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/OffsetConfig.java
@@ -28,7 +28,6 @@ public class OffsetConfig {
public final short offsetsTopicReplicationFactor;
public final CompressionType offsetsTopicCompressionType;
public final int offsetCommitTimeoutMs;
- public final short offsetCommitRequiredAcks;
/**
* Configuration settings for in-built offset management
@@ -50,8 +49,6 @@ public class OffsetConfig {
* order to achieve "atomic" commits.
* @param offsetCommitTimeoutMs The offset commit will be delayed until all replicas for the offsets topic receive the
* commit or this timeout is reached. (Similar to the producer request timeout.)
- * @param offsetCommitRequiredAcks The required acks before the commit can be accepted. In general, the default (-1)
- * should not be overridden.
*/
public OffsetConfig(int maxMetadataSize,
int loadBufferSize,
@@ -61,8 +58,7 @@ public class OffsetConfig {
int offsetsTopicSegmentBytes,
short offsetsTopicReplicationFactor,
CompressionType offsetsTopicCompressionType,
- int offsetCommitTimeoutMs,
- short offsetCommitRequiredAcks
+ int offsetCommitTimeoutMs
) {
this.maxMetadataSize = maxMetadataSize;
this.loadBufferSize = loadBufferSize;
@@ -73,6 +69,5 @@ public class OffsetConfig {
this.offsetsTopicReplicationFactor = offsetsTopicReplicationFactor;
this.offsetsTopicCompressionType = offsetsTopicCompressionType;
this.offsetCommitTimeoutMs = offsetCommitTimeoutMs;
- this.offsetCommitRequiredAcks = offsetCommitRequiredAcks;
}
}
diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorConfigTest.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorConfigTest.java
index e9e2c073b69..457fa16b80f 100644
--- a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorConfigTest.java
+++ b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorConfigTest.java
@@ -68,7 +68,6 @@ public class GroupCoordinatorConfigTest {
configs.put(GroupCoordinatorConfig.OFFSETS_LOAD_BUFFER_SIZE_CONFIG, 555);
configs.put(GroupCoordinatorConfig.OFFSETS_TOPIC_PARTITIONS_CONFIG, 111);
configs.put(GroupCoordinatorConfig.OFFSETS_TOPIC_REPLICATION_FACTOR_CONFIG, (short) 11);
- configs.put(GroupCoordinatorConfig.OFFSET_COMMIT_REQUIRED_ACKS_CONFIG, (short) 0);
configs.put(GroupCoordinatorConfig.CONSUMER_GROUP_MIN_SESSION_TIMEOUT_MS_CONFIG, 333);
configs.put(GroupCoordinatorConfig.CONSUMER_GROUP_MAX_SESSION_TIMEOUT_MS_CONFIG, 666);
configs.put(GroupCoordinatorConfig.CONSUMER_GROUP_MIN_HEARTBEAT_INTERVAL_MS_CONFIG, 111);
@@ -97,7 +96,6 @@ public class GroupCoordinatorConfigTest {
assertEquals(555, config.offsetsLoadBufferSize());
assertEquals(111, config.offsetsTopicPartitions());
assertEquals(11, config.offsetsTopicReplicationFactor());
- assertEquals(0, config.offsetCommitRequiredAcks());
assertEquals(333, config.consumerGroupMinSessionTimeoutMs());
assertEquals(666, config.consumerGroupMaxSessionTimeoutMs());
assertEquals(111, config.consumerGroupMinHeartbeatIntervalMs());
@@ -107,12 +105,6 @@ public class GroupCoordinatorConfigTest {
@Test
public void testInvalidConfigs() {
Map<String, Object> configs = new HashMap<>();
- configs.put(GroupCoordinatorConfig.OFFSET_COMMIT_REQUIRED_ACKS_CONFIG, (short) -2);
- configs.put(GroupCoordinatorConfig.OFFSETS_TOPIC_REPLICATION_FACTOR_CONFIG, (short) 3);
- assertEquals("offsets.commit.required.acks must be greater or equal to -1 and less or equal to offsets.topic.replication.factor",
- assertThrows(IllegalArgumentException.class, () -> createConfig(configs)).getMessage());
-
- configs.clear();
configs.put(GroupCoordinatorConfig.CONSUMER_GROUP_MAX_HEARTBEAT_INTERVAL_MS_CONFIG, 10);
configs.put(GroupCoordinatorConfig.CONSUMER_GROUP_MIN_HEARTBEAT_INTERVAL_MS_CONFIG, 20);
configs.put(GroupCoordinatorConfig.CONSUMER_GROUP_HEARTBEAT_INTERVAL_MS_CONFIG, 20);