KAFKA-18465: Remove MetadataVersions older than 3.0-IV1 #18468

ijuma · 2025-01-10T01:10:53Z

Apache Kafka 4.0 will only support KRaft and 3.0-IV1 is the minimum version supported by KRaft. So, we can assume that Apache Kafka 4.0 will only communicate with brokers that are 3.0-IV1 or newer.

Note that KRaft was only marked as production-ready in 3.3, so we could go further and set the baseline to 3.3. I think we should have that discussion, but it made sense to start with the non controversial parts.

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

chia7712

@ijuma nice cleanup!

chia7712 · 2025-01-10T02:36:43Z

clients/src/test/java/org/apache/kafka/common/requests/OffsetsForLeaderEpochRequestTest.java

-            if (version < 3)
-                assertEquals(OffsetsForLeaderEpochRequest.DEBUGGING_REPLICA_ID, parsed.replicaId());
-            else
-                assertEquals(replicaId, parsed.replicaId());


Should we keep the test for version=4 ?

The value of this test is a bit low, but I added it back.

chia7712 · 2025-01-10T02:49:15Z

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorShard.java

@@ -731,7 +731,6 @@ private void cancelClassicGroupSizeCounter() {
    public void onLoaded(MetadataImage newImage) {
        MetadataDelta emptyDelta = new MetadataDelta(newImage);
        groupMetadataManager.onNewMetadataImage(newImage, emptyDelta);
-        offsetMetadataManager.onNewMetadataImage(newImage, emptyDelta);


I guess we don't need to clean up GroupMetadataManager.scala, as 4.0 uses a new coordinator by default?

I updated that class too where it made sense. Did you spot additional areas of clean-up?

chia7712 · 2025-01-10T03:23:03Z

...inator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTestContext.java

@@ -238,7 +238,7 @@ public static CoordinatorRecord newGroupMetadataRecord(
            ),
            new ApiMessageAndVersion(
                value,
-                metadataVersion.groupMetadataValueVersion()


Could you please remove variable MetadataVersion metadataVersion as it is unused now?

chia7712 · 2025-01-10T03:37:10Z

...rdinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorRecordHelpers.java

    ) {
-        short version = metadataVersion.offsetCommitValueVersion(offsetAndMetadata.expireTimestampMs.isPresent());


It seems we need to use version=1 in order to keep "expireTimestamp". otherwise, in v3 expireTimestamp will be ignored in serialization.

Yeah, this change is wrong - a few things will be reverted as a result.

Fixed this one - still working through a few other things.

ijuma · 2025-01-10T04:10:50Z

Thanks for the quick review @chia7712. I should have left a note, but I'm still working on this PR. I'll take a look at your comments soon.

Apache Kafka 4.0 will only support KRaft and 3.0 was the first release where KRaft was in a reasonable state (even if not production ready yet). So, we can assume that Apache Kafka 4.0 will only communicate with brokers that are 3.0 or newer. A couple of notes: 1. KRaft was only marked as production-ready in 3.3, so we could go further and set the baseline to 3.3. I think we should have that discussion, but it made sense to start with the non controversial parts. 2. I have not removed the actual IBP definitions - leaving that for a separate PR.

ijuma · 2025-01-10T08:22:33Z

I removed the relevant IBP constants too and fixed the known test failures. I'll check the full test results tomorrow and will do a file by file pass then too.

dajac · 2025-01-10T13:02:05Z

core/src/main/scala/kafka/coordinator/group/GroupMetadataManager.scala

-      // until a tagged field is introduced or the version is bumped.
-      else 3.toShort
+      if (offsetAndMetadata.expireTimestampMs.isPresent) 1.toShort
+      else maxVersion


Is there a need to pass it as an argument to the method or could we just hardcode it?

This is needed for some tests that verify the ability to read messages encoded with the old format. I'm not sure if there's a way to make sure there are no messages written with the old format by the time the broker is upgraded to 4.0. Given the usual retention rules, it should be fine, but not sure if we can be sure. Thoughts?

dajac · 2025-01-10T13:02:19Z

core/src/main/scala/kafka/coordinator/group/GroupMetadataManager.scala

-      // until a tagged field is introduced or the version is bumped.
-      else 3.toShort
-
+                         version: Short = 3): Array[Byte] = {


See the reply to the other comment.

ijuma · 2025-01-10T13:28:50Z

There was a single failing test - pushed a simple fix.

ijuma · 2025-01-10T15:01:04Z

core/src/test/scala/unit/kafka/coordinator/group/GroupMetadataManagerTest.scala

-  }
-
-  @Test
-  def testSerdeOffsetCommitValueWithNoneExpireTimestamp(): Unit = {


This basically duplicates testSerdeOffsetCommitValue and hence I removed it.

…emove-metadata-version-methods-for-versions-older-than-3.0 * apache-github/trunk: KAFKA-18340: Change Dockerfile to use log4j2 yaml instead log4j properties (apache#18378) MINOR: fix flaky RemoteLogManagerTest#testStopPartitionsWithDeletion (apache#18474) KAFKA-18311: Enforcing copartitioned topics (4/N) (apache#18397) KAFKA-18308; Update CoordinatorSerde (apache#18455) KAFKA-18440: Convert AuthorizationException to fatal error in AdminClient (apache#18435) KAFKA-17671: Create better documentation for transactions (apache#17454) KAFKA-18304; Introduce json converter generator (apache#18458) MINOR: Clean up classic group tests (apache#18473) KAFKA-18399 Remove ZooKeeper from KafkaApis (2/N): CONTROLLED_SHUTDOWN and ENVELOPE (apache#18422) MINOR: improve StreamThread periodic processing log (apache#18430)

ijuma · 2025-01-10T15:02:36Z

core/src/test/scala/unit/kafka/server/AbstractFetcherThreadTest.scala

@@ -390,43 +390,6 @@ class AbstractFetcherThreadTest {
    assertEquals(leaderState.highWatermark, replicaState.highWatermark)
  }

-  @Test
-  def testTruncateToHighWatermarkIfLeaderEpochRequestNotSupported(): Unit = {


No longer relevant since leader epoch request is now always supported.

ijuma · 2025-01-10T15:03:02Z

core/src/test/scala/unit/kafka/server/AlterPartitionManagerTest.scala

@@ -486,74 +473,6 @@ class AlterPartitionManagerTest {
    assertFutureThrows(future2, classOf[UnknownServerException])
  }

-  @ParameterizedTest
-  @MethodSource(Array("provideMetadataVersions"))
-  def testPartialTopicIds(metadataVersion: MetadataVersion): Unit = {


We now always have topic ids.

ijuma · 2025-01-10T15:04:03Z

core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala

@@ -1640,15 +1639,6 @@ class KafkaConfigTest {
    }
  }

-  @Test
-  def testInvalidInterBrokerProtocolVersionKRaft(): Unit = {


The baseline version now supports kraft.

ijuma · 2025-01-10T15:08:07Z

core/src/test/scala/unit/kafka/server/ReplicaFetcherThreadTest.scala

    assertEquals(
      ApiKeys.LIST_OFFSETS.latestVersion(true),
      testingVersion.listOffsetRequestVersion
    )
  }

-  @Disabled("KAFKA-18370")
-  @Test
-  def testFetchLeaderEpochRequestIfLastEpochDefinedForSomePartitions(): Unit = {


Lots of disabled tests that rely on the "no truncate on fetch", which is no longer applicable.

It's probably better to remove those tests related OffsetForLeaderEpochRequest until we remove the code in AbstractFetcherThread.truncateToEpochEndOffsets(). This can be done in a separate jira.

We could remove all tests that depend on old IBPs.

Sorry @junrao - I didn't fully understand. Are you saying we should not remove these tests just yet or that it's ok to remove these now and remove the rest in a separate PR?

KAFKA-18370 says that we could remove these tests because 4.0 doesn't support ZK anymore. This seems incorrect since this part of the code is independent of the ZK change. However, it does seem that the OffsetForLeaderEpochRequest is not really exercised because of the truncating on fetch support. I am just not 100% sure. So, it would useful to try to remove that part of the code first and then remove the corresponding tests.

cc @chia7712

I see. The reason I removed these tests is:

They relied on IBP 2.6-IV0 via the kafkaConfigNoTruncateOnFetch method.

They were disabled.

I was under the impression that the truncation approach from IBP 2.7 onwards does not rely on offset for leader epochs requests: KAFKA-10554; Perform follower truncation based on diverging epochs in Fetch response #9382

So, I wasn't sure how to make the tests work.

Your point is that the legacy code is still there and hence removing the tests is premature. Is that right?

Since these tests were already disabled, is it OK if I keep them deleted and follow up with a separate change that removes the supposed replica fetcher dead code?

Sounds good. I filed https://issues.apache.org/jira/browse/KAFKA-18477 to track the removal work in replica fetcher.

ijuma · 2025-01-10T15:11:33Z

storage/src/test/java/org/apache/kafka/storage/internals/log/LogValidatorTest.java

        ).validateMessagesAndAssignOffsets(
                PrimitiveRef.ofLong(0L), metricsRecorder, RequestLocal.withThreadConfinedCaching().bufferSupplier()
        ));

        assertEquals(metricsRecorder.recordInvalidOffsetCount, 1);
    }

-    @Test
-    public void testZStdCompressedWithUnavailableIBPVersion() {


zstd is now always available.

ijuma · 2025-01-10T15:12:04Z

This is ready for review now.

ijuma · 2025-01-10T18:26:03Z

The tests passed.

junrao

@ijuma : Thanks for the PR. Left a few comments.

junrao · 2025-01-10T17:49:10Z

clients/src/main/java/org/apache/kafka/common/requests/AlterPartitionRequest.java

         */
-        public Builder(AlterPartitionRequestData data, boolean canUseTopicIds) {


v2 request is added in 3.3.0. To be consistent, we need to move the MV baseline to 3.3.

The claim is that we're using protocol negotiation to decide the version to use. The following is stated in AlterPartitionManager.buildRequest.

While building the request, we don't know which version of the AlterPartition API is

supported by the controller. The final decision is taken when the AlterPartitionRequest

is built in the network client based on the advertised api versions of the controller.

And the code does seem to use clientChannelBuilder.

Thanks for the explanation. We can keep the change as it is then.

junrao · 2025-01-10T18:00:04Z

core/src/main/scala/kafka/server/ApiVersionManager.scala

@@ -150,14 +150,12 @@ class DefaultApiVersionManager(
    }
    val apiVersions = if (controllerApiVersions.isDefined) {
      ApiVersionsResponse.controllerApiVersions(
-        finalizedFeatures.metadataVersion().highestSupportedRecordVersion,


Should we leave this param here to support record format change in the future?

I can revert this (and related code) if you think it's useful since it's reasonably self-contained. The benefit of keeping vs removing is probably borderline given that there are no concrete plans for adding new record versions (including the specifics regarding the compatibility story). Still, this is a bit unrelated to the main point of this PR (MV/IBP), so I can change it back.

Yes, I agree that it's borderline. It's probably ok to keep this change for now.

junrao · 2025-01-10T18:31:46Z

core/src/main/scala/kafka/coordinator/group/GroupMetadataManager.scala

@@ -1112,21 +1109,13 @@ object GroupMetadataManager {
   *
   * @param groupMetadata current group metadata
   * @param assignment the assignment for the rebalancing generation
-   * @param metadataVersion the api version
+   * @param version the version to serialize it with, the default is `3`, the highest supported non-flexible version
+   *        until a tagged version is bumped. The default should always be used outside of tests


until a tagged version is bumped => until a tagged field is introduced or the version is bumped

junrao · 2025-01-10T19:12:47Z

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/classic/ClassicGroup.java

@@ -1445,7 +1445,7 @@ public void createClassicGroupRecords(
            assignments.put(classicGroupMember.memberId(), classicGroupMember.assignment())
        );

-        records.add(GroupCoordinatorRecordHelpers.newGroupMetadataRecord(this, assignments, metadataVersion));
+        records.add(GroupCoordinatorRecordHelpers.newGroupMetadataRecord(this, assignments));


metadataVersion is no longer used.

junrao · 2025-01-10T19:38:38Z

core/src/test/scala/unit/kafka/log/LogLoaderTest.scala

@@ -128,7 +127,7 @@ class LogLoaderTest {
        brokerTopicStats = new BrokerTopicStats(),
        logDirFailureChannel = logDirFailureChannel,
        time = time,
-        keepPartitionMetadataFile = config.usesTopicId,
+        keepPartitionMetadataFile = true,


We can probably remove keepPartitionMetadataFile from LogManager. This can be done in a followup PR.

junrao · 2025-01-10T19:48:44Z

core/src/test/scala/unit/kafka/server/ReplicaFetcherThreadTest.scala

    assertEquals(
      ApiKeys.LIST_OFFSETS.latestVersion(true),
      testingVersion.listOffsetRequestVersion
    )
  }

-  @Disabled("KAFKA-18370")
-  @Test
-  def testFetchLeaderEpochRequestIfLastEpochDefinedForSomePartitions(): Unit = {


It's probably better to remove those tests related OffsetForLeaderEpochRequest until we remove the code in AbstractFetcherThread.truncateToEpochEndOffsets(). This can be done in a separate jira.

We could remove all tests that depend on old IBPs.

junrao · 2025-01-10T19:55:30Z

core/src/test/scala/unit/kafka/server/ReplicaFetcherThreadTest.scala

@@ -48,7 +46,7 @@ import org.junit.jupiter.params.ParameterizedTest
 import org.junit.jupiter.params.provider.ValueSource
 import org.mockito.ArgumentCaptor
 import org.mockito.ArgumentMatchers.{any, anyBoolean, anyLong}
-import org.mockito.Mockito.{mock, never, times, verify, when}
+import org.mockito.Mockito.{mock, times, verify, when}


We could delete shouldUseLeaderEndOffsetIfInterBrokerVersionBelow20.

ijuma · 2025-01-11T01:39:41Z

@junrao I have addressed your comments. Let me know if you have any other comments or if this looks good. Thanks!

junrao

@ijuma : Thanks for the updated PR. LGTM.

Filed https://issues.apache.org/jira/browse/KAFKA-18479 to track the followup item.

Apache Kafka 4.0 will only support KRaft and 3.0-IV1 is the minimum version supported by KRaft. So, we can assume that Apache Kafka 4.0 will only communicate with brokers that are 3.0-IV1 or newer. Note that KRaft was only marked as production-ready in 3.3, so we could go further and set the baseline to 3.3. I think we should have that discussion, but it made sense to start with the non controversial parts. Reviewers: Jun Rao <[email protected]>, Chia-Ping Tsai <[email protected]>, David Jacot <[email protected]>

After apache#18468, interBrokerProtocolVersion is no longer used in GroupMetadataManager, we should remove it.

Apache Kafka 4.0 will only support KRaft and 3.0-IV1 is the minimum version supported by KRaft. So, we can assume that Apache Kafka 4.0 will only communicate with brokers that are 3.0-IV1 or newer. Note that KRaft was only marked as production-ready in 3.3, so we could go further and set the baseline to 3.3. I think we should have that discussion, but it made sense to start with the non controversial parts. Reviewers: Jun Rao <[email protected]>, Chia-Ping Tsai <[email protected]>, David Jacot <[email protected]>

ijuma requested a review from junrao January 10, 2025 01:11

github-actions bot added core Kafka Broker kraft clients labels Jan 10, 2025

ijuma mentioned this pull request Jan 10, 2025

KAFKA-13093: Log compaction should write new segments with record version v2 (KIP-724) #18321

Merged

3 tasks

ijuma force-pushed the kafka-18465-remove-metadata-version-methods-for-versions-older-than-3.0 branch from 1097696 to c557e0b Compare January 10, 2025 01:22

chia7712 reviewed Jan 10, 2025

View reviewed changes

ijuma force-pushed the kafka-18465-remove-metadata-version-methods-for-versions-older-than-3.0 branch from c557e0b to 7449859 Compare January 10, 2025 05:24

Remove IBP versions older than 3.0-IV1

5d5ca3a

ijuma changed the title ~~KAFKA-18465: Remove MetadataVersion methods for versions older than 3.0~~ KAFKA-18465: Remove MetadataVersions older than 3.0-IV1 Jan 10, 2025

ijuma added 2 commits January 9, 2025 23:55

Remove no longer relevant test

92206c0

Add. back modified test

05048cf

github-actions bot added tools performance storage Pull requests that target the storage module labels Jan 10, 2025

Remove unused method parameter

d0f8d49

dajac reviewed Jan 10, 2025

View reviewed changes

Fix single test that failed

a55c7cb

Minor tweak to GroupMetadataManager.offsetCommitValue

d6c543d

ijuma commented Jan 10, 2025

View reviewed changes

junrao reviewed Jan 10, 2025

View reviewed changes

ijuma added 2 commits January 10, 2025 16:22

Address some of the review comments

7e67346

Fix comment.

65dd7de

junrao approved these changes Jan 11, 2025

View reviewed changes

ijuma merged commit d4aee71 into apache:trunk Jan 11, 2025
9 checks passed

ijuma deleted the kafka-18465-remove-metadata-version-methods-for-versions-older-than-3.0 branch January 11, 2025 17:42

TaiJuWu mentioned this pull request Jan 16, 2025

KAFKA-18373: Remove ZkMetadataCache #18553

Merged

3 tasks

frankvicky added a commit to frankvicky/kafka that referenced this pull request Jan 19, 2025

MINOR: Remove unused interBrokerProtocolVersion in GroupMetadataManager

ef7978d

After apache#18468, interBrokerProtocolVersion is no longer used in GroupMetadataManager, we should remove it.

frankvicky mentioned this pull request Jan 19, 2025

KAFKA-18589: Cleanup GroupMetadataManager #18619

Merged

3 tasks

		) {
		short version = metadataVersion.offsetCommitValueVersion(offsetAndMetadata.expireTimestampMs.isPresent());

		*/
		public Builder(AlterPartitionRequestData data, boolean canUseTopicIds) {

KAFKA-18465: Remove MetadataVersions older than 3.0-IV1 #18468

KAFKA-18465: Remove MetadataVersions older than 3.0-IV1 #18468

Conversation

ijuma commented Jan 10, 2025 • edited Loading

Committer Checklist (excluded from commit message)

chia7712 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ijuma Jan 10, 2025 • edited Loading

Choose a reason for hiding this comment

ijuma commented Jan 10, 2025

ijuma commented Jan 10, 2025

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ijuma commented Jan 10, 2025

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ijuma Jan 10, 2025 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ijuma commented Jan 10, 2025

ijuma commented Jan 10, 2025

junrao left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ijuma commented Jan 11, 2025

junrao left a comment

Choose a reason for hiding this comment

ijuma commented Jan 10, 2025 •

edited

Loading

ijuma Jan 10, 2025 •

edited

Loading

ijuma Jan 10, 2025 •

edited

Loading