satishd commented on a change in pull request #11033:
URL: https://github.com/apache/kafka/pull/11033#discussion_r704021954
##########
File path:
storage/src/main/java/org/apache/kafka/server/log/remote/metadata/storage/TopicBasedRemoteLogMetadataManager.java
##########
@@ -129,37 +132,44 @@ public void
updateRemoteLogSegmentMetadata(RemoteLogSegmentMetadataUpdate segmen
}
// Publish the message to the topic.
-
doPublishMetadata(segmentMetadataUpdate.remoteLogSegmentId().topicIdPartition(),
segmentMetadataUpdate);
+ return
doPublishMetadata(segmentMetadataUpdate.remoteLogSegmentId().topicIdPartition(),
segmentMetadataUpdate);
} finally {
lock.readLock().unlock();
}
}
@Override
- public void putRemotePartitionDeleteMetadata(RemotePartitionDeleteMetadata
remotePartitionDeleteMetadata)
+ public CompletableFuture<Void>
putRemotePartitionDeleteMetadata(RemotePartitionDeleteMetadata
remotePartitionDeleteMetadata)
throws RemoteStorageException {
Objects.requireNonNull(remotePartitionDeleteMetadata,
"remotePartitionDeleteMetadata can not be null");
lock.readLock().lock();
try {
ensureInitializedAndNotClosed();
-
doPublishMetadata(remotePartitionDeleteMetadata.topicIdPartition(),
remotePartitionDeleteMetadata);
+ return
doPublishMetadata(remotePartitionDeleteMetadata.topicIdPartition(),
remotePartitionDeleteMetadata);
} finally {
lock.readLock().unlock();
}
}
- private void doPublishMetadata(TopicIdPartition topicIdPartition,
RemoteLogMetadata remoteLogMetadata)
+ private CompletableFuture<Void> doPublishMetadata(TopicIdPartition
topicIdPartition,
+ RemoteLogMetadata
remoteLogMetadata)
throws RemoteStorageException {
log.debug("Publishing metadata for partition: [{}] with context:
[{}]", topicIdPartition, remoteLogMetadata);
try {
// Publish the message to the topic.
- RecordMetadata recordMetadata =
producerManager.publishMessage(remoteLogMetadata);
- // Wait until the consumer catches up with this offset. This will
ensure read-after-write consistency
- // semantics.
- consumerManager.waitTillConsumptionCatchesUp(recordMetadata);
+ CompletableFuture<RecordMetadata> produceFuture = new
CompletableFuture<>();
+ producerManager.publishMessage(remoteLogMetadata, produceFuture);
Review comment:
I was checking out a few other ways and passing future to the respective
functions. But the latest code does not require that to be passed and it can be
returned as you suggested.
--
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]
