Skip to content

[fix][broker] Reject create non-existent persistent partitions #19086

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
eolivelli merged 20 commits into from
Jan 9, 2023
Merged

[fix][broker] Reject create non-existent persistent partitions #19086

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
20 commits
Select commit Hold shift + click to select a range
73547fb
[fix][broker] Reject create non existent partitions.
mattisonchao Dec 28, 2022
a2350f1
Revert format change
mattisonchao Dec 28, 2022
3c7b9e9
Add partition number test
mattisonchao Dec 28, 2022
e3763c4
Fix test
mattisonchao Dec 28, 2022
75e88b0
Apply comment
mattisonchao Dec 28, 2022
5d06688
Fix checkstyle
mattisonchao Dec 28, 2022
390f12e
Fix checkstyle
mattisonchao Dec 28, 2022
2d68680
Fix unexpected behaviour
mattisonchao Dec 29, 2022
8266e89
Fix checkstyle
mattisonchao Dec 29, 2022
3c5e3aa
Fix some failed test
mattisonchao Dec 29, 2022
9f49db0
Rollback some logic
mattisonchao Dec 30, 2022
c77a774
Add non-persistent topic test
mattisonchao Dec 30, 2022
6fce4df
Fix checkstyle
mattisonchao Dec 30, 2022
655a6ab
Revert test fix
mattisonchao Dec 30, 2022
5b460cb
Revert some logic to help review
mattisonchao Dec 30, 2022
3d59543
Remove unused import
mattisonchao Dec 30, 2022
bf67473
Fix checkstyle
mattisonchao Dec 30, 2022
68f2bd5
Fix failed test
mattisonchao Dec 30, 2022
58f516f
Fix test
mattisonchao Jan 6, 2023
5e2c937
Apply comment
mattisonchao Jan 9, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
266 changes: 130 additions & 136 deletions pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -411,67 +411,85 @@ protected CompletableFuture<Void> internalCreateNonPartitionedTopicAsync(boolean
* recreate them at application so, newly created producers and consumers can connect to newly added partitions as
* well. Therefore, it can violate partition ordering at producers until all producers are restarted at application.
*
* @param numPartitions
* @param expectPartitions
* @param updateLocalTopicOnly
* @param authoritative
* @param force
*/
protected CompletableFuture<Void> internalUpdatePartitionedTopicAsync(int numPartitions,
protected CompletableFuture<Void> internalUpdatePartitionedTopicAsync(int expectPartitions,
boolean updateLocalTopicOnly,
boolean authoritative, boolean force) {
if (numPartitions <= 0) {
return FutureUtil.failedFuture(new RestException(Status.NOT_ACCEPTABLE,
"Number of partitions should be more than 0"));
if (expectPartitions <= 0) {
return FutureUtil.failedFuture(
new RestException(Status.NOT_ACCEPTABLE, "Number of partitions should be more than 0"));
}
return validateTopicOwnershipAsync(topicName, authoritative)
.thenCompose(__ -> validateTopicPolicyOperationAsync(topicName, PolicyName.PARTITION,
PolicyOperation.WRITE))
.thenCompose(__ ->
validateTopicPolicyOperationAsync(topicName, PolicyName.PARTITION, PolicyOperation.WRITE))
.thenCompose(__ -> {
if (!updateLocalTopicOnly && !force) {
return validatePartitionTopicUpdateAsync(topicName.getLocalName(), numPartitions);
return validatePartitionTopicUpdateAsync(topicName.getLocalName(), expectPartitions);
} else {
return CompletableFuture.completedFuture(null);
}
}).thenCompose(__ -> pulsar().getBrokerService().fetchPartitionedTopicMetadataAsync(topicName))
.thenCompose(topicMetadata -> {
final int maxPartitions = pulsar().getConfig().getMaxNumPartitionsPerPartitionedTopic();
if (maxPartitions > 0 && numPartitions > maxPartitions) {
if (maxPartitions > 0 && expectPartitions > maxPartitions) {
throw new RestException(Status.NOT_ACCEPTABLE,
"Number of partitions should be less than or equal to " + maxPartitions);
}
// Only do the validation if it's the first hop.
if (topicName.isGlobal() && isNamespaceReplicated(topicName.getNamespaceObject())) {
return getNamespaceReplicatedClustersAsync(topicName.getNamespaceObject())
.thenApply(clusters -> {
if (!clusters.contains(pulsar().getConfig().getClusterName())) {
log.error("[{}] local cluster is not part of replicated cluster for namespace {}",
clientAppId(), topicName);
throw new RestException(Status.FORBIDDEN, "Local cluster is not part of replicate"
+ " cluster list");
}
return clusters;
})
.thenCompose(clusters ->
tryCreateExtendedPartitionsAsync(topicMetadata.partitions, numPartitions)
.thenApply(ignore -> clusters))
.thenCompose(clusters -> createSubscriptions(topicName, numPartitions, force).thenApply(
ignore -> clusters))
.thenCompose(clusters -> {
if (!updateLocalTopicOnly) {
return updatePartitionInOtherCluster(numPartitions, clusters)
.thenCompose(v -> namespaceResources().getPartitionedTopicResources()
.updatePartitionedTopicAsync(topicName, p ->
new PartitionedTopicMetadata(numPartitions,
p.properties)
));
} else {
return CompletableFuture.completedFuture(null);
}
});
} else {
return tryCreateExtendedPartitionsAsync(topicMetadata.partitions, numPartitions)
.thenCompose(ignore -> updatePartitionedTopic(topicName, numPartitions, force));
final PulsarAdmin adminClient;
try {
adminClient = pulsar().getAdminClient();
} catch (PulsarServerException e) {
throw new RuntimeException(e);
}
return adminClient.topics().getListAsync(topicName.getNamespace())
.thenCompose(topics -> {
long existPartitions = topics.stream()
.filter(t -> TopicName.get(t).getPartitionedTopicName()
.equals(topicName.getPartitionedTopicName()))
.count();
if (existPartitions >= expectPartitions) {
throw new RestException(Status.CONFLICT,
"Number of new partitions must be greater than existing number of partitions");
}
// Only do the validation if it's the first hop.
if (topicName.isGlobal() && isNamespaceReplicated(topicName.getNamespaceObject())) {
return getNamespaceReplicatedClustersAsync(topicName.getNamespaceObject())
.thenApply(clusters -> {
if (!clusters.contains(pulsar().getConfig().getClusterName())) {
log.error("[{}] local cluster is not part of"
+ " replicated cluster for namespace {}",
clientAppId(), topicName);
throw new RestException(Status.FORBIDDEN,
"Local cluster is not part of replicate cluster list");
}
return clusters;
})
.thenCompose(clusters ->
tryCreatePartitionsAsync(expectPartitions)
.thenApply(ignore -> clusters))
.thenCompose(clusters -> {
if (!updateLocalTopicOnly) {
return namespaceResources().getPartitionedTopicResources()
.updatePartitionedTopicAsync(topicName, p ->
new PartitionedTopicMetadata(expectPartitions,
p.properties))
.thenCompose(__ ->
updatePartitionInOtherCluster(expectPartitions,
clusters));
} else {
return CompletableFuture.completedFuture(null);
}
}).thenCompose(clusters -> createSubscriptions(topicName,
expectPartitions));
} else {
return tryCreatePartitionsAsync(expectPartitions)
.thenCompose(ignore -> updatePartitionedTopic(topicName, expectPartitions));
}
});
});
}

Expand Down Expand Up @@ -4363,124 +4381,100 @@ private PersistentReplicator getReplicatorReference(String replName, PersistentT
}
}

private CompletableFuture<Void> updatePartitionedTopic(TopicName topicName, int numPartitions, boolean force) {
CompletableFuture<Void> result = new CompletableFuture<>();
createSubscriptions(topicName, numPartitions, force).thenCompose(__ -> {
CompletableFuture<Void> future = namespaceResources().getPartitionedTopicResources()
.updatePartitionedTopicAsync(topicName, p ->
new PartitionedTopicMetadata(numPartitions, p.properties));
future.exceptionally(ex -> {
// If the update operation fails, clean up the partitions that were created
getPartitionedTopicMetadataAsync(topicName, false, false).thenAccept(metadata -> {
int oldPartition = metadata.partitions;
for (int i = oldPartition; i < numPartitions; i++) {
topicResources().deletePersistentTopicAsync(topicName.getPartition(i)).exceptionally(ex1 -> {
log.warn("[{}] Failed to clean up managedLedger {}", clientAppId(), topicName,
ex1.getCause());
return null;
});
}
}).exceptionally(e -> {
log.warn("[{}] Failed to clean up managedLedger", topicName, e);
return null;
});
private CompletableFuture<Void> updatePartitionedTopic(TopicName topicName, int expectPartitions) {
CompletableFuture<Void> future = namespaceResources().getPartitionedTopicResources()
.updatePartitionedTopicAsync(topicName, p ->
new PartitionedTopicMetadata(expectPartitions, p.properties));
future.exceptionally(ex -> {
// If the update operation fails, clean up the partitions that were created
getPartitionedTopicMetadataAsync(topicName, false, false)
.thenAccept(metadata -> {
int oldPartition = metadata.partitions;
for (int i = oldPartition; i < expectPartitions; i++) {
topicResources().deletePersistentTopicAsync(topicName.getPartition(i)).exceptionally(ex1 -> {
Comment thread
mattisonchao marked this conversation as resolved.
log.warn("[{}] Failed to clean up managedLedger {}", clientAppId(), topicName,
ex1.getCause());
return null;
});
}
}).exceptionally(e -> {
log.warn("[{}] Failed to clean up managedLedger", topicName, e);
return null;
});
return future;
}).thenAccept(__ -> result.complete(null)).exceptionally(ex -> {
result.completeExceptionally(ex);
return null;
});
return result;
return future.thenCompose(__ -> createSubscriptions(topicName, expectPartitions));
}

/**
* It creates subscriptions for new partitions of existing partitioned-topics.
*
* @param topicName : topic-name: persistent://prop/cluster/ns/topic
* @param numPartitions : number partitions for the topics
* @param ignoreConflictException : If true, ignore ConflictException: subscription already exists for topic
* @param expectPartitions : number of expected partitions
*
*/
private CompletableFuture<Void> createSubscriptions(TopicName topicName, int numPartitions,
boolean ignoreConflictException) {
private CompletableFuture<Void> createSubscriptions(TopicName topicName, int expectPartitions) {
CompletableFuture<Void> result = new CompletableFuture<>();
pulsar().getBrokerService().fetchPartitionedTopicMetadataAsync(topicName).thenAccept(partitionMetadata -> {
if (partitionMetadata.partitions < 1) {
result.completeExceptionally(new RestException(Status.CONFLICT, "Topic is not partitioned topic"));
return;
}

if (partitionMetadata.partitions >= numPartitions) {
result.completeExceptionally(new RestException(Status.CONFLICT,
"number of partitions must be more than existing " + partitionMetadata.partitions));
return;
}

PulsarAdmin admin;
try {
admin = pulsar().getAdminClient();
} catch (PulsarServerException e1) {
result.completeExceptionally(e1);
return;
}
if (expectPartitions < 1) {
return FutureUtil.failedFuture(new RestException(Status.CONFLICT, "Topic is not partitioned topic"));
}
PulsarAdmin admin;
try {
admin = pulsar().getAdminClient();
} catch (PulsarServerException e1) {
return FutureUtil.failedFuture(e1);
}

admin.topics().getStatsAsync(topicName.getPartition(0).toString()).thenAccept(stats -> {
List<CompletableFuture<Void>> subscriptionFutures = new ArrayList<>();
admin.topics().getStatsAsync(topicName.getPartition(0).toString()).thenAccept(stats -> {
List<CompletableFuture<Void>> subscriptionFutures = new ArrayList<>();

stats.getSubscriptions().entrySet().forEach(e -> {
String subscription = e.getKey();
SubscriptionStats ss = e.getValue();
if (!ss.isDurable()) {
// We must not re-create non-durable subscriptions on the new partitions
return;
}
boolean replicated = ss.isReplicated();

for (int i = partitionMetadata.partitions; i < numPartitions; i++) {
final String topicNamePartition = topicName.getPartition(i).toString();
CompletableFuture<Void> future = new CompletableFuture<>();
admin.topics().createSubscriptionAsync(topicNamePartition,
subscription, MessageId.latest, replicated).whenComplete((__, ex) -> {
if (ex == null) {
stats.getSubscriptions().entrySet().forEach(e -> {
String subscription = e.getKey();
SubscriptionStats ss = e.getValue();
if (!ss.isDurable()) {
// We must not re-create non-durable subscriptions on the new partitions
return;
}
boolean replicated = ss.isReplicated();

for (int i = 0; i < expectPartitions; i++) {
final String topicNamePartition = topicName.getPartition(i).toString();
CompletableFuture<Void> future = new CompletableFuture<>();
admin.topics().createSubscriptionAsync(topicNamePartition,
subscription, MessageId.latest, replicated).whenComplete((__, ex) -> {
Comment thread
mattisonchao marked this conversation as resolved.
if (ex == null) {
future.complete(null);
} else {
Throwable realCause = FutureUtil.unwrapCompletionException(ex);
if (realCause instanceof PulsarAdminException.ConflictException) {
future.complete(null);
} else {
if (ignoreConflictException
&& ex instanceof PulsarAdminException.ConflictException) {
future.complete(null);
} else {
future.completeExceptionally(ex);
}
future.completeExceptionally(realCause);
}
});
subscriptionFutures.add(future);
}
});
}
});
subscriptionFutures.add(future);
}
});

FutureUtil.waitForAll(subscriptionFutures).thenRun(() -> {
log.info("[{}] Successfully created subscriptions on new partitions {}", clientAppId(), topicName);
result.complete(null);
}).exceptionally(ex -> {
log.warn("[{}] Failed to create subscriptions on new partitions for {}",
clientAppId(), topicName, ex);
result.completeExceptionally(ex);
return null;
});
FutureUtil.waitForAll(subscriptionFutures).thenRun(() -> {
log.info("[{}] Successfully created subscriptions on new partitions {}", clientAppId(), topicName);
result.complete(null);
}).exceptionally(ex -> {
if (ex.getCause() instanceof PulsarAdminException.NotFoundException) {
// The first partition doesn't exist, so there are currently to subscriptions to recreate
result.complete(null);
} else {
log.warn("[{}] Failed to get list of subscriptions of {}",
clientAppId(), topicName.getPartition(0), ex);
result.completeExceptionally(ex);
}
log.warn("[{}] Failed to create subscriptions on new partitions for {}",
clientAppId(), topicName, ex);
result.completeExceptionally(ex);
return null;
});
}).exceptionally(ex -> {
log.warn("[{}] Failed to get partition metadata for {}",
clientAppId(), topicName.toString());
result.completeExceptionally(ex);
if (ex.getCause() instanceof PulsarAdminException.NotFoundException) {
// The first partition doesn't exist, so there are currently to subscriptions to recreate
result.complete(null);
} else {
log.warn("[{}] Failed to get list of subscriptions of {}",
clientAppId(), topicName.getPartition(0), ex);
result.completeExceptionally(ex);
}
return null;
});
return result;
Expand Down
Loading