chia7712 commented on code in PR #23689:
URL: https://github.com/apache/kafka/pull/23689#discussion_r4176659114
##########
storage/src/main/java/org/apache/kafka/server/log/remote/metadata/storage/serialization/RemoteLogMetadataSerde.java:
##########
@@ -33,76 +35,65 @@
import
org.apache.kafka.server.log.remote.storage.RemotePartitionDeleteMetadata;
import java.io.PrintStream;
+import java.nio.ByteBuffer;
/**
* This class provides serialization and deserialization for {@link
RemoteLogMetadata}. This is the root serde
* for the messages that are stored in internal remote log metadata topic.
*/
public class RemoteLogMetadataSerde {
- private static final short REMOTE_LOG_SEGMENT_METADATA_API_KEY = new
RemoteLogSegmentMetadataRecord().apiKey();
- private static final short REMOTE_LOG_SEGMENT_METADATA_UPDATE_API_KEY =
new RemoteLogSegmentMetadataUpdateRecord().apiKey();
- private static final short REMOTE_PARTITION_DELETE_API_KEY = new
RemotePartitionDeleteMetadataRecord().apiKey();
- private static final short REMOTE_LOG_SEGMENT_METADATA_SNAPSHOT_API_KEY =
new RemoteLogSegmentMetadataSnapshotRecord().apiKey();
-
- private final BytesApiMessageSerde bytesApiMessageSerde;
-
- private final RemoteLogSegmentMetadataTransform segmentTransform;
- private final RemoteLogSegmentMetadataUpdateTransform
segmentUpdateTransform;
- private final RemotePartitionDeleteMetadataTransform
partitionDeleteTransform;
- private final RemoteLogSegmentMetadataSnapshotTransform
segmentSnapshotTransform;
-
- public RemoteLogMetadataSerde() {
- bytesApiMessageSerde = new BytesApiMessageSerde() {
- @Override
- public ApiMessage apiMessageFor(short apiKey) {
- return newApiMessage(apiKey);
- }
- };
- segmentTransform = new RemoteLogSegmentMetadataTransform();
- segmentUpdateTransform = new RemoteLogSegmentMetadataUpdateTransform();
- partitionDeleteTransform = new
RemotePartitionDeleteMetadataTransform();
- segmentSnapshotTransform = new
RemoteLogSegmentMetadataSnapshotTransform();
- }
+ private static final AbstractApiMessageSerde API_MESSAGE_SERDE = new
AbstractApiMessageSerde() {
+ @Override
+ public ApiMessage apiMessageFor(short apiKey) {
+ return MetadataRecordType.fromId(apiKey).newMetadataRecord();
+ }
+ };
- protected ApiMessage newApiMessage(short apiKey) {
- return MetadataRecordType.fromId(apiKey).newMetadataRecord();
- }
+ private final RemoteLogSegmentMetadataTransform segmentTransform = new
RemoteLogSegmentMetadataTransform();
+ private final RemoteLogSegmentMetadataUpdateTransform
segmentUpdateTransform = new RemoteLogSegmentMetadataUpdateTransform();
+ private final RemotePartitionDeleteMetadataTransform
partitionDeleteTransform = new RemotePartitionDeleteMetadataTransform();
+ private final RemoteLogSegmentMetadataSnapshotTransform
segmentSnapshotTransform = new RemoteLogSegmentMetadataSnapshotTransform();
public byte[] serialize(RemoteLogMetadata remoteLogMetadata) {
+ ApiMessageAndVersion apiMessageAndVersion =
toApiMessageAndVersion(remoteLogMetadata);
- ApiMessageAndVersion apiMessageAndVersion;
- if (remoteLogMetadata instanceof RemoteLogSegmentMetadata) {
- apiMessageAndVersion =
segmentTransform.toApiMessageAndVersion((RemoteLogSegmentMetadata)
remoteLogMetadata);
- } else if (remoteLogMetadata instanceof
RemoteLogSegmentMetadataUpdate) {
- apiMessageAndVersion =
segmentUpdateTransform.toApiMessageAndVersion((RemoteLogSegmentMetadataUpdate)
remoteLogMetadata);
- } else if (remoteLogMetadata instanceof RemotePartitionDeleteMetadata)
{
- apiMessageAndVersion =
partitionDeleteTransform.toApiMessageAndVersion((RemotePartitionDeleteMetadata)
remoteLogMetadata);
- } else if (remoteLogMetadata instanceof
RemoteLogSegmentMetadataSnapshot) {
- apiMessageAndVersion =
segmentSnapshotTransform.toApiMessageAndVersion((RemoteLogSegmentMetadataSnapshot)
remoteLogMetadata);
- } else {
- throw new IllegalArgumentException("RemoteLogMetadataTransform for
given RemoteStorageMetadata class: " + remoteLogMetadata.getClass()
- + " does not exist.");
- }
-
- return bytesApiMessageSerde.serialize(apiMessageAndVersion);
+ ObjectSerializationCache cache = new ObjectSerializationCache();
+ int size = API_MESSAGE_SERDE.recordSize(apiMessageAndVersion, cache);
+ ByteBufferAccessor writable = new
ByteBufferAccessor(ByteBuffer.allocate(size));
+ API_MESSAGE_SERDE.write(apiMessageAndVersion, cache, writable);
+ return writable.buffer().array();
}
public RemoteLogMetadata deserialize(byte[] data) {
- ApiMessageAndVersion apiMessageAndVersion =
bytesApiMessageSerde.deserialize(data);
+ ApiMessageAndVersion apiMessageAndVersion = API_MESSAGE_SERDE.read(new
ByteBufferAccessor(ByteBuffer.wrap(data)), data.length);
- short apiKey = apiMessageAndVersion.message().apiKey();
- if (apiKey == REMOTE_LOG_SEGMENT_METADATA_API_KEY) {
+ ApiMessage message = apiMessageAndVersion.message();
+ if (message instanceof RemoteLogSegmentMetadataRecord) {
return
segmentTransform.fromApiMessageAndVersion(apiMessageAndVersion);
- } else if (apiKey == REMOTE_LOG_SEGMENT_METADATA_UPDATE_API_KEY) {
+ } else if (message instanceof RemoteLogSegmentMetadataUpdateRecord) {
return
segmentUpdateTransform.fromApiMessageAndVersion(apiMessageAndVersion);
- } else if (apiKey == REMOTE_PARTITION_DELETE_API_KEY) {
+ } else if (message instanceof RemotePartitionDeleteMetadataRecord) {
return
partitionDeleteTransform.fromApiMessageAndVersion(apiMessageAndVersion);
- } else if (apiKey == REMOTE_LOG_SEGMENT_METADATA_SNAPSHOT_API_KEY) {
+ } else if (message instanceof RemoteLogSegmentMetadataSnapshotRecord) {
return
segmentSnapshotTransform.fromApiMessageAndVersion(apiMessageAndVersion);
} else {
- throw new IllegalArgumentException("RemoteLogMetadataTransform for
apikey: " + apiKey + " does not exist.");
+ throw new IllegalArgumentException("RemoteLogMetadataTransform for
apikey: " + message.apiKey() + " does not exist.");
}
+ }
+ private ApiMessageAndVersion toApiMessageAndVersion(RemoteLogMetadata
remoteLogMetadata) {
+ if (remoteLogMetadata instanceof RemoteLogSegmentMetadata metadata) {
+ return segmentTransform.toApiMessageAndVersion(metadata);
+ } else if (remoteLogMetadata instanceof RemoteLogSegmentMetadataUpdate
metadataUpdate) {
+ return
segmentUpdateTransform.toApiMessageAndVersion(metadataUpdate);
+ } else if (remoteLogMetadata instanceof RemotePartitionDeleteMetadata
deleteMetadata) {
+ return
partitionDeleteTransform.toApiMessageAndVersion(deleteMetadata);
+ } else if (remoteLogMetadata instanceof
RemoteLogSegmentMetadataSnapshot snapshot) {
+ return segmentSnapshotTransform.toApiMessageAndVersion(snapshot);
+ } else {
+ throw new IllegalArgumentException("RemoteLogMetadataTransform for
given RemoteStorageMetadata class: " + remoteLogMetadata.getClass()
Review Comment:
`RemoteStorageMetadata` -> `RemoteLogMetadata`?
--
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]
