[improve][broker] PIP-433: Ensure topic creation before starting GEO #24652
Conversation
Signed-off-by: Zixuan Liu <nodeces@gmail.com>
nodece
force-pushed
the
improve-topic-creation-and-partition-update-before-starting-geo
branch
from
9f91e51 to
a613da2
Compare
| } | ||
| // Set useFallbackForNonPIP344Brokers to true when mix of PIP-344 and non-PIP-344 brokers are used, it | ||
| // can still work. | ||
| return client.getLookup().getPartitionedTopicMetadata(baseTopicName, false, true) |
There was a problem hiding this comment.
Since we have pulsarAdmin object, can we use pulsar-admin.topics.getPartitionedTopicMetadataAsync here?
There was a problem hiding this comment.
Good point, I will refactor this logic.
| .getClusterAsync(remoteCluster)) | ||
| .thenAccept((clusterData) -> { | ||
| PulsarClient replicationClient = brokerService.getReplicationClient(remoteCluster, clusterData); | ||
| PulsarAdmin replicationAdmin = brokerService.getClusterPulsarAdmin(remoteCluster, clusterData); |
There was a problem hiding this comment.
brokerService.getClusterPulsarAdmin may throw a runtime exception. Should we consider it?
There was a problem hiding this comment.
I will move this logic from PersistentTopic to the replicator.
brokerService.getClusterPulsarAdminmay throw a runtime exception. Should we consider it?
If an exception is thrown, this future will be complete with an exception.
|
|
||
| Awaitility.await().untilAsserted(() -> { | ||
| if (isPartitioned){ | ||
| assertThat(admin2.topics().getPartitionedTopicList(ns)).contains(tp); |
There was a problem hiding this comment.
We'd better check that partitions are also created
| } | ||
|
|
||
| @Test | ||
| public void testReplicatorWhenPartitionCountsDiffer() throws Exception { |
| } | ||
| assertEquals(receive.getValue(), "msg-p2-1"); | ||
| } | ||
| } |
There was a problem hiding this comment.
Could you also add another test?
- Both clusters have different topic-auto-creation rules
- The partitioned topic will by created by replicator at remote-side
PIP: #24485
Motivation
In a GEO replication scenario, if the remote cluster does not have the replicated topic and the auto-creation type differs between the local and remote clusters, message replication may fail. To ensure seamless replication, the topic metadata must be properly synchronized across clusters.
This is part of PIP-433.
#24136 is the same as this PR.
Modifications
When both the local and remote partitioned topic metadata indicate
partitions=0, this means the topic is non-partitioned. In this case, the local cluster sends a non-partitioned topic creation request to the remote cluster.If the local partitioned topic metadata has
partitions>0, this means the topic is partitioned:partitions=0, the local cluster sends a partitioned topic creation request to the remote cluster.Documentation
docdoc-requireddoc-not-neededdoc-complete