You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "Owen-CH-Leung (via GitHub)" <gi...@apache.org> on 2023/08/01 14:07:50 UTC
[GitHub] [kafka] Owen-CH-Leung opened a new pull request, #14136: Add metadatacache into RemoteLogManager, and refactor all relevant codes
Owen-CH-Leung opened a new pull request, #14136:
URL: https://github.com/apache/kafka/pull/14136
PR in response to [KAFKA-15038](https://issues.apache.org/jira/browse/KAFKA-15038). This PR incorporates `MetaDataCache` into `RemoteLogManager` and refactor all other relevant codes
### Committer Checklist (excluded from commit message)
- [x] Verify design and implementation
- [x] Verify test coverage and CI build status
- [x] Verify documentation (including upgrade notes)
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15038: Add metadatacache into RemoteLogManager, and refactor all relevant codes [kafka]
Posted by "Owen-CH-Leung (via GitHub)" <gi...@apache.org>.
Owen-CH-Leung commented on PR #14136:
URL: https://github.com/apache/kafka/pull/14136#issuecomment-1787367129
I've just rebased trunk and resolved conflict. However when I run the test locally, I find that there're 2 failing tests at `kafka.admin.RemoteTopicCrudTest` on `testTopicDeletion`.
I tried to debug further and find that when switch to metadatacache & when `stopPartitions` is called, `metadataCache.contains(tp)` is evaluated to `False` and thus the rest of the deletion logic is not processed.
I believe this shouldn't be the case as the controller should update the metadataCache accordingly. Am I missing something that need to be changed ?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] Owen-CH-Leung commented on pull request #14136: Add metadatacache into RemoteLogManager, and refactor all relevant codes
Posted by "Owen-CH-Leung (via GitHub)" <gi...@apache.org>.
Owen-CH-Leung commented on PR #14136:
URL: https://github.com/apache/kafka/pull/14136#issuecomment-1663794337
Oh yes....Thanks and sorry for the dumb question!
I've fixed the CI failing test and all should be good for review now. Let me know your feedback.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15038: Add metadatacache into RemoteLogManager, and refactor all relevant codes [kafka]
Posted by "Owen-CH-Leung (via GitHub)" <gi...@apache.org>.
Owen-CH-Leung commented on PR #14136:
URL: https://github.com/apache/kafka/pull/14136#issuecomment-1816419916
@kamalcph
Thanks. Just rebased.
Btw when I rebased, I notice that there's a failing test `kafka.server.AssignmentsManagerTest.testAssignmentAggregation` at trunk. Maybe we need to fix that also
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] Owen-CH-Leung commented on a diff in pull request #14136: Add metadatacache into RemoteLogManager, and refactor all relevant codes
Posted by "Owen-CH-Leung (via GitHub)" <gi...@apache.org>.
Owen-CH-Leung commented on code in PR #14136:
URL: https://github.com/apache/kafka/pull/14136#discussion_r1285857919
##########
core/src/main/java/kafka/log/remote/RemoteLogManager.java:
##########
@@ -144,8 +145,9 @@ public class RemoteLogManager implements Closeable {
private final ConcurrentHashMap<TopicIdPartition, RLMTaskWithFuture> leaderOrFollowerTasks = new ConcurrentHashMap<>();
- // topic ids that are received on leadership changes, this map is cleared on stop partitions
- private final ConcurrentMap<TopicPartition, Uuid> topicPartitionIds = new ConcurrentHashMap<>();
+ // topic ids that are received on leadership changes & on stop partitions are stored in this cache
+ private final MetadataCache metaDataCache;
Review Comment:
Got it. Will revise accordingly
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] clolov commented on pull request #14136: Add metadatacache into RemoteLogManager, and refactor all relevant codes
Posted by "clolov (via GitHub)" <gi...@apache.org>.
clolov commented on PR #14136:
URL: https://github.com/apache/kafka/pull/14136#issuecomment-1678516932
To provide some clarity, I will just do one more pass for the MetadataCache itself, because I would like to confirm that it is updated when we would expect it to be (as per Divij's comment)
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] satishd commented on pull request #14136: KAFKA-15038: Add metadatacache into RemoteLogManager, and refactor all relevant codes
Posted by "satishd (via GitHub)" <gi...@apache.org>.
satishd commented on PR #14136:
URL: https://github.com/apache/kafka/pull/14136#issuecomment-1693159099
Right. This is good to have refactoring and not blocking any feature for 3.6.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] satishd commented on pull request #14136: KAFKA-15038: Add metadatacache into RemoteLogManager, and refactor all relevant codes
Posted by "satishd (via GitHub)" <gi...@apache.org>.
satishd commented on PR #14136:
URL: https://github.com/apache/kafka/pull/14136#issuecomment-1690950262
@divijvaidya Let me know if you still have any comments on this PR. We can merge this PR if there are no blocking comments.
There are a few unrelated test failures in jenkins jobs.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] divijvaidya commented on pull request #14136: KAFKA-15038: Add metadatacache into RemoteLogManager, and refactor all relevant codes
Posted by "divijvaidya (via GitHub)" <gi...@apache.org>.
divijvaidya commented on PR #14136:
URL: https://github.com/apache/kafka/pull/14136#issuecomment-1693154262
Hey folks
Let not target this change for 3.6 if we have doubts. It's anyways a perf improvement change. Let us get back to review this in detail after code freeze for 3.6 in another 2 weeks.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] Owen-CH-Leung commented on pull request #14136: Add metadatacache into RemoteLogManager, and refactor all relevant codes
Posted by "Owen-CH-Leung (via GitHub)" <gi...@apache.org>.
Owen-CH-Leung commented on PR #14136:
URL: https://github.com/apache/kafka/pull/14136#issuecomment-1681516264
> Overall LGTM. Can we hold this patch until #13947 lands? Otherwise, this patch needs one integration test to verify end-to-end change.
No problem. I can perform rebase after the PR was merged
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] Owen-CH-Leung commented on pull request #14136: Add metadatacache into RemoteLogManager, and refactor all relevant codes
Posted by "Owen-CH-Leung (via GitHub)" <gi...@apache.org>.
Owen-CH-Leung commented on PR #14136:
URL: https://github.com/apache/kafka/pull/14136#issuecomment-1685205287
@clolov @kamalcph I've merged base and refactored the test suite a bit. There're a few failed test but I believe those are unrelated to the changes I made. Could you please take a look and let me know your comments ?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] Owen-CH-Leung commented on a diff in pull request #14136: Add metadatacache into RemoteLogManager, and refactor all relevant codes
Posted by "Owen-CH-Leung (via GitHub)" <gi...@apache.org>.
Owen-CH-Leung commented on code in PR #14136:
URL: https://github.com/apache/kafka/pull/14136#discussion_r1303025695
##########
core/src/main/java/kafka/log/remote/RemoteLogManager.java:
##########
@@ -378,7 +372,7 @@ public void stopPartitions(Set<TopicPartition> topicPartitions,
remoteLogMetadataManager.onStopPartitions(topicIdPartitions);
if (delete) {
// NOTE: this#stopPartitions method is called when Replica state changes to Offline and ReplicaDeletionStarted
- topicPartitions.forEach(topicIdByPartitionMap::remove);
+// topicPartitions.forEach(topicIdByPartitionMap::remove);
Review Comment:
Thanks. Removed.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] Owen-CH-Leung commented on a diff in pull request #14136: Add metadatacache into RemoteLogManager, and refactor all relevant codes
Posted by "Owen-CH-Leung (via GitHub)" <gi...@apache.org>.
Owen-CH-Leung commented on code in PR #14136:
URL: https://github.com/apache/kafka/pull/14136#discussion_r1303026451
##########
core/src/main/java/kafka/log/remote/RemoteLogManager.java:
##########
@@ -285,19 +288,15 @@ public RemoteStorageManager storageManager() {
return remoteLogStorageManager;
}
+ public MetadataCache metadataCache() {
+ return metadataCache;
Review Comment:
Yes agree that we don't really need this. I've removed it
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] showuon commented on pull request #14136: KAFKA-15038: Add metadatacache into RemoteLogManager, and refactor all relevant codes
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on PR #14136:
URL: https://github.com/apache/kafka/pull/14136#issuecomment-1692686212
@clolov , thanks for the correction. :)
So, I think we need to have some solution for the problem in ZK's mode (which LISR will be sent out first before updateMetadataRequest), or we keep this PR open before ZK got removed in v4.0? Thoughts?
For me, I think we keep this PR open, which is the simplest way.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15038: Add metadatacache into RemoteLogManager, and refactor all relevant codes [kafka]
Posted by "Owen-CH-Leung (via GitHub)" <gi...@apache.org>.
Owen-CH-Leung commented on PR #14136:
URL: https://github.com/apache/kafka/pull/14136#issuecomment-1793384128
Hi all - I've rebased trunk. Can I ask for some feedback again ? Thanks
@clolov @showuon @satishd @kamalcph
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] Owen-CH-Leung commented on a diff in pull request #14136: Add metadatacache into RemoteLogManager, and refactor all relevant codes
Posted by "Owen-CH-Leung (via GitHub)" <gi...@apache.org>.
Owen-CH-Leung commented on code in PR #14136:
URL: https://github.com/apache/kafka/pull/14136#discussion_r1303027263
##########
core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala:
##########
@@ -3624,14 +3624,18 @@ class ReplicaManagerTest {
val remoteLogManagerConfig = new RemoteLogManagerConfig(config)
val mockLog = mock(classOf[UnifiedLog])
val brokerTopicStats = new BrokerTopicStats(java.util.Optional.of(KafkaConfig.fromProps(props)))
+ val mockMetadataCache = mock(classOf[MetadataCache])
val remoteLogManager = new RemoteLogManager(
remoteLogManagerConfig,
0,
TestUtils.tempRelativeDir("data").getAbsolutePath,
"clusterId",
time,
_ => Optional.of(mockLog),
- brokerTopicStats)
+ brokerTopicStats,
+ mockMetadataCache
+ )
+ when(remoteLogManager.metadataCache().getTopicId(topic)).thenReturn(topicId)
Review Comment:
Thanks. I've changed to use `mockMetadataCache` instead
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] clolov commented on pull request #14136: KAFKA-15038: Add metadatacache into RemoteLogManager, and refactor all relevant codes
Posted by "clolov (via GitHub)" <gi...@apache.org>.
clolov commented on PR #14136:
URL: https://github.com/apache/kafka/pull/14136#issuecomment-1691749594
I am possibly wrong, but I think that even though this piece of code sends LISR requests first
https://github.com/apache/kafka/blob/88d2c4460a1c8c8cf5dbcc9edb43f42fe898ca00/core/src/main/scala/kafka/controller/ControllerChannelManager.scala#L720-L722 there are never requests of different kinds at the same time based on the creation of a new batch in these scenarios:
* https://github.com/apache/kafka/blob/88d2c4460a1c8c8cf5dbcc9edb43f42fe898ca00/core/src/main/scala/kafka/controller/KafkaController.scala#L1117-L1127
* https://github.com/apache/kafka/blob/88d2c4460a1c8c8cf5dbcc9edb43f42fe898ca00/core/src/main/scala/kafka/controller/KafkaController.scala#L1237-L1239
* https://github.com/apache/kafka/blob/88d2c4460a1c8c8cf5dbcc9edb43f42fe898ca00/core/src/main/scala/kafka/controller/KafkaController.scala#L1396-L1400
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] Owen-CH-Leung commented on a diff in pull request #14136: Add metadatacache into RemoteLogManager, and refactor all relevant codes
Posted by "Owen-CH-Leung (via GitHub)" <gi...@apache.org>.
Owen-CH-Leung commented on code in PR #14136:
URL: https://github.com/apache/kafka/pull/14136#discussion_r1285865061
##########
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:
Got it. So can I say for `stopPartition` and `onLeadershipChange`, there's no need to update the metadatacache at all ? Is there any extra logic that should be addded into these 2 APIs ?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] Owen-CH-Leung commented on a diff in pull request #14136: Add metadatacache into RemoteLogManager, and refactor all relevant codes
Posted by "Owen-CH-Leung (via GitHub)" <gi...@apache.org>.
Owen-CH-Leung commented on code in PR #14136:
URL: https://github.com/apache/kafka/pull/14136#discussion_r1296606986
##########
core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala:
##########
@@ -3600,7 +3600,10 @@ class ReplicaManagerTest {
"clusterId",
time,
_ => Optional.of(mockLog),
- brokerTopicStats)
+ brokerTopicStats,
+ mock(classOf[MetadataCache])
Review Comment:
Sure. Moved the `mock(classOf[MetadataCache])` as variable now.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15038: Add metadatacache into RemoteLogManager, and refactor all relevant codes [kafka]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #14136:
URL: https://github.com/apache/kafka/pull/14136#issuecomment-2563280648
This PR is being marked as stale since it has not had any activity in 90 days. If you
would like to keep this PR alive, please leave a comment asking for a review. If the PR has
merge conflicts, update it with the latest from the base branch.
<p>
If you are having difficulty finding a reviewer, please reach out on the
[mailing list](https://kafka.apache.org/contact).
<p>
If this PR is no longer valid or desired, please feel free to close it. If no activity
occurs in the next 30 days, it will be automatically closed.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] divijvaidya commented on pull request #14136: Add metadatacache into RemoteLogManager, and refactor all relevant codes
Posted by "divijvaidya (via GitHub)" <gi...@apache.org>.
divijvaidya commented on PR #14136:
URL: https://github.com/apache/kafka/pull/14136#issuecomment-1661714026
I am not sure I understand why we need to access the factory method.
The metadataCache instance is created on broker startup. It is then passed to ReplicaManager. RemoteLogManager is also created on start-up and you can pass this metadataCache in the constructor of RemoteLogManager at https://github.com/apache/kafka/blob/8aaf7daff393f2b26438fb7fe28016e06e23558c/core/src/main/scala/kafka/server/BrokerServer.scala#L573 . I observed that you are already doing this in the change in this PR.
Once we have an instance inside RemoteLogManager, you can call Zk/Kraft specific functions using: `val zkMetadataCache = metadataCache.asInstanceOf[ZkMetadataCache]`
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] Owen-CH-Leung commented on a diff in pull request #14136: Add metadatacache into RemoteLogManager, and refactor all relevant codes
Posted by "Owen-CH-Leung (via GitHub)" <gi...@apache.org>.
Owen-CH-Leung commented on code in PR #14136:
URL: https://github.com/apache/kafka/pull/14136#discussion_r1292378925
##########
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:
Thanks. I've removed this logic
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] clolov commented on a diff in pull request #14136: Add metadatacache into RemoteLogManager, and refactor all relevant codes
Posted by "clolov (via GitHub)" <gi...@apache.org>.
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] showuon commented on a diff in pull request #14136: Add metadatacache into RemoteLogManager, and refactor all relevant codes
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on code in PR #14136:
URL: https://github.com/apache/kafka/pull/14136#discussion_r1302927787
##########
core/src/main/java/kafka/log/remote/RemoteLogManager.java:
##########
@@ -285,19 +288,15 @@ public RemoteStorageManager storageManager() {
return remoteLogStorageManager;
}
+ public MetadataCache metadataCache() {
+ return metadataCache;
Review Comment:
Why do we need this? For test only? I think the test doesn't have to use it. See my comment below.
##########
core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala:
##########
@@ -3624,14 +3624,18 @@ class ReplicaManagerTest {
val remoteLogManagerConfig = new RemoteLogManagerConfig(config)
val mockLog = mock(classOf[UnifiedLog])
val brokerTopicStats = new BrokerTopicStats(java.util.Optional.of(KafkaConfig.fromProps(props)))
+ val mockMetadataCache = mock(classOf[MetadataCache])
val remoteLogManager = new RemoteLogManager(
remoteLogManagerConfig,
0,
TestUtils.tempRelativeDir("data").getAbsolutePath,
"clusterId",
time,
_ => Optional.of(mockLog),
- brokerTopicStats)
+ brokerTopicStats,
+ mockMetadataCache
+ )
+ when(remoteLogManager.metadataCache().getTopicId(topic)).thenReturn(topicId)
Review Comment:
What's the difference between `mockMetadataCache` and `remoteLogManager.metadataCache()`? If no difference, could we use the former one and remove the `remoteLogManager#metadataCache()` method?
##########
core/src/main/java/kafka/log/remote/RemoteLogManager.java:
##########
@@ -378,7 +372,7 @@ public void stopPartitions(Set<TopicPartition> topicPartitions,
remoteLogMetadataManager.onStopPartitions(topicIdPartitions);
if (delete) {
// NOTE: this#stopPartitions method is called when Replica state changes to Offline and ReplicaDeletionStarted
- topicPartitions.forEach(topicIdByPartitionMap::remove);
+// topicPartitions.forEach(topicIdByPartitionMap::remove);
Review Comment:
Should we delete this line directly?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15038: Add metadatacache into RemoteLogManager, and refactor all relevant codes [kafka]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] closed pull request #14136: KAFKA-15038: Add metadatacache into RemoteLogManager, and refactor all relevant codes
URL: https://github.com/apache/kafka/pull/14136
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15038: Add metadatacache into RemoteLogManager, and refactor all relevant codes [kafka]
Posted by "kamalcph (via GitHub)" <gi...@apache.org>.
kamalcph commented on PR #14136:
URL: https://github.com/apache/kafka/pull/14136#issuecomment-1815724658
@Owen-CH-Leung
Can you please rebase your PR against trunk? I'll review the PR in 2 days.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] Owen-CH-Leung commented on pull request #14136: Add metadatacache into RemoteLogManager, and refactor all relevant codes
Posted by "Owen-CH-Leung (via GitHub)" <gi...@apache.org>.
Owen-CH-Leung commented on PR #14136:
URL: https://github.com/apache/kafka/pull/14136#issuecomment-1661525755
Hello @divijvaidya , I'm still in the process of writing the PR and would like to ask something :
For this ticket, in order to reuse the metadata cache inside broker (which I believe is already instantiated when the broker server is set up), I think I need to access the following factory method code snippet :
https://github.com/apache/kafka/blob/trunk/core/src/main/scala/kafka/server/MetadataCache.scala#L114-L125
However, this method will always return a new instance of metadatacache instead of the one being instantiated inside broker. So does it make sense to change it to a Singleton class ?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] showuon commented on pull request #14136: KAFKA-15038: Add metadatacache into RemoteLogManager, and refactor all relevant codes
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on PR #14136:
URL: https://github.com/apache/kafka/pull/14136#issuecomment-1693159129
> Let not target this change for 3.6 if we have doubts. It's anyways a perf improvement change. Let us get back to review this in detail after code freeze for 3.6 in another 2 weeks.
Agree.
> For stopPartition, I found not only Controller will trigger it, the client can send deleteTopic to trigger it.
I found I was wrong. It seems that the deleteTopic triggering `stopPartition` will only happen in KRaft, not ZK mode. But still unless we can 100% make sure there won't be any corner case, the time gap between LISR and UMR is still a concern to me.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] showuon commented on pull request #14136: KAFKA-15038: Add metadatacache into RemoteLogManager, and refactor all relevant codes
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on PR #14136:
URL: https://github.com/apache/kafka/pull/14136#issuecomment-1693132900
@clolov , thanks for the detailed analysis. I admit I didn't check it so detailed last time. I just saw in `RLM#onLeadershipChange`, we did use `metadataCache`, so we should make sure `metadataCache` comes first before LISR. Had another look, we can still pass the `topicIDs` into `RLM#onLeadershipChange` to bypass it. So it looks fine.
For `fetchRemoteLogSegmentMetadata` and `findOffsetByTimestamp`, I have no concern since like you said, they are used in consumer, which means the metadataCache must be have updated. Although they can also be used by adminClient, some delay for the adminClient usage should be acceptable.
For `stopPartition`, I found not only Controller will trigger it, the client can send `deleteTopic` to trigger it. In this case, we might delete less partitions than expected like this case:
1. LISR arrived in RLM, created some RLM tasks for the leader partitions
2. delete topic comes in, miss the new added partitions in step (1)
3. updateMetadata arrived, updated the cache
I'm not quite sure if it's possible. TBH, I'm a little worried about introducing potential bugs due to some edge case we didn't know, due to we already confirmed there are some time gap between LISR and UMR. That's my thought. I'd also like to hear other reviewers' comments.
Thanks again for the detailed analysis.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] showuon commented on pull request #14136: KAFKA-15038: Add metadatacache into RemoteLogManager, and refactor all relevant codes
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on PR #14136:
URL: https://github.com/apache/kafka/pull/14136#issuecomment-1691622348
> I want to ensure that there are no race conditions between LISR and updateMetadata requests from controller i.e. validate that metadata cache is updated during LISR request. Give me a day to validate that.
This is a really great point. I've taken a look:
- For KRaft, the metadata cache is synced up when `RLM#onLeadershipChange`.
- For ZK, the controller will send out LISR and updateMetadataRequest basically at the same time (but LISR is enqueued first). I think in ZK, there might be chances the topicID is still not updated in metadataCache before `RLM#onLeadershipChange` is invoked.
I think we need to have some solution for it, or we revert this change before ZK got removed in v4.0? Thoughts?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] kamalcph commented on a diff in pull request #14136: Add metadatacache into RemoteLogManager, and refactor all relevant codes
Posted by "kamalcph (via GitHub)" <gi...@apache.org>.
kamalcph commented on code in PR #14136:
URL: https://github.com/apache/kafka/pull/14136#discussion_r1296153252
##########
core/src/test/java/kafka/log/remote/RemoteLogManagerTest.java:
##########
@@ -802,41 +817,25 @@ private void verifyInCache(TopicIdPartition... topicIdPartitions) {
});
}
- private void verifyNotInCache(TopicIdPartition... topicIdPartitions) {
- Arrays.stream(topicIdPartitions).forEach(topicIdPartition -> {
- assertThrows(KafkaException.class, () ->
- remoteLogManager.fetchRemoteLogSegmentMetadata(topicIdPartition.topicPartition(), 0, 0L));
- });
- }
-
@Test
void testTopicIdCacheUpdates() throws RemoteStorageException {
Review Comment:
Let's remove this test.
##########
core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala:
##########
@@ -3600,7 +3600,10 @@ class ReplicaManagerTest {
"clusterId",
time,
_ => Optional.of(mockLog),
- brokerTopicStats)
+ brokerTopicStats,
+ mock(classOf[MetadataCache])
Review Comment:
Could you please extract the `mock` as variable before using it to keep it uniform?
```
val metadataCache = mock(classOf[MetadataCache])
...
when(metadataCache.getTopicId(topic))...
```
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15038: Add metadatacache into RemoteLogManager, and refactor all relevant codes [kafka]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #14136:
URL: https://github.com/apache/kafka/pull/14136#issuecomment-2614791117
This PR has been closed since it has not had any activity in 120 days. If you feel like this
was a mistake, or you would like to continue working on it, please feel free to re-open the
PR and ask for a review.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] Owen-CH-Leung commented on a diff in pull request #14136: Add metadatacache into RemoteLogManager, and refactor all relevant codes
Posted by "Owen-CH-Leung (via GitHub)" <gi...@apache.org>.
Owen-CH-Leung commented on code in PR #14136:
URL: https://github.com/apache/kafka/pull/14136#discussion_r1296606546
##########
core/src/test/java/kafka/log/remote/RemoteLogManagerTest.java:
##########
@@ -802,41 +817,25 @@ private void verifyInCache(TopicIdPartition... topicIdPartitions) {
});
}
- private void verifyNotInCache(TopicIdPartition... topicIdPartitions) {
- Arrays.stream(topicIdPartitions).forEach(topicIdPartition -> {
- assertThrows(KafkaException.class, () ->
- remoteLogManager.fetchRemoteLogSegmentMetadata(topicIdPartition.topicPartition(), 0, 0L));
- });
- }
-
@Test
void testTopicIdCacheUpdates() throws RemoteStorageException {
Review Comment:
Sure. Removed the test
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] divijvaidya commented on a diff in pull request #14136: Add metadatacache into RemoteLogManager, and refactor all relevant codes
Posted by "divijvaidya (via GitHub)" <gi...@apache.org>.
divijvaidya commented on code in PR #14136:
URL: https://github.com/apache/kafka/pull/14136#discussion_r1283058109
##########
core/src/main/java/kafka/log/remote/RemoteLogManager.java:
##########
@@ -144,8 +145,9 @@ public class RemoteLogManager implements Closeable {
private final ConcurrentHashMap<TopicIdPartition, RLMTaskWithFuture> leaderOrFollowerTasks = new ConcurrentHashMap<>();
- // topic ids that are received on leadership changes, this map is cleared on stop partitions
- private final ConcurrentMap<TopicPartition, Uuid> topicPartitionIds = new ConcurrentHashMap<>();
+ // topic ids that are received on leadership changes & on stop partitions are stored in this cache
+ private final MetadataCache metaDataCache;
Review Comment:
nit
s\metaData\metadata
(everywhere)
##########
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:
The idea of replacing this cache with Metadata cache is that the existing metadata cache will serve as the source of truth. It should already be updated to contains the correct topics Id <-> topicPartition mapping irrespective of whether tiered storage is turned on or not. TS code should only access this cache and not modify it in any way. The modification (adding/removal) should already exist is rest of the normal path Kafka code. (If it doesn't then we need to fix it).
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] Owen-CH-Leung commented on a diff in pull request #14136: Add metadatacache into RemoteLogManager, and refactor all relevant codes
Posted by "Owen-CH-Leung (via GitHub)" <gi...@apache.org>.
Owen-CH-Leung commented on code in PR #14136:
URL: https://github.com/apache/kafka/pull/14136#discussion_r1292378582
##########
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:
@clolov Thanks a lot for your comment. I've incorporated your comments
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] divijvaidya commented on pull request #14136: KAFKA-15038: Add metadatacache into RemoteLogManager, and refactor all relevant codes
Posted by "divijvaidya (via GitHub)" <gi...@apache.org>.
divijvaidya commented on PR #14136:
URL: https://github.com/apache/kafka/pull/14136#issuecomment-1691361285
> @divijvaidya Let me know if you still have any comments on this PR. We can merge this PR if there are no blocking comments.
>
> There are a few unrelated test failures in jenkins jobs.
I want to ensure that there are no race conditions between LISR and updateMetadata requests from controller i.e. validate that metadata cache is updated during LISR request. Give me a day to validate that.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] clolov commented on pull request #14136: KAFKA-15038: Add metadatacache into RemoteLogManager, and refactor all relevant codes
Posted by "clolov (via GitHub)" <gi...@apache.org>.
clolov commented on PR #14136:
URL: https://github.com/apache/kafka/pull/14136#issuecomment-1693019922
Sorry, I do not understand the concern. Let me explain my reasoning.
Based on https://github.com/apache/kafka/pull/14136#issuecomment-1691749594 the controller sends blocking requests in the order LISR, UpdateMetadata, StopReplica (which eventually trickle down to `stopPartitions`).
Let's consider the usage of the pseudo metadata cache (from now on referred to as the concurrent map) in the RemoteLogManager before the change proposed in this pull request.
`onLeadershipChange` does not lookup topic identifiers in the concurrent map, it only populates it with entries.
The places where topic identifiers are looked up are in `stopPartitions`, `fetchRemoteLogSegmentMetadata`, and `findOffsetByTimestamp`.
`stopPartitions` is only called last in the list of requests sent by the controller. This cannot arrive before the previous update metadata has propagated due to the blocking behaviour of the controller sending requests.
Both `fetchRemoteLogSegmentMetadata` and `findOffsetByTimestamp` are on the read path of Tiered Storage i.e. they are triggered by consumers. In order for consumers to continue consuming after a leadership change they need to get the latest metadata, as such update metadata requests need to have propagated through the cluster before brokers start returning the new leaders.
Now let's change the concurrent map with the metadata cache. I believe the same logic still holds true i.e. an update metadata request containing the new topic identifiers needs to have propagated through the cluster before any requests using said topic identifiers can start to be serviced by the broker.
Am I wrong somewhere in my reasoning?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] Owen-CH-Leung commented on pull request #14136: KAFKA-15038: Add metadatacache into RemoteLogManager, and refactor all relevant codes
Posted by "Owen-CH-Leung (via GitHub)" <gi...@apache.org>.
Owen-CH-Leung commented on PR #14136:
URL: https://github.com/apache/kafka/pull/14136#issuecomment-1716985288
Thanks all for your comments. I'll resolve conflict and work on this again after 3.6 is released
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15038: Add metadatacache into RemoteLogManager, and refactor all relevant codes [kafka]
Posted by "kamalcph (via GitHub)" <gi...@apache.org>.
kamalcph commented on PR #14136:
URL: https://github.com/apache/kafka/pull/14136#issuecomment-1818250640
> RLM#onLeadershipChange, we did use metadataCache, so we should make sure metadataCache comes first before LISR. Had another look, we can still pass the topicIDs into RLM#onLeadershipChange to bypass it. So it looks fine.
We are still using metadata-cache in RLM#onLeadershipChange, we have to ensure that the `MetadataCache` is updated before handling the LeaderAndIsrRequest (LISR). Otherwise, we can follow the suggestion posted by @showuon. Also, there are existing test failures related to remote storage, can you please take a look?
```
Build / JDK 11 and Scala 2.13 / executeTieredStorageTest(String).quorum=kraft – org.apache.kafka.tiered.storage.integration.ReassignReplicaShrinkTest
Build / JDK 11 and Scala 2.13 / testSendOffsetsWithGroupMetadata(String).quorum=kraft – org.apache.kafka.tiered.storage.integration.TransactionsWithTieredStoreTest
Build / JDK 11 and Scala 2.13 / testBasicTransactions(String).quorum=kraft – org.apache.kafka.tiered.storage.integration.TransactionsWithTieredStoreTest
Build / JDK 11 and Scala 2.13 / testSendOffsetsWithGroupId(String).quorum=kraft – org.apache.kafka.tiered.storage.integration.TransactionsWithTieredStoreTest
Build / JDK 11 and Scala 2.13 / testDelayedFetchIncludesAbortedTransaction(String).quorum=kraft – org.apache.kafka.tiered.storage.integration.TransactionsWithTieredStoreTest
Build / JDK 21 and Scala 2.13 / testAbortTransactionTimeout(String).quorum=kraft – org.apache.kafka.tiered.storage.integration.TransactionsWithTieredStoreTest
```
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org