Copilot commented on code in PR #22774:
URL: https://github.com/apache/kafka/pull/22774#discussion_r3537617445
##########
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;
+ }
Review Comment:
`metadataKey()` is a new public API but it currently has no contract/javadoc
and always returns null by default. Since producers now use this method to
build the Kafka record key, the API should clearly document whether null is
allowed and what callers should expect (especially if the goal is to eventually
make the topic fully keyed for compaction).
##########
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) {
Review Comment:
The key includes `stateId`, which means state transitions for the same
remote segment will use different Kafka keys. That undermines the main benefit
of compaction for bounding message count (you’ll retain one record per state
rather than one per segment), and forces deletion to emit multiple tombstones.
Consider using a key that is stable across state transitions (e.g.,
topicId+partition+segmentId only) and keeping state solely in the value.
##########
storage/src/main/java/org/apache/kafka/server/log/remote/metadata/storage/serialization/RemoteLogSegmentMetadataKeySerde.java:
##########
@@ -0,0 +1,75 @@
+/*
+ * 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 compact 37-byte binary representation:
+ * <ul>
+ * <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 int SERIALIZED_SIZE = 8 + 8 + 4 + 8 + 8 + 1; // 37 bytes
+
+ @Override
+ public Serializer<RemoteLogSegmentMetadataKey> serializer() {
+ return (topic, key) -> {
Review Comment:
`RemoteLogSegmentMetadataKeySerde` hard-codes a fixed binary layout without
a version byte. Since these keys are persisted in Kafka and may need to evolve,
consider prefixing the serialized key with a format/version byte (similar to
other Kafka protocol/record encodings) to allow forward-compatible changes.
##########
storage/src/main/java/org/apache/kafka/server/log/remote/metadata/storage/ProducerManager.java:
##########
@@ -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);
Review Comment:
`publishMessage` serializes the record key from
`remoteLogMetadata.metadataKey()`, but for metadata types that don’t override
`metadataKey()` (e.g., RemotePartitionDeleteMetadata,
RemoteLogSegmentMetadataSnapshot) this will still produce null keys. If the
long-term plan is to enable compaction, leaving some record types unkeyed will
block that or lead to mixed-key topics; consider defining keys for all record
types that can appear in `__remote_log_metadata`, or fail fast/emit a metric
when a null key is produced.
##########
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())) {
Review Comment:
New behavior skips tombstone records (null values) during consumption. This
is important to exercise in unit tests to ensure (a) null values don’t cause
deserialization failures, (b) offsets are still tracked for the metadata
partition, and (c) the handler is not invoked for tombstones.
##########
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);
}
- });
+ }).thenAcceptAsync(recordMetadata ->
producerManager.maybePublishTombstoneRecords(remoteLogMetadata));
Review Comment:
The second stage uses `thenAcceptAsync` even though the upstream future
completes with `Void`, so the lambda parameter is not a `RecordMetadata` (it’s
always null) and the name is misleading. Use `thenRun(...)`/`thenRunAsync(...)`
instead to make the intent clear, and consider using the same executor as the
preceding stage to keep ordering predictable.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
