KIP-1266: Send keyed events to the remote log metadata topic

storage/api/src/main/java/org/apache/kafka/server/log/remote/storage/RemoteLogMetadata.java

@@ -60,4 +60,8 @@ public int brokerId() {
      * @return TopicIdPartition for which this event is generated.
      */
     public abstract TopicIdPartition topicIdPartition();
+
+    public RemoteLogSegmentMetadataKey metadataKey() {
+        return null;
+    }
 }

storage/api/src/main/java/org/apache/kafka/server/log/remote/storage/RemoteLogSegmentMetadata.java

@@ -325,6 +325,11 @@ public TopicIdPartition topicIdPartition() {
         return remoteLogSegmentId.topicIdPartition();
     }
 
+    @Override
+    public RemoteLogSegmentMetadataKey metadataKey() {
+        return RemoteLogSegmentMetadataKey.of(remoteLogSegmentId, state);
+    }
+
     @Override
     public boolean equals(Object o) {
         if (this == o) {

storage/api/src/main/java/org/apache/kafka/server/log/remote/storage/RemoteLogSegmentMetadataKey.java

@@ -0,0 +1,43 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.server.log.remote.storage;
+
+import org.apache.kafka.common.Uuid;
+
+import java.util.Objects;
+
+/**
+ * Structured key for remote log segment metadata Kafka records.
+ */
+public record RemoteLogSegmentMetadataKey(Uuid topicId, int partition, Uuid segmentId, byte stateId) {
+
+    public RemoteLogSegmentMetadataKey(Uuid topicId, int partition, Uuid segmentId, byte stateId) {
+        this.topicId = Objects.requireNonNull(topicId, "topicId cannot be null");
+        this.partition = partition;
+        this.segmentId = Objects.requireNonNull(segmentId, "segmentId cannot be null");
+        this.stateId = stateId;
+    }
+
+    public static RemoteLogSegmentMetadataKey of(RemoteLogSegmentId segmentId, RemoteLogSegmentState state) {
+        return new RemoteLogSegmentMetadataKey(
+                segmentId.topicIdPartition().topicId(),
+                segmentId.topicIdPartition().partition(),
+                segmentId.id(),
+                state.id()
+        );
+    }
+}

storage/api/src/main/java/org/apache/kafka/server/log/remote/storage/RemoteLogSegmentMetadataUpdate.java

@@ -89,6 +89,11 @@ public TopicIdPartition topicIdPartition() {
         return remoteLogSegmentId.topicIdPartition();
     }
 
+    @Override
+    public RemoteLogSegmentMetadataKey metadataKey() {
+        return RemoteLogSegmentMetadataKey.of(remoteLogSegmentId, state);
+    }
+
     @Override
     public boolean equals(Object o) {
         if (this == o) {

storage/src/main/java/org/apache/kafka/server/log/remote/metadata/storage/ConsumerTask.java

@@ -168,13 +168,17 @@ void closeConsumer() {
     }
 
     private void processConsumerRecord(ConsumerRecord<byte[], byte[]> record) {
-        final RemoteLogMetadata remoteLogMetadata = serde.deserialize(record.value());
-        if (shouldProcess(remoteLogMetadata, record.offset())) {
-            remotePartitionMetadataEventHandler.handleRemoteLogMetadata(remoteLogMetadata);
-            readOffsetsByUserTopicPartition.put(remoteLogMetadata.topicIdPartition(), record.offset());
-        } else {
-            log.trace("The event {} is skipped because it is either already processed or not assigned to this consumer",
-                    remoteLogMetadata);
+        byte[] value = record.value();
+        // skip the tombstone records
+        if (value != null) {
+            final RemoteLogMetadata remoteLogMetadata = serde.deserialize(value);
+            if (shouldProcess(remoteLogMetadata, record.offset())) {
+                remotePartitionMetadataEventHandler.handleRemoteLogMetadata(remoteLogMetadata);
+                readOffsetsByUserTopicPartition.put(remoteLogMetadata.topicIdPartition(), record.offset());
+            } else {
+                log.trace("The event {} is skipped because it is either already processed or not assigned to this consumer",
+                        remoteLogMetadata);
+            }
         }
         log.trace("Updating consumed offset: {} for partition {}", record.offset(), record.partition());
         readOffsetsByMetadataPartition.put(record.partition(), record.offset());

storage/src/main/java/org/apache/kafka/server/log/remote/metadata/storage/ProducerManager.java

@@ -18,12 +18,18 @@
 
 import org.apache.kafka.clients.producer.Callback;
 import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.Producer;
 import org.apache.kafka.clients.producer.ProducerRecord;
 import org.apache.kafka.clients.producer.RecordMetadata;
 import org.apache.kafka.common.KafkaException;
 import org.apache.kafka.common.TopicIdPartition;
 import org.apache.kafka.server.log.remote.metadata.storage.serialization.RemoteLogMetadataSerde;
+import org.apache.kafka.server.log.remote.metadata.storage.serialization.RemoteLogSegmentMetadataKeySerde;
 import org.apache.kafka.server.log.remote.storage.RemoteLogMetadata;
+import org.apache.kafka.server.log.remote.storage.RemoteLogSegmentId;
+import org.apache.kafka.server.log.remote.storage.RemoteLogSegmentMetadataKey;
+import org.apache.kafka.server.log.remote.storage.RemoteLogSegmentMetadataUpdate;
+import org.apache.kafka.server.log.remote.storage.RemoteLogSegmentState;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -40,24 +46,36 @@
 public class ProducerManager implements Closeable {
     private static final Logger log = LoggerFactory.getLogger(ProducerManager.class);
 
+    private final RemoteLogSegmentMetadataKeySerde keySerde = new RemoteLogSegmentMetadataKeySerde();
     private final RemoteLogMetadataSerde serde = new RemoteLogMetadataSerde();
-    private final KafkaProducer<byte[], byte[]> producer;
+    private final Producer<byte[], byte[]> producer;
     private final RemoteLogMetadataTopicPartitioner topicPartitioner;
     private final TopicBasedRemoteLogMetadataManagerConfig rlmmConfig;
+    private final Callback tombstoneRecordsCallback = (metadata, exception) -> {
+        if (exception != null) {
+            log.error("Failed to publish tombstone records for: {}", metadata, exception);
+        }
+    };
 
     public ProducerManager(TopicBasedRemoteLogMetadataManagerConfig rlmmConfig,
                            RemoteLogMetadataTopicPartitioner rlmmTopicPartitioner) {
+        this(rlmmConfig, rlmmTopicPartitioner, new KafkaProducer<>(rlmmConfig.producerProperties()));
+    }
+
+    public ProducerManager(TopicBasedRemoteLogMetadataManagerConfig rlmmConfig,
+                           RemoteLogMetadataTopicPartitioner rlmmTopicPartitioner,
+                           Producer<byte[], byte[]> producer) {
         this.rlmmConfig = rlmmConfig;
-        this.producer = new KafkaProducer<>(rlmmConfig.producerProperties());
-        topicPartitioner = rlmmTopicPartitioner;
+        this.topicPartitioner = rlmmTopicPartitioner;
+        this.producer = producer;
     }
 
     /**
      * Returns {@link CompletableFuture} which will complete only after publishing of the given {@code remoteLogMetadata}
      * is considered complete.
      *
      * @param remoteLogMetadata RemoteLogMetadata to be published
-     * @return CompletableFuture that completes with RecordMetadata when the message is successfully published
+     * @return a future with acknowledgement.
      */
     CompletableFuture<RecordMetadata> publishMessage(RemoteLogMetadata remoteLogMetadata) {
         CompletableFuture<RecordMetadata> future = new CompletableFuture<>();
@@ -80,15 +98,47 @@ CompletableFuture<RecordMetadata> publishMessage(RemoteLogMetadata remoteLogMeta
                     future.complete(metadata);
                 }
             };
-            producer.send(new ProducerRecord<>(rlmmConfig.remoteLogMetadataTopicName(), metadataPartitionNum, null,
-                                               serde.serialize(remoteLogMetadata)), callback);
+            String topic = rlmmConfig.remoteLogMetadataTopicName();
+            byte[] serializedKey = keySerde.serializer().serialize(topic, remoteLogMetadata.metadataKey());
+            byte[] serializedValue = serde.serialize(remoteLogMetadata);
+            producer.send(new ProducerRecord<>(topic, metadataPartitionNum, serializedKey, serializedValue), callback);
         } catch (Exception ex) {
             future.completeExceptionally(ex);
         }
 
         return future;
     }
 
+    /**
+     * Publishes tombstone records to mark the deletion of remote log segment metadata when the state of the provided
+     * {@link RemoteLogMetadata} instance is {@link RemoteLogSegmentState#DELETE_SEGMENT_FINISHED}.
+     *
+     * @param remoteLogMetadata The {@link RemoteLogMetadata} instance containing metadata information. If the metadata is an instance
+     *                          of {@link RemoteLogSegmentMetadataUpdate} and its state is {@link RemoteLogSegmentState#DELETE_SEGMENT_FINISHED},
+     *                          tombstone records are published to indicate its deletion.
+     */
+    public void maybePublishTombstoneRecords(RemoteLogMetadata remoteLogMetadata) {
+        // TODO: Gate sending tombstone records behind a feature flag.
+        //       Send the tombstone records only when the consumerTask can handle the null values.
+        if (remoteLogMetadata instanceof RemoteLogSegmentMetadataUpdate metadataUpdate) {
+            if (metadataUpdate.state() == RemoteLogSegmentState.DELETE_SEGMENT_FINISHED) {
+                RemoteLogSegmentId remoteLogSegmentId = metadataUpdate.remoteLogSegmentId();
+                int metadataPartition = topicPartitioner.metadataPartition(remoteLogSegmentId.topicIdPartition());
+                String topic = rlmmConfig.remoteLogMetadataTopicName();
+                try {
+                    // Send the tombstone records for all the RemoteLogSegment state to cleanup the expired segment metadata.
+                    for (RemoteLogSegmentState state : RemoteLogSegmentState.values()) {
+                        RemoteLogSegmentMetadataKey metadataKey = RemoteLogSegmentMetadataKey.of(remoteLogSegmentId, state);
+                        byte[] serializedKey = keySerde.serializer().serialize(topic, metadataKey);
+                        producer.send(new ProducerRecord<>(topic, metadataPartition, serializedKey, null), tombstoneRecordsCallback);
+                    }
+                } catch (Exception ex) {
+                    log.error("Failed to publish tombstone records for: {}", metadataUpdate, ex);
+                }
+            }
+        }
+    }
+
     public void close() {
         try {
             producer.close(Duration.ofSeconds(30));

storage/src/main/java/org/apache/kafka/server/log/remote/metadata/storage/TopicBasedRemoteLogMetadataManager.java

@@ -162,7 +162,7 @@ private CompletableFuture<Void> storeRemoteLogMetadata(RemoteLogMetadata remoteL
                 } catch (TimeoutException e) {
                     throw new KafkaException(e);
                 }
-            });
+            }).thenRunAsync(() -> producerManager.maybePublishTombstoneRecords(remoteLogMetadata));
         } catch (KafkaException e) {
             if (e instanceof RetriableException) {
                 throw e;

storage/src/main/java/org/apache/kafka/server/log/remote/metadata/storage/serialization/RemoteLogSegmentMetadataKeySerde.java

@@ -0,0 +1,85 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.server.log.remote.metadata.storage.serialization;
+
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.serialization.Deserializer;
+import org.apache.kafka.common.serialization.Serde;
+import org.apache.kafka.common.serialization.Serializer;
+import org.apache.kafka.server.log.remote.storage.RemoteLogSegmentMetadataKey;
+
+import java.nio.ByteBuffer;
+
+/**
+ * Serde for {@link RemoteLogSegmentMetadataKey}.
+ * <p>
+ * Serializes to a 38-byte binary representation:
+ * <ul>
+ *   <li>1 byte  – format version (currently {@value #VERSION})</li>
+ *   <li>8 bytes – topicId most-significant bits</li>
+ *   <li>8 bytes – topicId least-significant bits</li>
+ *   <li>4 bytes – partition (big-endian int)</li>
+ *   <li>8 bytes – segmentId most-significant bits</li>
+ *   <li>8 bytes – segmentId least-significant bits</li>
+ *   <li>1 byte  – stateId</li>
+ * </ul>
+ */
+public class RemoteLogSegmentMetadataKeySerde implements Serde<RemoteLogSegmentMetadataKey> {
+
+    static final byte VERSION = 0;
+    static final int SERIALIZED_SIZE = 1 + 8 + 8 + 4 + 8 + 8 + 1; // 38 bytes
+
+    @Override
+    public Serializer<RemoteLogSegmentMetadataKey> serializer() {
+        return (topic, key) -> {
+            if (key == null) return null;
+            ByteBuffer buf = ByteBuffer.allocate(SERIALIZED_SIZE);
+            buf.put(VERSION);
+            buf.putLong(key.topicId().getMostSignificantBits());
+            buf.putLong(key.topicId().getLeastSignificantBits());
+            buf.putInt(key.partition());
+            buf.putLong(key.segmentId().getMostSignificantBits());
+            buf.putLong(key.segmentId().getLeastSignificantBits());
+            buf.put(key.stateId());
+            return buf.array();
+        };
+    }
+
+    @Override
+    public Deserializer<RemoteLogSegmentMetadataKey> deserializer() {
+        return (topic, data) -> {
+            if (data == null)
+                return null;
+
+            if (data.length != SERIALIZED_SIZE) {
+                throw new IllegalArgumentException(
+                        "Expected " + SERIALIZED_SIZE + " bytes but got " + data.length);
+            }
+            ByteBuffer buf = ByteBuffer.wrap(data);
+            byte version = buf.get();
+            if (version != VERSION) {
+                throw new IllegalArgumentException(
+                        "Unsupported RemoteLogSegmentMetadataKey version: " + version);
+            }
+            Uuid topicId = new Uuid(buf.getLong(), buf.getLong());
+            int partition = buf.getInt();
+            Uuid segmentId = new Uuid(buf.getLong(), buf.getLong());
+            byte stateId = buf.get();
+            return new RemoteLogSegmentMetadataKey(topicId, partition, segmentId, stateId);
+        };
+    }
+}