coderzc commented on code in PR #25312:
URL: https://github.com/apache/pulsar/pull/25312#discussion_r2923749036
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java:
##########
@@ -3458,6 +3478,78 @@ private CompletableFuture<PartitionedTopicMetadata>
createDefaultPartitionedTopi
});
}
+ private CompletableFuture<TopicExistsInfo>
getRemotePartitionedTopicMetadataForAutoCreation(
+ TopicName topicName, Optional<Policies> policies) {
+ if (!pulsar.getConfig().isCreateTopicToRemoteClusterForReplication()) {
+ return
CompletableFuture.completedFuture(TopicExistsInfo.newTopicNotExists());
+ }
+ if (topicName.isPartitioned() || !topicName.isPersistent() ||
policies.isEmpty()) {
+ return
CompletableFuture.completedFuture(TopicExistsInfo.newTopicNotExists());
+ }
+ Set<String> replicationClusters = policies.get().replication_clusters;
+ if (replicationClusters == null || replicationClusters.isEmpty()) {
+ return
CompletableFuture.completedFuture(TopicExistsInfo.newTopicNotExists());
+ }
+ String localCluster = pulsar.getConfiguration().getClusterName();
+ if (!replicationClusters.contains(localCluster) ||
replicationClusters.size() <= 1) {
+ return
CompletableFuture.completedFuture(TopicExistsInfo.newTopicNotExists());
+ }
+ List<String> remoteClusters = replicationClusters.stream()
+ .filter(cluster -> !cluster.equals(localCluster))
+ .sorted()
+ .toList();
+ return findRemoteTopicMetadataForAutoCreation(topicName,
remoteClusters, 0, null);
+ }
+
+ private CompletableFuture<TopicExistsInfo>
findRemoteTopicMetadataForAutoCreation(
+ TopicName topicName, List<String> remoteClusters, int index,
Throwable errOccurred) {
+ if (index >= remoteClusters.size()) {
+ if (errOccurred != null) {
+ log.error("[{}] Failed to check remote topic partitioned
metadata on cluster {}. Fallback to "
+ + "default auto topic creation policy.",
+ topicName, remoteClusters, errOccurred);
+ }
+ return
CompletableFuture.completedFuture(TopicExistsInfo.newTopicNotExists());
+ }
+ final String remoteCluster = remoteClusters.get(index);
+ return
pulsar.getPulsarResources().getClusterResources().getClusterAsync(remoteCluster)
+ .thenCompose(clusterData -> {
+ if (clusterData.isEmpty()) {
+ log.warn("[{}] Skip checking remote cluster {} because
cluster data is missing",
+ topicName, remoteCluster);
+ return findRemoteTopicMetadataForAutoCreation(topicName,
remoteClusters, index + 1, null);
+ }
+ PulsarClient client = getReplicationClient(remoteCluster,
clusterData);
+ CompletableFuture<TopicExistsInfo> future = new
CompletableFuture<>();
+ client.getPartitionsForTopic(topicName.toString(),
false).handle((topics, t) -> {
Review Comment:
This new fallback path can still fail with a `NullPointerException` when the
remote lookup returns an exception.
Inside `client.getPartitionsForTopic(...).handle((topics, t) -> { ... })`,
the code handles `t != null`, but then continues to evaluate `topics.isEmpty()`
afterwards. In the exceptional case, `topics` is null, so this turns the
intended “fallback to the local default auto-creation policy” path into an
unexpected failure.
Please return immediately after completing the future in the exception
branch, or restructure the handler so the success path only runs when `t ==
null`.
--
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]
