clolov commented on code in PR #14136:
URL: https://github.com/apache/kafka/pull/14136#discussion_r1288687969
##########
core/src/main/java/kafka/log/remote/RemoteLogManager.java:
##########
@@ -347,15 +353,16 @@ public void onLeadershipChange(Set<Partition>
partitionsBecomeLeader,
public void stopPartitions(TopicPartition topicPartition, boolean delete) {
if (delete) {
// Delete from internal datastructures only if it is to be deleted.
- Uuid topicIdPartition = topicPartitionIds.remove(topicPartition);
- LOGGER.debug("Removed partition: {} from topicPartitionIds",
topicIdPartition);
+ Map<String, Uuid> mapping = metaDataCache.topicNamesToIds();
+ Uuid topicIdPartition = mapping.remove(topicPartition.topic());
Review Comment:
I don't think there is anything to be done here. The topic partitions should
disappear once the controller lets us know they should.
##########
core/src/main/java/kafka/log/remote/RemoteLogManager.java:
##########
@@ -280,13 +285,18 @@ public RemoteStorageManager storageManager() {
return remoteLogStorageManager;
}
+ public MetadataCache metadataCache() {
+ return metaDataCache;
+ }
+
private Stream<Partition> filterPartitions(Set<Partition> partitions) {
// We are not specifically checking for internal topics etc here as
`log.remoteLogEnabled()` already handles that.
return partitions.stream().filter(partition ->
partition.log().exists(UnifiedLog::remoteLogEnabled));
}
private void cacheTopicPartitionIds(TopicIdPartition topicIdPartition) {
- Uuid previousTopicId =
topicPartitionIds.put(topicIdPartition.topicPartition(),
topicIdPartition.topicId());
+ Map<String, Uuid> mapping = metaDataCache.topicNamesToIds();
Review Comment:
Adding to Divij's comment - this is an unmodifiable map. Even if you wanted
to you wouldn't be able to add to it:
```
def topicNamesToIds(): util.Map[String, Uuid] = {
Collections.unmodifiableMap(metadataSnapshot.topicIds.asJava)
}
```
--
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: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
