Skip to content

KAFKA-16968: Make 3.8-IV0 a stable MetadataVersion and create 3.9-IV0 #16347

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

Merged
cmccabe merged 16 commits into from
Jun 27, 2024
Merged

KAFKA-16968: Make 3.8-IV0 a stable MetadataVersion and create 3.9-IV0 #16347

Show file tree
Hide file tree
Changes from 13 commits
Commits
Show all changes
16 commits
Select commit Hold shift + click to select a range
fbb566f
KAFKA-16968: Make 3.8-IV0 a stable MetadataVersion and create 3.9-IV0
cmccabe Jun 15, 2024
ea90660
re-add 4.0 and add a comment about unstable MVs
cmccabe Jun 17, 2024
43daca5
fix test
cmccabe Jun 18, 2024
12370b4
Bump ListOffsetsRequest to v9, mark latest version unstable
cmccabe Jun 18, 2024
f96af84
address review comments
cmccabe Jun 18, 2024
4e9c025
fixes
cmccabe Jun 18, 2024
68d1265
more updates
cmccabe Jun 20, 2024
84ea68c
address review comments
cmccabe Jun 24, 2024
77e8b86
Merge branch 'trunk' into KAFKA-16968
cmccabe Jun 24, 2024
95e1a49
Add comment
cmccabe Jun 24, 2024
933aed5
fix tests
cmccabe Jun 25, 2024
a2c7f0a
Merge branch 'trunk' into KAFKA-16968
cmccabe Jun 25, 2024
c06985d
fix checkstyle
cmccabe Jun 25, 2024
c183b4d
test fixes
cmccabe Jun 26, 2024
1545da7
Fix ApiVersionsRequestTest
cmccabe Jun 27, 2024
fccf7cb
Fix SaslApiVersionsRequestTest
cmccabe Jun 27, 2024
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
5 changes: 4 additions & 1 deletion clients/src/main/resources/common/message/ListOffsetsRequest.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,9 +34,12 @@
// Version 7 enables listing offsets by max timestamp (KIP-734).
//
// Version 8 enables listing offsets by local log start offset (KIP-405).
"validVersions": "0-8",
//
// Version 9 enables listing offsets by last tiered offset (KIP-1005).
"validVersions": "0-9",
"deprecatedVersions": "0",
"flexibleVersions": "6+",
"latestVersionUnstable": true,
Comment thread
cmccabe marked this conversation as resolved.
Comment thread
cmccabe marked this conversation as resolved.
"fields": [
{ "name": "ReplicaId", "type": "int32", "versions": "0+", "entityType": "brokerId",
"about": "The broker ID of the requester, or -1 if this request is being made by a normal consumer." },
Expand Down
4 changes: 3 additions & 1 deletion clients/src/main/resources/common/message/ListOffsetsResponse.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,7 +34,9 @@
//
// Version 8 enables listing offsets by local log start offset.
// This is the earliest log start offset in the local log. (KIP-405).
"validVersions": "0-8",
//
// Version 9 enables listing offsets by last tiered offset (KIP-1005).
"validVersions": "0-9",
"flexibleVersions": "6+",
"fields": [
{ "name": "ThrottleTimeMs", "type": "int32", "versions": "2+", "ignorable": true,
Expand Down
2 changes: 1 addition & 1 deletion core/src/test/java/kafka/test/ClusterTestExtensionsTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -187,7 +187,7 @@ public void testNoAutoStart() {

@ClusterTest
public void testDefaults(ClusterInstance clusterInstance) {
Assertions.assertEquals(MetadataVersion.IBP_4_0_IV0, clusterInstance.config().metadataVersion());
Assertions.assertEquals(MetadataVersion.latestTesting(), clusterInstance.config().metadataVersion());
}

@ClusterTests({
Expand Down
8 changes: 5 additions & 3 deletions core/src/test/scala/integration/kafka/zk/ZkMigrationIntegrationTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -73,7 +73,9 @@ object ZkMigrationIntegrationTest {
MetadataVersion.IBP_3_7_IV2,
MetadataVersion.IBP_3_7_IV4,
MetadataVersion.IBP_3_8_IV0,
MetadataVersion.IBP_4_0_IV0
MetadataVersion.IBP_3_9_IV0,
MetadataVersion.IBP_3_9_IV1
// Note: ZK Migration is not supported in Apache Kafka 4.0 and beyond.
).map { mv =>
val serverProperties = new util.HashMap[String, String]()
serverProperties.put("inter.broker.listener.name", "EXTERNAL")
Expand Down Expand Up @@ -492,7 +494,7 @@ class ZkMigrationIntegrationTest {
}
}

@ClusterTest(types = Array(Type.ZK), brokers = 3, metadataVersion = MetadataVersion.IBP_3_8_IV0, serverProperties = Array(
@ClusterTest(types = Array(Type.ZK), brokers = 3, metadataVersion = MetadataVersion.IBP_3_9_IV1, serverProperties = Array(
new ClusterConfigProperty(key = "inter.broker.listener.name", value = "EXTERNAL"),
new ClusterConfigProperty(key = "listeners", value = "PLAINTEXT://localhost:0,EXTERNAL://localhost:0"),
new ClusterConfigProperty(key = "advertised.listeners", value = "PLAINTEXT://localhost:0,EXTERNAL://localhost:0"),
Expand All @@ -514,7 +516,7 @@ class ZkMigrationIntegrationTest {
val clusterId = zkCluster.clusterId()
val kraftCluster = new KafkaClusterTestKit.Builder(
new TestKitNodes.Builder().
setBootstrapMetadataVersion(MetadataVersion.IBP_3_8_IV0).
setBootstrapMetadataVersion(MetadataVersion.IBP_3_9_IV1).
setClusterId(Uuid.fromString(clusterId)).
setNumBrokerNodes(0).
setNumControllerNodes(1).build())
Expand Down
9 changes: 5 additions & 4 deletions core/src/test/scala/unit/kafka/server/ApiVersionsRequestTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -47,7 +47,7 @@ object ApiVersionsRequestTest {
List(ClusterConfig.defaultBuilder()
.setTypes(java.util.Collections.singleton(Type.ZK))
.setServerProperties(serverProperties)
.setMetadataVersion(MetadataVersion.IBP_4_0_IV0)
.setMetadataVersion(MetadataVersion.latestTesting())
Comment thread
cmccabe marked this conversation as resolved.
.build()).asJava
}

Expand All @@ -67,7 +67,7 @@ object ApiVersionsRequestTest {
serverProperties.put("unstable.feature.versions.enable", "false")
List(ClusterConfig.defaultBuilder()
.setTypes(java.util.Collections.singleton(Type.ZK))
.setMetadataVersion(MetadataVersion.IBP_3_7_IV4)
.setMetadataVersion(MetadataVersion.latestProduction())
.build()).asJava
}

Expand All @@ -83,7 +83,7 @@ object ApiVersionsRequestTest {
class ApiVersionsRequestTest(cluster: ClusterInstance) extends AbstractApiVersionsRequestTest(cluster) {

@ClusterTemplate("testApiVersionsRequestTemplate")
@ClusterTest(types = Array(Type.KRAFT, Type.CO_KRAFT), metadataVersion = MetadataVersion.IBP_4_0_IV0, serverProperties = Array(
@ClusterTest(types = Array(Type.KRAFT, Type.CO_KRAFT), serverProperties = Array(
new ClusterConfigProperty(key = "unstable.api.versions.enable", value = "false"),
new ClusterConfigProperty(key = "unstable.feature.versions.enable", value = "true")
))
Expand Down Expand Up @@ -131,8 +131,9 @@ class ApiVersionsRequestTest(cluster: ClusterInstance) extends AbstractApiVersio
assertEquals(ApiKeys.API_VERSIONS.latestVersion(), apiVersion.maxVersion())
}

// Use the latest production MV for this test
@ClusterTemplate("testApiVersionsRequestValidationV0Template")
@ClusterTest(types = Array(Type.KRAFT, Type.CO_KRAFT), metadataVersion = MetadataVersion.IBP_3_7_IV4, serverProperties = Array(
@ClusterTest(types = Array(Type.KRAFT, Type.CO_KRAFT), metadataVersion = MetadataVersion.IBP_3_8_IV0, serverProperties = Array(

@junrao junrao Jun 24, 2024

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

Could we add a comment that we should use latest production MV for this test?

@cmccabe cmccabe Jun 24, 2024

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

ok

new ClusterConfigProperty(key = "unstable.api.versions.enable", value = "false"),
new ClusterConfigProperty(key = "unstable.feature.versions.enable", value = "false"),
))
Expand Down
11 changes: 10 additions & 1 deletion core/src/test/scala/unit/kafka/server/ListOffsetsRequestTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,18 +22,27 @@ import org.apache.kafka.common.message.ListOffsetsResponseData.ListOffsetsPartit
import org.apache.kafka.common.protocol.{ApiKeys, Errors}
import org.apache.kafka.common.requests.{ListOffsetsRequest, ListOffsetsResponse}
import org.apache.kafka.common.{IsolationLevel, TopicPartition}
import org.apache.kafka.server.config.ServerConfigs
import org.junit.jupiter.api.Assertions._
import org.junit.jupiter.params.ParameterizedTest
import org.junit.jupiter.params.provider.ValueSource

import java.util.Optional
import java.util.{Optional, Properties}
import scala.collection.Seq
Comment thread
cmccabe marked this conversation as resolved.
import scala.jdk.CollectionConverters._

class ListOffsetsRequestTest extends BaseRequestTest {

val topic = "topic"
val partition = new TopicPartition(topic, 0)

override def modifyConfigs(props: Seq[Properties]): Unit = {
super.modifyConfigs(props)
props.foreach { p =>
p.put(ServerConfigs.UNSTABLE_API_VERSIONS_ENABLE_CONFIG, "true")
}
}

@ParameterizedTest
@ValueSource(strings = Array("zk", "kraft"))
def testListOffsetsErrorCodes(quorum: String): Unit = {
Expand Down
2 changes: 1 addition & 1 deletion core/src/test/scala/unit/kafka/server/ReplicationQuotasTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -294,7 +294,7 @@ class ReplicationQuotasTest extends QuorumTestHarness {
features.add(new BrokerRegistrationRequestData.Feature()
.setName(MetadataVersion.FEATURE_NAME)
.setMinSupportedVersion(MetadataVersion.IBP_3_0_IV1.featureLevel())
.setMaxSupportedVersion(MetadataVersion.IBP_4_0_IV0.featureLevel()))
.setMaxSupportedVersion(MetadataVersion.latestTesting().featureLevel()))
controllerServer.controller.registerBroker(
ControllerRequestContextUtil.ANONYMOUS_CONTEXT,
new BrokerRegistrationRequestData()
Expand Down
14 changes: 7 additions & 7 deletions metadata/src/test/java/org/apache/kafka/controller/PartitionChangeBuilderTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -126,7 +126,7 @@ private static MetadataVersion metadataVersionForPartitionChangeRecordVersion(sh
case (short) 1:
return MetadataVersion.IBP_3_7_IV2;
case (short) 2:
return MetadataVersion.IBP_3_8_IV0;
return MetadataVersion.IBP_3_9_IV1;
Comment thread
cmccabe marked this conversation as resolved.
default:
throw new RuntimeException("Unknown PartitionChangeRecord version " + version);
}
Expand Down Expand Up @@ -313,7 +313,7 @@ public void testNoLeaderEpochBumpIfNothingChanged(short version) {
* Test that shrinking the ISR doesn't increase the leader epoch in later MVs.
*/
@ParameterizedTest
@ValueSource(strings = {"3.6-IV0", "3.7-IV4"})
@ValueSource(strings = {"3.6-IV0", "3.7-IV2", "3.9-IV1"})
public void testNoLeaderEpochBumpOnIsrShrink(String metadataVersionString) {
MetadataVersion metadataVersion = MetadataVersion.fromVersionString(metadataVersionString);
testTriggerLeaderEpochBumpIfNeeded(
Expand Down Expand Up @@ -342,7 +342,7 @@ public void testLeaderEpochBumpOnIsrShrink(String metadataVersionString) {
* Test that shrinking the ISR does increase the leader epoch in later MVs when ZK migration is on.
*/
@ParameterizedTest
@ValueSource(strings = {"3.6-IV0", "3.7-IV4"})
@ValueSource(strings = {"3.6-IV0", "3.7-IV2", "3.9-IV1"})
public void testLeaderEpochBumpOnIsrShrinkWithZkMigration(String metadataVersionString) {
MetadataVersion metadataVersion = MetadataVersion.fromVersionString(metadataVersionString);
testTriggerLeaderEpochBumpIfNeeded(
Expand All @@ -358,7 +358,7 @@ public void testLeaderEpochBumpOnIsrShrinkWithZkMigration(String metadataVersion
* Test that expanding the ISR doesn't increase the leader epoch.
*/
@ParameterizedTest
@ValueSource(strings = {"3.4-IV0", "3.5-IV2", "3.6-IV0", "3.7-IV4"})
@ValueSource(strings = {"3.4-IV0", "3.5-IV2", "3.6-IV0", "3.7-IV2", "3.9-IV1"})
public void testNoLeaderEpochBumpOnIsrExpansion(String metadataVersionString) {
MetadataVersion metadataVersion = MetadataVersion.fromVersionString(metadataVersionString);
testTriggerLeaderEpochBumpIfNeeded(
Expand All @@ -372,7 +372,7 @@ public void testNoLeaderEpochBumpOnIsrExpansion(String metadataVersionString) {
* Test that expanding the ISR doesn't increase the leader epoch during ZK migration.
*/
@ParameterizedTest
@ValueSource(strings = {"3.4-IV0", "3.5-IV2", "3.6-IV0", "3.7-IV4"})
@ValueSource(strings = {"3.4-IV0", "3.5-IV2", "3.6-IV0", "3.7-IV2", "3.9-IV1"})
public void testNoLeaderEpochBumpOnIsrExpansionDuringMigration(String metadataVersionString) {
MetadataVersion metadataVersion = MetadataVersion.fromVersionString(metadataVersionString);
testTriggerLeaderEpochBumpIfNeeded(
Expand All @@ -389,7 +389,7 @@ public void testNoLeaderEpochBumpOnIsrExpansionDuringMigration(String metadataVe
* always results in a leader epoch increase.
*/
@ParameterizedTest
@ValueSource(strings = {"3.4-IV0", "3.5-IV2", "3.6-IV0", "3.7-IV4"})
@ValueSource(strings = {"3.4-IV0", "3.5-IV2", "3.6-IV0", "3.7-IV2", "3.9-IV1"})
public void testLeaderEpochBumpOnNewReplicaSetDisjoint(String metadataVersionString) {
MetadataVersion metadataVersion = MetadataVersion.fromVersionString(metadataVersionString);
testTriggerLeaderEpochBumpIfNeeded(
Expand All @@ -403,7 +403,7 @@ public void testLeaderEpochBumpOnNewReplicaSetDisjoint(String metadataVersionStr
* cannot actually change the ISR, triggerLeaderEpochBumpForIsrShrinkIfNeeded does not engage.
*/
@ParameterizedTest
@ValueSource(strings = {"3.4-IV0", "3.5-IV2", "3.6-IV0", "3.7-IV4"})
@ValueSource(strings = {"3.4-IV0", "3.5-IV2", "3.6-IV0", "3.7-IV2"})

@junrao junrao Jul 1, 2024

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

@cmccabe : Is it intentional that we leave out "3.9-IV1" in this test?

public void testNoLeaderEpochBumpOnEmptyTargetIsr(String metadataVersionString) {
MetadataVersion metadataVersion = MetadataVersion.fromVersionString(metadataVersionString);
PartitionRegistration partition = new PartitionRegistration.Builder().
Expand Down
16 changes: 8 additions & 8 deletions metadata/src/test/java/org/apache/kafka/controller/QuorumControllerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -197,7 +197,7 @@ public void testConfigurationOperations() throws Throwable {
) {
controlEnv.activeController().registerBroker(ANONYMOUS_CONTEXT,
new BrokerRegistrationRequestData().
setFeatures(brokerFeatures(MetadataVersion.IBP_3_0_IV1, MetadataVersion.IBP_4_0_IV0)).
setFeatures(brokerFeatures(MetadataVersion.IBP_3_0_IV1, MetadataVersion.latestTesting())).
setBrokerId(0).
setLogDirs(Collections.singletonList(Uuid.fromString("iiaQjkRPQcuMULNII0MUeA"))).
setClusterId(logEnv.clusterId())).get();
Expand Down Expand Up @@ -241,7 +241,7 @@ public void testDelayedConfigurationOperations() throws Throwable {
) {
controlEnv.activeController().registerBroker(ANONYMOUS_CONTEXT,
new BrokerRegistrationRequestData().
setFeatures(brokerFeatures(MetadataVersion.IBP_3_0_IV1, MetadataVersion.IBP_4_0_IV0)).
setFeatures(brokerFeatures(MetadataVersion.IBP_3_0_IV1, MetadataVersion.latestTesting())).
setBrokerId(0).
setLogDirs(Collections.singletonList(Uuid.fromString("sTbzRAMnTpahIyIPNjiLhw"))).
setClusterId(logEnv.clusterId())).get();
Expand Down Expand Up @@ -299,7 +299,7 @@ public void testFenceMultipleBrokers() throws Throwable {
new BrokerRegistrationRequestData().
setBrokerId(brokerId).
setClusterId(active.clusterId()).
setFeatures(brokerFeatures(MetadataVersion.IBP_3_0_IV1, MetadataVersion.IBP_4_0_IV0)).
setFeatures(brokerFeatures(MetadataVersion.IBP_3_0_IV1, MetadataVersion.latestTesting())).
setIncarnationId(Uuid.randomUuid()).
setListeners(listeners));
brokerEpochs.put(brokerId, reply.get().epoch());
Expand Down Expand Up @@ -381,7 +381,7 @@ public void testUncleanShutdownBroker() throws Throwable {
}).
setSessionTimeoutMillis(OptionalLong.of(sessionTimeoutMillis)).

setBootstrapMetadata(BootstrapMetadata.fromVersion(MetadataVersion.IBP_3_8_IV0, "test-provided bootstrap ELR enabled")).
setBootstrapMetadata(BootstrapMetadata.fromVersion(MetadataVersion.IBP_3_9_IV1, "test-provided bootstrap ELR enabled")).
build()
) {
ListenerCollection listeners = new ListenerCollection();
Expand All @@ -395,7 +395,7 @@ public void testUncleanShutdownBroker() throws Throwable {
new BrokerRegistrationRequestData().
setBrokerId(brokerId).
setClusterId(active.clusterId()).
setFeatures(brokerFeatures(MetadataVersion.IBP_3_0_IV1, MetadataVersion.IBP_3_8_IV0)).
setFeatures(brokerFeatures(MetadataVersion.IBP_3_0_IV1, MetadataVersion.IBP_3_9_IV1)).
setIncarnationId(Uuid.randomUuid()).
setLogDirs(Collections.singletonList(Uuid.randomUuid())).
setListeners(listeners));
Expand Down Expand Up @@ -464,7 +464,7 @@ public void testUncleanShutdownBroker() throws Throwable {
new BrokerRegistrationRequestData().
setBrokerId(brokerToUncleanShutdown).
setClusterId(active.clusterId()).
setFeatures(brokerFeatures(MetadataVersion.IBP_3_0_IV1, MetadataVersion.IBP_3_8_IV0)).
setFeatures(brokerFeatures(MetadataVersion.IBP_3_0_IV1, MetadataVersion.IBP_3_9_IV1)).
setIncarnationId(Uuid.randomUuid()).
setLogDirs(Collections.singletonList(Uuid.randomUuid())).
setListeners(listeners)).get();
Expand All @@ -477,7 +477,7 @@ public void testUncleanShutdownBroker() throws Throwable {
new BrokerRegistrationRequestData().
setBrokerId(lastKnownElr[0]).
setClusterId(active.clusterId()).
setFeatures(brokerFeatures(MetadataVersion.IBP_3_0_IV1, MetadataVersion.IBP_3_8_IV0)).
setFeatures(brokerFeatures(MetadataVersion.IBP_3_0_IV1, MetadataVersion.IBP_3_9_IV1)).
setIncarnationId(Uuid.randomUuid()).
setLogDirs(Collections.singletonList(Uuid.randomUuid())).
setListeners(listeners)).get();
Expand Down Expand Up @@ -718,7 +718,7 @@ public void testUnregisterBroker() throws Throwable {
setBrokerId(0).
setClusterId(active.clusterId()).
setIncarnationId(Uuid.fromString("kxAT73dKQsitIedpiPtwBA")).
setFeatures(brokerFeatures(MetadataVersion.IBP_3_0_IV1, MetadataVersion.IBP_4_0_IV0)).
setFeatures(brokerFeatures(MetadataVersion.IBP_3_0_IV1, MetadataVersion.latestTesting())).
setLogDirs(Collections.singletonList(Uuid.fromString("vBpaRsZVSaGsQT53wtYGtg"))).
setListeners(listeners));
assertEquals(5L, reply.get().epoch());
Expand Down
4 changes: 2 additions & 2 deletions metadata/src/test/java/org/apache/kafka/metadata/PartitionRegistrationTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -287,7 +287,7 @@ private static Stream<Arguments> metadataVersionsForTestPartitionRegistration()
return Stream.of(
MetadataVersion.IBP_3_7_IV1,
MetadataVersion.IBP_3_7_IV2,
MetadataVersion.IBP_3_8_IV0
MetadataVersion.IBP_3_9_IV1
).map(mv -> Arguments.of(mv));
}

Expand Down Expand Up @@ -373,7 +373,7 @@ public void testPartitionRegistrationToRecord_ElrShouldBeNullIfEmpty() {
setPartitionEpoch(0);
List<UnwritableMetadataException> exceptions = new ArrayList<>();
ImageWriterOptions options = new ImageWriterOptions.Builder().
setMetadataVersion(MetadataVersion.IBP_3_8_IV0).
setMetadataVersion(MetadataVersion.IBP_3_9_IV1).
setLossHandler(exceptions::add).
build();
assertEquals(new ApiMessageAndVersion(expectRecord, (short) 2), partitionRegistration.toRecord(topicID, 0, options));
Expand Down
24 changes: 19 additions & 5 deletions server-common/src/main/java/org/apache/kafka/server/common/MetadataVersion.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -202,11 +202,23 @@ public enum MetadataVersion {
// Add new fetch request version for KIP-951
IBP_3_7_IV4(19, "3.7", "IV4", false),

// New version for the Kafka 3.8.0 release.
IBP_3_8_IV0(20, "3.8", "IV0", false),

//
// NOTE: MetadataVersions after this point are unstable and may be changed.
// If users attempt to use an unstable MetadataVersion, they will get an error.
// Please move this comment when updating the LATEST_PRODUCTION constant.
//

// Support ListOffsetRequest v9 for KIP-1005.
IBP_3_9_IV0(21, "3.9", "IV0", false),

// Add ELR related supports (KIP-966).
Comment thread
cmccabe marked this conversation as resolved.
IBP_3_8_IV0(20, "3.8", "IV0", true),
IBP_3_9_IV1(22, "3.9", "IV1", true),

// Introduce version 1 of the GroupVersion feature (KIP-848).
IBP_4_0_IV0(21, "4.0", "IV0", false);
IBP_4_0_IV0(23, "4.0", "IV0", false);

// NOTES when adding a new version:
// Update the default version in @ClusterTest annotation to point to the latest version
Expand All @@ -232,7 +244,7 @@ public enum MetadataVersion {
* <strong>Think carefully before you update this value. ONCE A METADATA VERSION IS PRODUCTION,
* IT CANNOT BE CHANGED.</strong>
*/
public static final MetadataVersion LATEST_PRODUCTION = IBP_3_7_IV4;
public static final MetadataVersion LATEST_PRODUCTION = IBP_3_8_IV0;
Comment thread
cmccabe marked this conversation as resolved.

/**
* An array containing all of the MetadataVersion entries.
Expand Down Expand Up @@ -331,7 +343,7 @@ public boolean isDirectoryAssignmentSupported() {
}

public boolean isElrSupported() {
return this.isAtLeast(IBP_3_8_IV0);
return this.isAtLeast(IBP_3_9_IV1);
}

public boolean isKRaftSupported() {
Expand Down Expand Up @@ -459,7 +471,9 @@ public short offsetForLeaderEpochRequestVersion() {
}

public short listOffsetRequestVersion() {
if (this.isAtLeast(IBP_3_5_IV0)) {
if (this.isAtLeast(IBP_3_9_IV0)) {

@junrao junrao Jun 25, 2024 •
edited
Loading

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

  1. The following test failed because of this. I guess we need to use IBP_3_9_IV0 in the test.
kafka.server.ReplicaFetcherThreadTest.shouldSendLatestRequestVersionsByDefault()

org.opentest4j.AssertionFailedError: expected: <9> but was: <8>
	at app//org.junit.jupiter.api.AssertionFailureBuilder.build(AssertionFailureBuilder.java:151)
	at app//org.junit.jupiter.api.AssertionFailureBuilder.buildAndThrow(AssertionFailureBuilder.java:132)
	at app//org.junit.jupiter.api.AssertEquals.failNotEqual(AssertEquals.java:197)
	at app//org.junit.jupiter.api.AssertEquals.assertEquals(AssertEquals.java:134)
	at app//org.junit.jupiter.api.AssertEquals.assertEquals(AssertEquals.java:129)
	at app//org.junit.jupiter.api.Assertions.assertEquals(Assertions.java:329)
	at app//kafka.server.ReplicaFetcherThreadTest.shouldSendLatestRequestVersionsByDefault(ReplicaFetcherThreadTest.scala:132)
  1. The following test failed because of this. I guess we need to enable unstable.feature.versions.enable.
org.opentest4j.AssertionFailedError: Expected follower to discover new log start offset 3
	at app//org.junit.jupiter.api.AssertionUtils.fail(AssertionUtils.java:38)
	at app//org.junit.jupiter.api.Assertions.fail(Assertions.java:138)
	at app//kafka.api.PlaintextAdminIntegrationTest.waitForFollowerLog$1(PlaintextAdminIntegrationTest.scala:846)
	at app//kafka.api.PlaintextAdminIntegrationTest.testReplicaCanFetchFromLogStartOffsetAfterDeleteRecords(PlaintextAdminIntegrationTest.scala:868)

org.apache.kafka.common.errors.InvalidRequestException: Received request api key LIST_OFFSETS with version 9 which is not enabled
  1. org.apache.kafka.tiered.storage.integration.ReassignReplicaExpandTest.executeTieredStorageTest seems to fail consistently now for similar reasons.

org.apache.kafka.common.errors.InvalidRequestException: Received request api key LIST_OFFSETS with version 9 which is not enabled

@cmccabe cmccabe Jun 25, 2024

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

fixed

@junrao junrao Jun 25, 2024

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

It seems that you reverted the fix for ReplicaFetcherThreadTest during merging. Also, PlaintextAdminIntegrationTest.testReplicaCanFetchFromLogStartOffsetAfterDeleteRecords still fails.

@cmccabe cmccabe Jun 26, 2024

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

Both of these tests should pass now.

return 9;
Comment thread
cmccabe marked this conversation as resolved.
} else if (this.isAtLeast(IBP_3_5_IV0)) {
return 8;
} else if (this.isAtLeast(IBP_3_0_IV1)) {
return 7;
Expand Down
4 changes: 2 additions & 2 deletions server-common/src/main/java/org/apache/kafka/server/common/TestFeatureVersion.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,8 +23,8 @@ public enum TestFeatureVersion implements FeatureVersion {

// TEST_1 released right before MV 3.7-IVO was released, and it has no dependencies
TEST_1(1, MetadataVersion.IBP_3_7_IV0, Collections.emptyMap()),
// TEST_2 released right before MV 3.8-IVO was released, and it depends on this metadata version
TEST_2(2, MetadataVersion.IBP_3_8_IV0, Collections.singletonMap(MetadataVersion.FEATURE_NAME, MetadataVersion.IBP_3_8_IV0.featureLevel()));
// TEST_2 released right before MV 3.9-IVO was released, and it depends on this metadata version
TEST_2(2, MetadataVersion.IBP_3_9_IV0, Collections.singletonMap(MetadataVersion.FEATURE_NAME, MetadataVersion.IBP_3_9_IV0.featureLevel()));

private final short featureLevel;
private final MetadataVersion metadataVersionMapping;
Expand Down
2 changes: 1 addition & 1 deletion server-common/src/test/java/org/apache/kafka/server/common/FeaturesTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -110,7 +110,7 @@ public void testLatestProductionMapsToLatestMetadataVersion(Features features) {
@EnumSource(MetadataVersion.class)
public void testDefaultTestVersion(MetadataVersion metadataVersion) {
short expectedVersion;
if (!metadataVersion.isLessThan(MetadataVersion.IBP_3_8_IV0)) {
if (!metadataVersion.isLessThan(MetadataVersion.IBP_3_9_IV0)) {
Comment thread
cmccabe marked this conversation as resolved.
expectedVersion = 2;
} else if (!metadataVersion.isLessThan(MetadataVersion.IBP_3_7_IV0)) {
expectedVersion = 1;
Expand Down
Loading