Status

Current state: ["Under Discussion"]

Discussion thread: here
JIRA: KAFKA-20033 - Getting issue details... STATUS

Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).

Motivation

Background: What is __remote_log_metadata and how is it used?

• Tiered storage: When local segments close, brokers upload them to remote storage (S3/Azure/GCS).

• remote_log_metadata: An internal topic used by TopicBasedRemoteLogMetadataManager to record remote-segment lifecycle events per topic-partition (COPY* and DELETE*), including start/end offsets, leader epoch and timestamps.

• Why it matters: Brokers maintain an in-memory RemoteLogMetadataCache to answer “which segment contains offset X?” and “what is the highest remote offset?” This cache drives remote fetches and cleanup.

• How state is built today: On startup/leadership changes, brokers consume the relevant metadata partitions from earliest available offset, replay all events, and rebuild the cache (apply STARTED → FINISHED → optional DELETE states; index by endOffset and leaderEpoch). Because the topic is append-only, full replay is required to compute current state.

Operational challenges

Problem 1: Unbounded Growth

• cleanup.policy=delete with infinite retention (-1)

• All lifecycle events retained forever

• ~1.2M metadata records/year at 1 TB/day upload, topic can grow out of bounds, stressing I/O and page cache

Problem 2: No Cleanup Mechanism

• Metadata persists after remote segments expire

• No removal for segments already deleted in remote storage

• Orphaned records from failed/retried uploads accumulate indefinitely

Problem 3: Slow Bootstrap

• Full replay from offset 0 to rebuild cache

• Bootstrap time grows with topic size; restarts and leadership changes take increasingly long time

This KIP aims to solve this issue through making the current topic __remote_log_metadata compacted.

Goals

1. Enable bounded metadata storage to support faster bootstrap for the broker restart/leadership changes: the number of messages in the remote metadata topic is proportional to active remote segments, not the total history.

2. Support metadata cleanup: remove metadata when segments expire from remote storage.

Non-Goals

1. Wire-level backward compatibility for keys in the compacted projection (breaking change there is acceptable).

2. Online migration with zero operational steps: clusters that have the tiered storage enabled before must go through an automated migration to adopt the compacted projection change, clusters that don’t have the tiered storage enabled can adopt the change directly without any migration.

3. Support for other RLMM implementations is not considered: focus is on TopicBasedRemoteLogMetadataManager.

Proposed Changes

1. Topics and Configurations
   This KIP updates the behavior and usage of the existing internal topic __remote_log_metadata , the current default configuration for this topic is:

   • cleanup.policy=delete

   • retention.ms=-1 
     
     With this change, the topic's configuration can be updated to be compacted:

   • cleanup.policy=compact

   • min.cleanable.dirty.ratio=0.1

   • delete.retention.ms=86400000 // 1 day

   • segment.ms=3600000 // 1 hour

     Once the topic becomes compacted, it will only has bounded number of messages for cache bootstrap and steady-state reads; supports tombstones

2. Keyed Metadata Records in __remote_log_metadata
   All metadata records in __remote_log_metadata become keyed. The value carries the detailed metadata; the key defines the uniqueness and compaction behavior.
   Key Format: TopicIdPartition:TopicName:Partition:EndOffset:BrokerLeaderEpoch

   Example:
   T8fJ9Kz3RyWxP2mQ4nL7vB:topicA:0:1999:5

   Key Components
   - TopicIdPartition: Unique UUID identifying the topic and partition
   - TopicName: Human-readable topic name
   - Partition: Partition number
   - EndOffset: Last offset in the segment (deterministic since tiered storage only uploads closed segments)
   - BrokerLeaderEpoch: Leader epoch of the broker that owned the topic-partition during upload, disambiguating concurrent/retry scenarios

   
   Properties and Behavior
   For Segment Metadata:

   • For the same segment under the same leader epoch, retries or updates share the same key (compaction keeps the latest state).

   • For the same segment under different leader epochs, records use different keys, allowing the system to distinguish concurrent or overlapping uploads across different brokers.

   
   State Transition Keys:
   All segment lifecycle state transitions use the same key with an :UPDATE suffix:
   TopicIdPartition:TopicName:Partition:EndOffset:BrokerLeaderEpoch:UPDATE

   This applies to all state transitions in the segment lifecycle:
   - COPY_SEGMENT_STARTED → COPY_SEGMENT_FINISHED
   - COPY_SEGMENT_FINISHED → DELETE_SEGMENT_STARTED
   - DELETE_SEGMENT_STARTED → DELETE_SEGMENT_FINISHED
   Rationale for :UPDATE Suffix:
   The :UPDATE suffix is necessary to differentiate state transition records from the initial segment metadata record. This design avoids duplicating fields (such as SegmentLeaderEpochs, MaxTimestampMs, SegmentSizeInBytes, etc.) in RemoteLogSegmentMetadataUpdateRecord that already exist in RemoteLogSegmentMetadataRecord. By using distinct keys:
   - The full segment metadata record compacts independently (contains all segment details)
   - State transition records compact independently (contains only state and timestamp)
   - Both can coexist without redundant field duplication

   
   

3. Broker Write Path
   The broker continues to write lifecycle events for remote log segments and partitions, but now with:

   • Deterministic keys.

   • Tombstone semantics once the log segment is expired.

           Lifecycle events

• • Segment upload:

    • COPY_SEGMENT_STARTED

    • COPY_SEGMENT_FINISHED

  • Segment deletion:

    • DELETE_SEGMENT_STARTED

    • DELETE_SEGMENT_FINISHED

  • Partition deletion:

    • DELETE_PARTITION_STARTED

    • DELETE_PARTITION_FINISHED

It will also receive the tombstone messages for those expired metadata messages, expired metadata messages are defined as the messages that belong to the log segments that are already in the DELETE_SEGMENT_FINISHED states. 

A series of tombstone messages will be published for all the keys that share the same prefix topicIdPartition:endOffset  and keys that have no bigger brokerLeaderEpoch than the current one. While the original storeRemoteLogMetadata() method will wait for the consumer of __remote_log_metadata  to finish the consumption of the latest DELETE_SEGMENT_FINISHED message, in order to minimize the impact for the performance, for the tombstone messages, the consumers will finish the consumption asynchronously.

4. Broker Read Path

There is no change in how the ConsumerTask read the messages and build the state in TopicBasedRemoteLogMetadataManager  since there is not much meaningful value format change.

There is no change to RemoteLogMetadataTopicPartitioner  neither.

User Scenario

RemoteLogSegmentState Enum Values

Scenario 1: Normal Segment Upload

Context: Topic orders   (ID:  abc123 )

Partition 0, Broker 101 at epoch 3

Timeline & Messages

After Compaction

Compacted topic: Retains only latest → abc123:topicA:0:1000:3 :UPDATE and abc123:topicA:0:1000:3, both keys will be retained.

Cache result: 1 segment with uuid=UUID-A, endOffset=1000, state=FINISHED

Scenario 2: Leadership Change During Upload

Context: Broker 101 (epoch 3) starts upload, leadership changes to Broker 102 (epoch 4), both complete

Timeline & Messages

After Compaction

Compacted topic: 2 keys, each with latest state

• abc123:topicA:0:2000:3
• abc123:topicA:0:2000:4

• abc123:topicA:0:2000:3:UPDATE → FINISHED (UUID-A, orphaned)

•  abc123:topicA:0:2000:4:UPDATE → FINISHED (UUID-B, active)

Cache result: Returns both segments, selects UUID-B (epoch 4 > 3) for reads

Scenario 3: Failed Upload Under the Same Leader Epoch

Context: Broker 101 (epoch 5) tries upload, fails, retries with new UUID

Timeline & Messages

After Compaction

Compacted topic: Retains only latest → abc123:topicA:0:3000:5:UPDATE  = FINISHED (UUID-B), and abc123:topicA:0:3000:5.

Cache result: Only UUID-B appears (UUID-A compacted away)

Key insight: Same epoch = same key, but new UUID for each attempt

Scenario 4: Segment Deletion

Context: Segment at endOffset=1000 has 3 upload keys (epochs 3,4,5), now being deleted by current leader (epoch 6)

Initial State

abc123:topicA:0:1000:3:UPDATE → FINISHED (UUID-A, orphaned)

abc123:topicA:0:1000:4:UPDATE → FINISHED (UUID-B, orphaned)

abc123:topicA:0:1000:5:UPDATE → FINISHED (UUID-C, active)

Timeline & Messages

Tombstone Details (Compacted Topic ONLY)

After DELETE_FINISHED (T4), tombstones written for ALL 8 keys:

abc123:topicA:0:1000:3 → null (epoch 3, UUID-A)

abc123:topicA:0:1000:3:UPDATE → null (epoch 3, UUID-A)

abc123:topicA:0:1000:4 → null (epoch 4, UUID-B)

abc123:topicA:0:1000:4:UPDATE → null (epoch 4, UUID-B)

abc123:topicA:0:1000:5 → null (epoch 5, UUID-C)

abc123:topicA:0:1000:5:UPDATE → null (epoch 5, UUID-C)

abc123:topicA:0:1000:6 → null (epoch 6, deletion marker)

abc123:topicA:0:1000:6:UPDATE → null (epoch 5, deletion marker)

After Compaction

Compacted topic: All 4 keys removed (segment completely forgotten)

Cache result: Empty (segment no longer exists)

Why tombstone all 8 keys?

• Remove all historical metadata (successful + orphaned uploads + deletion marker)

• Ensures compacted topic completely "forgets" this segment

Scenario 5: Partition Deletion

Context: Entire partition deleted, cleanup all segments

Timeline & Messages

Cleanup Process

Compacted topic: All segment keys for abc123:topicA:0:*:*  get tombstones, partition delete marker remains

Public Interfaces

This KIP will add 2 new fields to the following records to help construct the message’s key published to the compacted topic, one is the logSegment’s endOffset and another one is brokerLeaderEpoch.

A backfill strategy to convert the current message to the new message will be shared in the migration plan part to make sure there is no public visible changes to the Kafka users.

RemoteLogSegmentMetadataRecord.json

[
//...
  {
    "name": "BrokerId",
    "type": "int32",
    "versions": "0+"
  },
  {
    "name": "RemoteLogSegmentId",
    "type": "RemoteLogSegmentIdEntry",
    "versions": "0+"
  },
  {
    "name": "StartOffset",
    "type": "int64",
    "versions": "0+"
  },
  {
    "name": "EndOffset",
    "type": "int64",
    "versions": "0+"
  },
  {
    "name": "BrokerId",
    "type": "int32",
    "versions": "0+"
  },
    {
      "name": "BrokerLeaderEpoch",
      "type": "int32",
      "versions": "1+",
      "about": "The leader epoch of the broker (partition leader epoch) at the time this update is being published.",
      "taggedVersions": "1+",
      "tag": 0
    }
]

RemoteLogSegmentMetadataUpdateRecord.json

[
  {
    "name": "RemoteLogSegmentId",
    "type": "RemoteLogSegmentIdEntry",
    "versions": "0+"
  },
  {
    "name": "BrokerId",
    "type": "int32",
    "versions": "0+"
  },
  //...
  {
    "name": "RemoteLogSegmentState",
    "type": "int8",
    "versions": "0+"
  },
    {
      "name": "BrokerLeaderEpoch",
      "type": "int32",
      "versions": "1+",
      "about": "The leader epoch of the broker (partition leader epoch) at the time this update is being published.",
      "taggedVersions": "1+",
      "tag": 0
    },
    {
      "name": "EndOffset",
      "type": "int64",
      "versions": "1+",
      "about": "End offset of the segment being updated.",
      "taggedVersions": "1+",
      "tag": 1
    }
]

RemotePartitionDeleteMetadataRecord.json

[
  {
    "name": "TopicIdPartition",
    "type": "TopicIdPartitionEntry",
    "versions": "0+"
  },
  {
    "name": "BrokerId",
    "type": "int32",
    "versions": "0+"
  },
  {
    "name": "EventTimestampMs",
    "type": "int64",
    "versions": "0+"
  },
  {
    "name": "RemotePartitionDeleteState",
    "type": "int8",
    "versions": "0+"
  },
    {
      "name": "BrokerLeaderEpoch",
      "type": "int32",
      "versions": "1+",
      "about": "The leader epoch of the broker (partition leader epoch) at the time this update is being published.",
      "taggedVersions": "1+",
      "tag": 0
    },
    {
      "name": "EndOffset",
      "type": "int64",
      "versions": "1+",
      "about": "End offset of the segment being updated.",
      "taggedVersions": "1+",
      "tag": 1
    }
]

Compatibility, Deprecation, and Migration Plan

For the message compatibility in the remote metadata topic, since the new fields are tagged fields, there will be no forward compatibility issue.

Migration Plan

The remote.log.metadata.version feature is introduced to enables a safe migration from time-based deletion to space-efficient log compaction through a three-version upgrade path.Versions:
- Version 0: Messages use null keys with cleanup.policy=delete. Cannot use compaction. This is the default and starting value for the field.
- Version 1: Topic uses cleanup.policy=compact,delete with customized retention to safely expire old null-key messages while enabling compaction for new keyed messages.
- Version 2: Pure cleanup.policy=compact mode. Removes retention limits for more aggressive compaction. Requires validation that no null-key messages remain in the topic.

Once the feature level is upgraded, downgrading or disabling the feature is not supported.

Scenario 1: New Cluster (Fresh Install)

kafka-storage.sh format -t <cluster-id> -c server.properties
kafka-server-start.sh server.properties

- `remote.log.metadata.version` automatically set to 2 (LATEST_PRODUCTION)
- Topic created with `cleanup.policy=compact`
- All messages have keys from the start
- No migration required

Scenario 2: Existing Cluster with Tiered Storage Already Enabled

For clusters that already have tiered storage enabled and an existing  __remote_log_metadata topic populated with historical metadata, the migration is incremental and non-disruptive.

Once the broker image with this KIP change runs on the machine, the broker will start to emit messages with keys, but the topic config change only happens with human operation.

A script kafka-remote-log-metadata-migration.sh will be provided to assist the migration. A metadata record with key remote.log.metadata.version will be used to track the progress of migration.

Upgrade Path (0 → 1 → 2):

Step 1: Upgrade to Version 1

kafka-remote-log-metadata-migration.sh \
--bootstrap-server localhost:9092 \
--upgrade-to-v1 \
--retention-ms 1209600000 # 14 days

Result:
- Topic config: `cleanup.policy=compact,delete`, `retention.ms=1209600000`, `min.compaction.lag.ms=1209600000`, `segment.ms=604800000`
- Feature: `remote.log.metadata.version=1`
- New messages have keys, old null-key messages will expire via retention

Step 2: Wait for Retention Period

- Wait at least 14 days (or your specified retention period)
- Allows all null-key messages to expire naturally
- Log cleaner won't compact yet (blocked by min.compaction.lag.ms)

Step 3: Validate and Upgrade to Version 2

kafka-remote-log-metadata-migration.sh \
--bootstrap-server localhost:9092 \
--check \
--upgrade-to-v2

Result:
- Tool displays retention reminder and checks if enough time has passed
- Scans topic for null-key messages
- Records last null-key message timestamp and suggests retry time if validation fails
- If validation passes:
- Feature: `remote.log.metadata.version=2`
- Topic config: `cleanup.policy=compact` (min.compaction.lag.ms and retention.ms overrides removed)

Scenario 3: Existing Cluster Enabling Tiered Storage for First Time

# Edit server.properties
remote.log.storage.system.enable=true

# Restart broker
kafka-server-start.sh server.properties

# Manually upgrade feature
kafka-features.sh upgrade --feature remote.log.metadata.version=2

- Feature version starts at 0 (not automatically upgraded)
- Topic will be created with the correct config on first use
- The feature value needs to be upgraded to 2 manually while no change will be applied to the topic.

Migration Script Sample Running Result

Help

usage: kafka-remote-log-metadata-migration

       [-h] --bootstrap-server BOOTSTRAP_SERVER [--command-config COMMAND_CONFIG] [--upgrade-to-v1] [--check] [--upgrade-to-v2] [--auto-upgrade] [--force]

       [--retention-ms RETENTION_MS] [--segment-ms SEGMENT_MS] [--timeout-ms TIMEOUT_MS]

Tool to manage remote.log.metadata.version upgrades and migrate the __remote_log_metadata topic.

optional arguments:

  -h, --help             show this help message and exit

  --bootstrap-server BOOTSTRAP_SERVER

                         REQUIRED: A comma-separated list of host:port pairs to use for establishing the connection to the Kafka cluster.

  --command-config COMMAND_CONFIG

                         Property file containing configs to be passed to Admin/Consumer Client.

  --upgrade-to-v1        Upgrade from remote.log.metadata.version=0 to version 1, and configure topic with min.compaction.lag.ms. (default: false)

  --check                Check if the topic contains any messages with null keys. This is required before upgrading to version 2. (default: false)

  --upgrade-to-v2        Upgrade to remote.log.metadata.version=2 after validation. Requires --check and --auto-upgrade. (default: false)

  --auto-upgrade         Automatically upgrade to version 2 if validation passes. Must be  used  with  --check --upgrade-to-v2. This is a safety flag to prevent accidental upgrades. (default: false)

  --force                Force upgrade to version 2 even if null-key messages  are  found.  Use  with  caution: null-key messages will be lost during compaction. (default:

                         false)

  --retention-ms RETENTION_MS

                         Retention period in milliseconds for the __remote_log_metadata topic when  upgrading  to  version  1 (default: 1209600000, which is 14 days). This

                         parameter is CRITICAL: it serves two purposes: 1) retention.ms: Ensures  old-format  (null-key) messages expire and are deleted after this period.

                         2) min.compaction.lag.ms: Set to the same value to prevent log cleaner  from  compacting before old messages expire. Once the topic cleanup policy

                         is changed to 'compact,delete', the log cleaner could  immediately  delete  null-key  messages during compaction. By setting both retention.ms and

                         min.compaction.lag.ms to the same value, we ensure null-key  messages  expire  naturally  via  retention before the log cleaner begins compacting.

                         This prevents data loss during the migration period. Used with --upgrade-to-v1. (default: 1209600000)

  --segment-ms SEGMENT_MS

                         Segment rolling time in milliseconds for the __remote_log_metadata topic  when  upgrading  to  version 1 (default: 86400000, which is 1 day). This

                         controls how frequently new log segments are created. Smaller  values  create  more segments, which can improve compaction efficiency but increase

                         overhead. Used with --upgrade-to-v1. (default: 86400000)

  --timeout-ms TIMEOUT_MS

                         Maximum time in milliseconds to wait while checking for messages (default: 60000). Used with --check. (default: 60000)

Upgrade-to-v1

./bin/kafka-remote-log-metadata-migration.sh \
  --bootstrap-server localhost:9092 \
  --upgrade-to-v1 \
  --retention-ms 600000 
Initiating upgrade to remote.log.metadata.version=1...

SLF4J: Class path contains multiple SLF4J bindings.
SLF4J: Found binding in [jar:file:/Users/lijuntong/Documents/kafka/kafka/tools/build/dependant-libs-2.13.18/log4j-slf4j-impl-2.25.4.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: Found binding in [jar:file:/Users/lijuntong/Documents/kafka/kafka/connect/runtime/build/dependant-libs/log4j-slf4j-impl-2.25.4.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: See http://www.slf4j.org/codes.html#multiple_bindings for an explanation.
SLF4J: Actual binding is of type [org.apache.logging.slf4j.Log4jLoggerFactory]
Current remote.log.metadata.version: 0
Pre-configuring __remote_log_metadata topic...
  - cleanup.policy=compact,delete
  - retention.ms=600000 
  - min.compaction.lag.ms=600000
  - segment.ms=86400000

IMPORTANT: retention.ms and min.compaction.lag.ms are critical for safe migration.
Once cleanup.policy becomes 'compact,delete', the log cleaner can immediately delete null-key messages during compaction.
Setting retention.ms=600000ms ensures old-format (null-key) messages expire after 0 days.
Setting min.compaction.lag.ms to the same value ensures the log cleaner waits 0 days before compacting,
allowing null-key messages to expire naturally via retention before compaction begins.
This prevents data loss during the migration period.

✅ Topic configurations updated successfully.

Upgrading feature to version 1...
✅ Feature upgraded to version 1 successfully.

✅ Upgrade to version 1 completed successfully!

==================== NEXT STEPS ====================

CRITICAL: You must wait for old messages to be fully deleted before proceeding to version 2.

Expected deletion time: segment.ms + retention.ms = 1 + 0 = 1 days

Why this waiting period is necessary:
1. Messages deletion happens in two phases:
   Phase 1 (segment.ms = 1 days):
   - Old-format (null-key) messages stay in active segment
   - Cannot be deleted while in active segment, even if expired

   Phase 2 (retention.ms = 0 days):
   - After segment rolls, messages move to closed segment
   - Messages expire based on retention.ms=600000ms
   - Log cleaner deletes expired messages

2. During this period:
   - The log cleaner will NOT compact the topic yet (prevented by min.compaction.lag.ms=600000ms)
   - This ensures null-key messages are deleted via retention, NOT via compaction

3. After waiting 1 days, run validation and upgrade to version 2:
   kafka-remote-log-metadata-migration.sh --bootstrap-server localhost:9092 --check --auto-upgrade

4. The validation check will:
   - Scan the entire __remote_log_metadata topic for any remaining null-key messages
   - Only proceed with upgrade if NO null-key messages are found
   - Change cleanup.policy to 'compact' (removing 'delete')
   - Set retention.ms to -1 (infinite retention)
   - Remove min.compaction.lag.ms override (use broker default)

===================================================

Upgrade-to-v2

When there are still keyless messages

 ./bin/kafka-remote-log-metadata-migration.sh \
  --bootstrap-server localhost:9092 \
  --upgrade-to-v2 \
  --check               
SLF4J: Class path contains multiple SLF4J bindings.
SLF4J: Found binding in [jar:file:/Users/lijuntong/Documents/kafka/kafka/core/build/dependant-libs-2.13.18/log4j-slf4j-impl-2.25.4.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: Found binding in [jar:file:/Users/lijuntong/Documents/kafka/kafka/tools/build/dependant-libs-2.13.18/log4j-slf4j-impl-2.25.4.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: Found binding in [jar:file:/Users/lijuntong/Documents/kafka/kafka/connect/runtime/build/dependant-libs/log4j-slf4j-impl-2.25.4.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: See http://www.slf4j.org/codes.html#multiple_bindings for an explanation.
SLF4J: Actual binding is of type [org.apache.logging.slf4j.Log4jLoggerFactory]
========== IMPORTANT REMINDER ==========

Current __remote_log_metadata topic configuration:
  - retention.ms=600000ms (0 days)
  - min.compaction.lag.ms=600000ms (0 days)

Before proceeding with this validation, ensure that:
1. At least 0 days have passed since upgrading to version 1
2. This allows all old-format (null-key) messages to expire via retention
3. The log cleaner has NOT compacted the topic yet (prevented by min.compaction.lag.ms)

If you upgraded to version 1 recently (less than 0 days ago),
you should WAIT before running this validation to avoid false negatives.

========================================

Checking __remote_log_metadata topic for messages with null keys...
Bootstrap servers: localhost:9092
Timeout: 60000ms
Upgrade to V2: ENABLED (will upgrade to version 2 if validation passes)

Found 5 partition(s) in __remote_log_metadata
Scanning messages...
⚠️  Found message with null key at partition=4, offset=6, timestamp=1780448344546
⚠️  Found message with null key at partition=4, offset=7, timestamp=1780448350864
⚠️  Found message with null key at partition=4, offset=8, timestamp=1780448357366
⚠️  Found message with null key at partition=4, offset=9, timestamp=1780448357367

//Skip the remaining similar logs for message with null key

Scan completed.
Total messages scanned: 1878
Messages with null keys: 24

❌ VALIDATION FAILED: Found 24 message(s) with null keys.

Last null-key message timestamp: 1780448367505
Last null-key message age: 0 days (0 hours)

Topic configuration:
  - segment.ms: 86400000ms (1 days)
  - retention.ms: 600000ms (0 days)
  - Total wait time for deletion: 87000000ms (1 days)

💡 SUGGESTION:

Estimated cleanup time: Thu Jun 04 09:09:27 CST 2026

Remaining wait time: approximately 24 hours.

Please retry this validation after the estimated cleanup time.

Action required:
1. Wait for null-key messages to expire based on retention.ms setting
2. Then run this tool again to verify all null-key messages are gone
3. Only then proceed with the upgrade to remote.log.metadata.version=2

To force upgrade despite null-key messages (NOT RECOMMENDED), use --force flag.
Cannot upgrade to version 2: null-key messages found in __remote_log_metadata

When there are no keyless messages

./bin/kafka-remote-log-metadata-migration.sh \
  --bootstrap-server localhost:9092 \
  --upgrade-to-v2 \
  --check --auto-upgrade
SLF4J: Class path contains multiple SLF4J bindings.
SLF4J: Found binding in [jar:file:/Users/lijuntong/Documents/kafka/kafka/tools/build/dependant-libs-2.13.18/log4j-slf4j-impl-2.25.4.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: Found binding in [jar:file:/Users/lijuntong/Documents/kafka/kafka/connect/runtime/build/dependant-libs/log4j-slf4j-impl-2.25.4.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: See http://www.slf4j.org/codes.html#multiple_bindings for an explanation.
SLF4J: Actual binding is of type [org.apache.logging.slf4j.Log4jLoggerFactory]
========== IMPORTANT REMINDER ==========

Current __remote_log_metadata topic configuration:
  - retention.ms=600000ms 
  - min.compaction.lag.ms=600000ms

Before proceeding with this validation, ensure that:
1. At least 0 days have passed since upgrading to version 1
2. This allows all old-format (null-key) messages to expire via retention
3. The log cleaner has NOT compacted the topic yet (prevented by min.compaction.lag.ms)

If you upgraded to version 1 recently (less than 0 days ago),
you should WAIT before running this validation to avoid false negatives.

========================================

Checking __remote_log_metadata topic for messages with null keys...
Bootstrap servers: localhost:9092
Timeout: 60000ms
Upgrade to V2: ENABLED (will upgrade to version 2 if validation passes)

Found 5 partition(s) in __remote_log_metadata
Scanning messages...

Scan completed.
Total messages scanned: 78
Messages with null keys: 0

✅ VALIDATION PASSED: No null-key messages found.
✅ Safe to upgrade to remote.log.metadata.version=2.

Initiating upgrade to remote.log.metadata.version=2...

Current remote.log.metadata.version: 1
Upgrading from version 1 to version 2...

✅ Successfully upgraded to remote.log.metadata.version=2!

The controller has automatically updated __remote_log_metadata topic configuration:
  - cleanup.policy changed to 'compact' (compact-only)
  - min.compaction.lag.ms override removed
  - retention.ms override removed

All metadata messages now have proper keys and will be retained indefinitely via compaction.

Test Plan

Topic Functionality Tests

• Confirm all metadata events are properly written to both topics with correct key formats

• Validate that the compacted topic maintains only the latest state per segment while the original topic preserves full history

Migration Verification

• Migration (Existing Clusters)

  • With finite retention.ms, upgrade brokers.
  • Confirm new records are keyed and readable.
  • Rebuild RemoteLogMetadataCache from __remote_log_metadata and verify:
    • All expected remote segments are present.
    • Offsets and states are correct.
  • After ≥ one retention window and with compaction enabled (if not already), confirm:
    • Logical state (live segments, states) is unchanged.
    • Topic only contains keyed records and tombstones.

Durability and Recovery Tests

• Simulate broker failures during various stages of segment lifecycle (upload, deletion)

• Test leader changes during metadata operations to verify correct handling of concurrent operations

• Verify metadata consistency after unclean shutdowns and broker restarts

Performance Benchmarks

• Measure broker startup time before and after implementation

• Compare metadata topic size growth over time between current and new implementation

Rejected Alternatives

Alternative A: Snapshot Approach (KAFKA-19265)

Rejected Reasons:

1. Distributed Coordination Complexity: Creating consistent snapshots across distributed brokers with different partition leaders is extremely challenging and error-prone.

2. Leader Transition Reliability: Ensuring new leaders receive the latest snapshot during transitions is difficult to guarantee, especially during network partitions or failures.

3. Recovery Vulnerabilities: Corrupted or incomplete snapshots would require complex recovery procedures, potentially defeating the purpose of having snapshots.

4. Operational Overhead: Additional storage requirements, snapshot management create significant operational burden.

Alternative B: Adjust Retention Time Based on Maximum Tiered Storage Topic Retention

Rejected Reasons:

1. Fails to Address Core Problem: This approach doesn't solve the unbounded metadata growth issue - it merely postpones it by extending retention periods.

2. Resource Inefficiency and Lack of Granularity: Most metadata records would be retained far longer than necessary, wasting the storage and impacting query and bootstrap performance.

3. Implementation Complexity: Creating a mechanism to track and dynamically update retention across all tiered storage topics introduces unnecessary complexity.

4. Risk of Irrecoverable Metadata Loss: If the remote storage module experiences downtime, truncating the __remote_log_metadata  topic based on retention policies could lead to permanent loss of remote log metadata, even when the remote storage still has those remote log segments, it will cause certain discrepancy between the remote storage and remote logsegment metadata state broker is preserving.