KAFKA-13093: Log compaction should write new segments with record version v2 (KIP-724) by ijuma · Pull Request #18321 · apache/kafka

ijuma · 2024-12-26T16:12:51Z

Convert v0/v1 record batches to v2 during compaction even if said record batches would be
written with no change otherwise. A few important details:

V0 compressed record batch with multiple records is converted into single V2 record batch
V0 uncompressed records are converted into single record V2 record batches
V0 records are converted to V2 records with timestampType set to CreateTime and the
timestamp is -1.
The KAFKA-4298 workaround is no longer needed since the conversion to V2 fixes
the issue too.
Removed a log warning applicable to consumers older than 0.10.1 - they are no longer
supported.
Added back the ability to append records with v0/v1 (for testing only).
The creation of the leader epoch cache is no longer optional since the record version
config is effectively always V2.

Add integration tests, these tests existed before #18267 - restored, modified and
extended them.

Committer Checklist (excluded from commit message)

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

ijuma · 2024-12-28T16:10:43Z

  }

+  @Test
+  def testLeaderEpochCacheCreatedAfterMessageFormatUpgrade(): Unit = {


This test was originally removed via #18267 - added it back and modified it to take into account that the leader epoch cache always exists now (since the configured record version is always V2).

…sion v2 (KIP-724)

… longer supported

…e it always creates it now

…'s still a useful test

ijuma · 2024-12-29T17:39:57Z

  def listProducerSnapshotOffsets(logDir: File): Seq[Long] =
    ProducerStateManager.listSnapshotFiles(logDir).asScala.map(_.offset).sorted.toSeq

-  def assertLeaderEpochCacheEmpty(log: UnifiedLog): Unit = {


ijuma · 2024-12-29T17:42:08Z


+  @ParameterizedTest
+  @ArgumentsSource(classOf[LogCleanerParameterizedIntegrationTest.ExcludeZstd])
+  def testCleanerWithMessageFormatV0V1V2(compressionType: CompressionType): Unit = {


The tests added in this class were originally removed via #18267 - added them back, modified them to fit the new reality and extended them to include more extensive verification of the log after compaction.

ijuma · 2024-12-29T17:47:25Z

@junrao Would you be able to review this PR? I would like to include it in Apache Kafka 4.0, if possible - it completes KIP-724.

junrao

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

…og-compaction-write-record-v2 * apache-github/trunk: (34 commits) MINOR: Bump year to 2025 in NOTICE file (apache#18427) KAFKA-18411 Remove ZkProducerIdManager (apache#18413) KAFKA-18408 tweak the 'tag' field for BrokerHeartbeatRequest.json, BrokerRegistrationChangeRecord.json and RegisterBrokerRecord.json (apache#18421) KAFKA-18414 Remove KRaftRegistrationResult (apache#18401) KAFKA-17921 Support SASL_PLAINTEXT protocol with java.security.auth.login.config (apache#17671) KAFKA-18384 Remove ZkAlterPartitionManager (apache#18364) KAFKA-10790: Add deadlock detection to producer#flush (apache#17946) KAFKA-18412: Remove EmbeddedZookeeper (apache#18399) MINOR : Improve Exception log in NotEnoughReplicasException(apache#12394) MINOR: Improve PlaintextAdminIntegrationTest#testConsumerGroups (apache#18409) MINOR: Remove unused local variable (apache#18410) MINOR: Remove RaftManager.maybeDeleteMetadataLogDir and AutoTopicCreationManagerTest.scala (apache#17365) KAFKA-18368 Remove TestUtils#MockZkConnect and remove zkConnect from TestUtils#createBrokerConfig (apache#18352) MINOR: Update Consumer group timeout default to 30 sec (apache#16406) MINOR: Fix typo in CommitRequestManager (apache#18407) MINOR: cleanup JavaDocs for deprecation warnings (apache#18402) KAFKA-18303; Update ShareCoordinator to use new record format (apache#18396) MINOR: Update Consumer and Producer JavaDocs for committing offsets (apache#18336) KAFKA-16446: Improve controller event duration logging (apache#15622) KAFKA-18388 test-kraft-server-start.sh should use log4j2.yaml (apache#18370) ...

…to LeaderEpochFileCache

ijuma · 2025-01-08T04:55:25Z

  }

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


@junrao do you agree that we don't need this test anymore? Or is there a purpose that is not made clear by the test name?

This test is still useful. We can just leave partition.futureLog.get.leaderEpochCache unchanged. It will be initialized as empty and the fetcher thread will default to initializing the fetch offset with HWM. We probably want to adjust the test name accordingly.

ijuma · 2025-01-08T04:58:49Z


        if (logOptional.isPresent()) {
-            Option<LeaderEpochFileCache> leaderEpochCache = logOptional.get().leaderEpochCache();
-            if (leaderEpochCache != null && leaderEpochCache.isDefined()) {


@junrao It's odd that we were checking for null here - do you know if we can remove this?

Yes, we can remove this check.

@OverRide

Also added @OverRide annotations to catch it at compile time in the future.

ijuma · 2025-01-08T15:09:28Z

-        leaderEpochCache.foreach(_.clear())
+        // `renameDir` with `shouldReinitialize=false` sets this to `null` and it's usually (but not always) called before this method
+        if (leaderEpochCache != null)
+          leaderEpochCache.clear()


@junrao This is interesting - do you think we should be setting the cache to null here too versus calling clear on it? It's a bit odd that we take different actions to achieve a similar outcome (clearing the in-memory state) in two different (but related) methods.

The case where we don't call renameDir before delete is during LogManager.loadLog.

if (logDir.getName.endsWith(UnifiedLog.DeleteDirSuffix)) { addLogToBeDeleted(log) }

Yes, I agree that we can just call clear on leaderEpochCache instead of setting it to null in renameDir.

junrao

@ijuma : Thanks for the updated PR. A few more minor comments.

junrao · 2025-01-08T17:53:02Z

  private def initializeLeaderEpochCache(): Unit = lock synchronized {
-    leaderEpochCache = UnifiedLog.maybeCreateLeaderEpochCache(
-      dir, topicPartition, logDirFailureChannel, logIdent, leaderEpochCache, scheduler)
+    leaderEpochCache = UnifiedLog.createLeaderEpochCache(


This is an existing issue. initializeLeaderEpochCache is better named as reinitializeLeaderEpochCache.

junrao · 2025-01-08T17:58:26Z

-        leaderEpochCache.foreach(_.clear())
+        // `renameDir` with `shouldReinitialize=false` sets this to `null` and it's usually (but not always) called before this method
+        if (leaderEpochCache != null)
+          leaderEpochCache.clear()


Yes, I agree that we can just call clear on leaderEpochCache instead of setting it to null in renameDir.

junrao · 2025-01-08T18:02:34Z


        if (logOptional.isPresent()) {
-            Option<LeaderEpochFileCache> leaderEpochCache = logOptional.get().leaderEpochCache();
-            if (leaderEpochCache != null && leaderEpochCache.isDefined()) {


Yes, we can remove this check.

junrao · 2025-01-08T18:53:34Z

  }

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


This test is still useful. We can just leave partition.futureLog.get.leaderEpochCache unchanged. It will be initialized as empty and the fetcher thread will default to initializing the fetch offset with HWM. We probably want to adjust the test name accordingly.

ijuma · 2025-01-09T15:31:31Z

@junrao Addressed your comments and the tests are passing.

ijuma · 2025-01-09T15:34:22Z

    assertTrue(log.partitionMetadataFile.isEmpty)
    assertEquals(0, log.logEndOffset)
-    // verify that records appending can still succeed
-    // even with the uninitialized leaderEpochCache and partitionMetadataFile


We don't support appending with the uninitialized leaderEpochCache anymore - it was questionable to support this after renameDir(reinitialize=false) since the former is invoked as part of the deletion of the log.

junrao

@ijuma : Thanks for the updated PR. LGTM.

Should we also remove the following methods in MetadataVersion?
highestSupportedRecordVersion
isOffsetForLeaderEpochSupported
isSaslInterBrokerHandshakeRequestEnabled

ijuma · 2025-01-09T16:51:01Z

@junrao Yes, good idea. Is it ok if I do that as a follow-up? It doesn't have to go into 4.0 and there are many references.

junrao

@ijuma : We can do the followup in a separate PR. This PR LGTM.

…sion v2 (KIP-724) (#18321) Convert v0/v1 record batches to v2 during compaction even if said record batches would be written with no change otherwise. A few important details: 1. V0 compressed record batch with multiple records is converted into single V2 record batch 2. V0 uncompressed records are converted into single record V2 record batches 3. V0 records are converted to V2 records with timestampType set to `CreateTime` and the timestamp is `-1`. 4. The `KAFKA-4298` workaround is no longer needed since the conversion to V2 fixes the issue too. 5. Removed a log warning applicable to consumers older than 0.10.1 - they are no longer supported. 6. Added back the ability to append records with v0/v1 (for testing only). 7. The creation of the leader epoch cache is no longer optional since the record version config is effectively always V2. Add integration tests, these tests existed before #18267 - restored, modified and extended them. Reviewers: Jun Rao <jun@confluent.io>

ijuma · 2025-01-10T01:11:40Z

@junrao Submitted the follow-up PR: #18468

…sion v2 (KIP-724) (apache#18321) Convert v0/v1 record batches to v2 during compaction even if said record batches would be written with no change otherwise. A few important details: 1. V0 compressed record batch with multiple records is converted into single V2 record batch 2. V0 uncompressed records are converted into single record V2 record batches 3. V0 records are converted to V2 records with timestampType set to `CreateTime` and the timestamp is `-1`. 4. The `KAFKA-4298` workaround is no longer needed since the conversion to V2 fixes the issue too. 5. Removed a log warning applicable to consumers older than 0.10.1 - they are no longer supported. 6. Added back the ability to append records with v0/v1 (for testing only). 7. The creation of the leader epoch cache is no longer optional since the record version config is effectively always V2. Add integration tests, these tests existed before apache#18267 - restored, modified and extended them. Reviewers: Jun Rao <jun@confluent.io>

github-actions Bot added triage PRs from the community core Kafka Broker consumer clients labels Dec 26, 2024

ijuma removed the triage PRs from the community label Dec 26, 2024

ijuma force-pushed the kafka-13093-log-compaction-write-record-v2 branch from f38e108 to d41a6e2 Compare December 27, 2024 18:50

github-actions Bot added the storage Pull requests that target the storage module label Dec 27, 2024

ijuma commented Dec 28, 2024

View reviewed changes

ijuma requested a review from junrao December 29, 2024 17:23

ijuma added 10 commits December 29, 2024 09:33

KAFKA-13093: Log compaction should write new segments with record ver…

e9c0e38

…sion v2 (KIP-724)

Remove warning related to consumers earlier than 0.10.1 - they are no…

8256a06

… longer supported

UnifiedLog.maybeCreateLeaderEpochCache -> createLeaderEpochCache sinc…

90c86db

…e it always creates it now

Add back testLeaderEpochCacheCreatedAfterMessageFormatUpgrade as it…

bb28607

…'s still a useful test

Remove unused parameters for LogCleaner.filterInfo

63b10f7

Remove unnecessary version check in LogCleaner.filterInfo

021d86d

Adjust the conversion of V0 to V2 during compaction

b92c3d6

Add tests

58e1547

Remove unused imports

593dc83

Remove unused method parameter

a50bd5d

ijuma commented Dec 29, 2024

View reviewed changes

ijuma force-pushed the kafka-13093-log-compaction-write-record-v2 branch from 0ede54e to a50bd5d Compare December 29, 2024 17:43

Merge branch 'trunk' into kafka-13093-log-compaction-write-record-v2

8ad3400

junrao reviewed Jan 7, 2025

View reviewed changes

ijuma added 2 commits January 7, 2025 19:38

Change UnifiedLog.leaderEpochCache from Option[LeaderEpochFileCache] …

ed6e240

…to LeaderEpochFileCache

github-actions Bot added the tiered-storage Related to the Tiered Storage feature label Jan 8, 2025

ijuma commented Jan 8, 2025

View reviewed changes

ijuma added 2 commits January 8, 2025 06:51

Fix issue where method was not overriden due to change in signature

b649179

Also added @OverRide annotations to catch it at compile time in the future.

Fix NPE in UnifiedLog.delete

7628099

ijuma commented Jan 8, 2025

View reviewed changes

Adjust assert to address review comment

56a93df

junrao reviewed Jan 8, 2025

View reviewed changes

ijuma added 4 commits January 8, 2025 11:42

leaderEpochCache is never null

fb0f06f

Add back test with adjustments (including name)

ef3ec42

Fix test

525d8df

Rename initializeLeaderEpochCache to reinitializeLeaderEpochCache

97448ce

ijuma commented Jan 9, 2025

View reviewed changes

junrao reviewed Jan 9, 2025

View reviewed changes

junrao approved these changes Jan 9, 2025

View reviewed changes

ijuma merged commit cf7029c into apache:trunk Jan 9, 2025

ijuma deleted the kafka-13093-log-compaction-write-record-v2 branch January 9, 2025 17:37

ijuma mentioned this pull request Jan 9, 2025

KAFKA-18180: Move OffsetResultHolder to storage module #18100

Merged

3 tasks

TaiJuWu mentioned this pull request Jan 10, 2025

HOTFIX: fix failed case RemoteLogManagerTest#testStopPartitionsWithDeletion #18474

Merged

3 tasks

Conversation

ijuma commented Dec 26, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Committer Checklist (excluded from commit message)

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ijuma commented Dec 29, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

junrao left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ijuma Jan 8, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

junrao left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

ijuma commented Jan 9, 2025

Uh oh!

Choose a reason for hiding this comment

Uh oh!

junrao left a comment

Choose a reason for hiding this comment

Uh oh!

ijuma commented Jan 9, 2025

Uh oh!

junrao left a comment

Choose a reason for hiding this comment

Uh oh!

ijuma commented Jan 10, 2025

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

ijuma commented Dec 26, 2024 •

edited

Loading

ijuma commented Dec 29, 2024 •

edited

Loading

ijuma Jan 8, 2025 •

edited

Loading