You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "nodece (via GitHub)" <gi...@apache.org> on 2025/04/28 08:45:45 UTC
[PR] [fix][broker] Allow recreation of partitioned topic after metadata loss [pulsar]
nodece opened a new pull request, #24225:
URL: https://github.com/apache/pulsar/pull/24225
### Motivation
If the partitioned topic metadata is lost, running `admin topics create-partitioned-topic` cannot successfully recreate the partitioned topic.
Example error:
```
> admin topics create-partitioned-topic my-topic -p 1
2025-04-25T19:45:14,373+0800 [AsyncHttpClient-7-2] WARN org.apache.pulsar.client.admin.internal.BaseResource - [http://localhost:8080/admin/v2/persistent/public/default/my-topic/partitions?createLocalTopicOnly=false] Failed to perform http put request: javax.ws.rs.ClientErrorException: HTTP 409 {"reason":"This topic already exists"}
This topic already exists
```
In this case, the actual topic partitions may still exist, but the partitioned metadata is missing, causing the recreation to fail with a `409 Conflict`.
### Modifications
- Use `pulsar().getNamespaceService().checkTopicExists(topicName)` to verify whether the topic/metadata exist. If the topic/metadata doesn't exist, we allow the recreation of the partitioned topic by proceeding with the create operation.
### Documentation
<!-- DO NOT REMOVE THIS SECTION. CHECK THE PROPER BOX ONLY. -->
- [ ] `doc` <!-- Your PR contains doc changes. -->
- [ ] `doc-required` <!-- Your PR changes impact docs and you will update later -->
- [x] `doc-not-needed` <!-- Your PR changes do not impact docs -->
- [ ] `doc-complete` <!-- Docs have been already added -->
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Allow recreation of partitioned topic after metadata loss [pulsar]
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece commented on code in PR #24225:
URL: https://github.com/apache/pulsar/pull/24225#discussion_r2084618803
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java:
##########
@@ -596,12 +598,32 @@ protected void internalCreatePartitionedTopic(AsyncResponse asyncResponse, int n
}
})
.thenCompose(__ -> checkTopicExistsAsync(topicName))
- .thenAccept(exists -> {
- if (exists) {
- log.warn("[{}] Failed to create already existing topic {}", clientAppId(), topicName);
- throw new RestException(Status.CONFLICT, "This topic already exists");
+ .thenAccept(topicExistsInfo -> {
+ try {
+ if (topicExistsInfo.isExists()) {
+ if (topicExistsInfo.getTopicType().equals(TopicType.NON_PARTITIONED)
+ || (topicExistsInfo.getTopicType().equals(TopicType.PARTITIONED)
+ && !createLocalTopicOnly)) {
Review Comment:
Ping @poorbarcode
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Allow recreation of partitioned topic after metadata loss [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #24225:
URL: https://github.com/apache/pulsar/pull/24225#discussion_r2085876990
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java:
##########
@@ -596,12 +598,32 @@ protected void internalCreatePartitionedTopic(AsyncResponse asyncResponse, int n
}
})
.thenCompose(__ -> checkTopicExistsAsync(topicName))
- .thenAccept(exists -> {
- if (exists) {
- log.warn("[{}] Failed to create already existing topic {}", clientAppId(), topicName);
- throw new RestException(Status.CONFLICT, "This topic already exists");
+ .thenAccept(topicExistsInfo -> {
+ try {
+ if (topicExistsInfo.isExists()) {
+ if (topicExistsInfo.getTopicType().equals(TopicType.NON_PARTITIONED)
+ || (topicExistsInfo.getTopicType().equals(TopicType.PARTITIONED)
+ && !createLocalTopicOnly)) {
Review Comment:
Your purpose does not seem to relate to the logic that continuously attempts to create a partitioned topic when partitioned metadata already exists and the request parameter `createLocalTopicOnly` is `false`
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Allow recreation of partitioned topic after metadata loss [pulsar]
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece commented on PR #24225:
URL: https://github.com/apache/pulsar/pull/24225#issuecomment-2835185003
`org.apache.pulsar.broker.namespace.NamespaceService#checkTopicExists` was enhanced in #24118 and #24154.
This PR depends on those improvements, as those branches may not include the necessary changes yet.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Allow recreation of partitioned topic after metadata loss [pulsar]
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece commented on code in PR #24225:
URL: https://github.com/apache/pulsar/pull/24225#discussion_r2068106804
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java:
##########
@@ -596,12 +598,32 @@ protected void internalCreatePartitionedTopic(AsyncResponse asyncResponse, int n
}
})
.thenCompose(__ -> checkTopicExistsAsync(topicName))
- .thenAccept(exists -> {
- if (exists) {
- log.warn("[{}] Failed to create already existing topic {}", clientAppId(), topicName);
- throw new RestException(Status.CONFLICT, "This topic already exists");
+ .thenAccept(topicExistsInfo -> {
+ try {
+ if (topicExistsInfo.isExists()) {
+ if (topicExistsInfo.getTopicType().equals(TopicType.NON_PARTITIONED)
+ || (topicExistsInfo.getTopicType().equals(TopicType.PARTITIONED)
+ && !createLocalTopicOnly)) {
Review Comment:
@poorbarcode The GEO replication is enabled at the namespace level. A partitioned topic is created on the local cluster, which also synchronizes the topic to the remote cluster and `createLocalTopicOnly=true`.
When the local and remote clusters share the global ZooKeeper, we allow the partitioned metadata to already exist.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Allow recreation of partitioned topic after metadata loss [pulsar]
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece commented on code in PR #24225:
URL: https://github.com/apache/pulsar/pull/24225#discussion_r2068093194
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest.java:
##########
@@ -3694,4 +3695,22 @@ public void testPermissionsAllowAclChangesOnNonExistentTopics() {
pulsar.getConfiguration().setAllowAclChangesOnNonExistentTopics(false);
}
}
+
+ @Test
+ public void testRecreatePartitionedTopicAfterMetadataLoss()
+ throws PulsarAdminException, ExecutionException, InterruptedException {
+ String namespace = "prop-xyz/ns1/";
+ final String random = UUID.randomUUID().toString();
+ final String topic = "persistent://" + namespace + random;
+ admin.topics().createPartitionedTopic(topic, 5);
+
+ // Delete the topic metadata.
+ pulsar.getPulsarResources().getNamespaceResources().getPartitionedTopicResources()
+ .deletePartitionedTopicAsync(TopicName.get(topic)).get();
+ List<String> partitionedTopicList = admin.topics().getPartitionedTopicList(namespace);
+ assertThat(partitionedTopicList).doesNotContain(topic);
+
+ // Create the partitioned topic again.
+ admin.topics().createPartitionedTopic(topic, 5);
Review Comment:
Reproduce the metadata loss:
1. Create the partitioned topic.
2. Delete the partitioned topic metadata.
3. Recreate the partitioned 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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Allow recreation of partitioned topic after metadata loss [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #24225:
URL: https://github.com/apache/pulsar/pull/24225#discussion_r2086219638
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java:
##########
@@ -596,12 +598,32 @@ protected void internalCreatePartitionedTopic(AsyncResponse asyncResponse, int n
}
})
.thenCompose(__ -> checkTopicExistsAsync(topicName))
- .thenAccept(exists -> {
- if (exists) {
- log.warn("[{}] Failed to create already existing topic {}", clientAppId(), topicName);
- throw new RestException(Status.CONFLICT, "This topic already exists");
+ .thenAccept(topicExistsInfo -> {
+ try {
+ if (topicExistsInfo.isExists()) {
+ if (topicExistsInfo.getTopicType().equals(TopicType.NON_PARTITIONED)
+ || (topicExistsInfo.getTopicType().equals(TopicType.PARTITIONED)
+ && !createLocalTopicOnly)) {
Review Comment:
> Once such a partition is mistakenly created as a non-partitioned topic, it is no longer possible to convert it into a partitioned topic, because Pulsar does not allow overwriting existing topics. This results in effective metadata loss. A similar issue can occur in geo-replication scenarios, where only the individual partition topics are replicated, but the partitioned topic metadata is not.
> Moreover, if the partitioned topic metadata is lost, running `pulsar-admin topics create-partitioned-topic` fails to recreate it:
> ```
> > pulsar-admin topics create-partitioned-topic test-partitioned-topic -p 4
> 2025-04-25T19:45:14,373+0800 [AsyncHttpClient-7-2] WARN org.apache.pulsar.client.admin.internal.BaseResource - [http://localhost:8080/admin/v2/persistent/public/default/test-partitioned-topic/partitions?createLocalTopicOnly=false] Failed to perform http put request: javax.ws.rs.ClientErrorException: HTTP 409 {"reason":"This topic already exists"}
This topic already exists
> ```
> In this case, the partition still exist, but the partitioned topic metadata is missing—causing a `409 Conflict` when attempting to recreate it.
The motivation of this PR is to create missed partitioned metadata if only partitions exist, but the purpose of the current modification is that it continuously creates the topic even though the partitioned metadata exists, right?
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Allow recreation of partitioned topic after metadata loss [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #24225:
URL: https://github.com/apache/pulsar/pull/24225#discussion_r2067970290
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java:
##########
@@ -596,12 +598,32 @@ protected void internalCreatePartitionedTopic(AsyncResponse asyncResponse, int n
}
})
.thenCompose(__ -> checkTopicExistsAsync(topicName))
- .thenAccept(exists -> {
- if (exists) {
- log.warn("[{}] Failed to create already existing topic {}", clientAppId(), topicName);
- throw new RestException(Status.CONFLICT, "This topic already exists");
+ .thenAccept(topicExistsInfo -> {
+ try {
+ if (topicExistsInfo.isExists()) {
+ if (topicExistsInfo.getTopicType().equals(TopicType.NON_PARTITIONED)
+ || (topicExistsInfo.getTopicType().equals(TopicType.PARTITIONED)
+ && !createLocalTopicOnly)) {
Review Comment:
Could you explain what `partitioned topic exists && !createLocalTopicOnly` mean?
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Allow recreation of partitioned topic after metadata loss [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #24225:
URL: https://github.com/apache/pulsar/pull/24225#discussion_r2087014471
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java:
##########
@@ -596,12 +598,32 @@ protected void internalCreatePartitionedTopic(AsyncResponse asyncResponse, int n
}
})
.thenCompose(__ -> checkTopicExistsAsync(topicName))
- .thenAccept(exists -> {
- if (exists) {
- log.warn("[{}] Failed to create already existing topic {}", clientAppId(), topicName);
- throw new RestException(Status.CONFLICT, "This topic already exists");
+ .thenAccept(topicExistsInfo -> {
+ try {
+ if (topicExistsInfo.isExists()) {
+ if (topicExistsInfo.getTopicType().equals(TopicType.NON_PARTITIONED)
+ || (topicExistsInfo.getTopicType().equals(TopicType.PARTITIONED)
+ && !createLocalTopicOnly)) {
Review Comment:
I reviewed the original purpose that added the param `createLocalTopicOnly`
- https://github.com/apache/pulsar/pull/10963/files#diff-6902810fd24ff29ec34052bcc6a0de36d76107649148d945b80ec2be888e9042R760-R768
It has two meanings
- The first one is the same as you explained here https://github.com/apache/pulsar/pull/24225/#discussion_r2086270383
- The second one: continuously creates partitions that pathed `/managed-ledgers` even if the partitioned topic metadata exists.
Regarding to the second meaning, you changed nothing
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Allow recreation of partitioned topic after metadata loss [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #24225:
URL: https://github.com/apache/pulsar/pull/24225#discussion_r2087114795
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java:
##########
@@ -635,6 +657,26 @@ && pulsar().getConfig().isCreateTopicToRemoteClusterForReplication()) {
});
}
+ private CompletableFuture<Integer> getMaxPartitionIndex(TopicName topicName) {
+ if (topicName.getDomain() == TopicDomain.persistent) {
+ AtomicInteger maxIndex = new AtomicInteger(-1);
+ return getPulsarResources().getTopicResources().listPersistentTopicsAsync(topicName.getNamespaceObject())
+ .thenApply(list -> {
+ for (String s : list) {
+ TopicName item = TopicName.get(s);
+ if (item.isPartitioned() && item.getPartitionedTopicName()
+ .startsWith(topicName.getPartitionedTopicName())) {
+ if (item.getPartitionIndex() > maxIndex.get()) {
+ maxIndex.set(item.getPartitionIndex());
+ }
+ }
+ }
+ return maxIndex.get();
Review Comment:
it looks like `AtomicInteger maxIndex` could be replaced with an ordinary local `int` variable in the `thenApply` lambda.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Allow recreation of partitioned topic after metadata loss [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #24225:
URL: https://github.com/apache/pulsar/pull/24225#discussion_r2085797819
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java:
##########
@@ -596,12 +598,32 @@ protected void internalCreatePartitionedTopic(AsyncResponse asyncResponse, int n
}
})
.thenCompose(__ -> checkTopicExistsAsync(topicName))
- .thenAccept(exists -> {
- if (exists) {
- log.warn("[{}] Failed to create already existing topic {}", clientAppId(), topicName);
- throw new RestException(Status.CONFLICT, "This topic already exists");
+ .thenAccept(topicExistsInfo -> {
+ try {
+ if (topicExistsInfo.isExists()) {
+ if (topicExistsInfo.getTopicType().equals(TopicType.NON_PARTITIONED)
+ || (topicExistsInfo.getTopicType().equals(TopicType.PARTITIONED)
+ && !createLocalTopicOnly)) {
Review Comment:
@nodece
> When a partitioned topic will be synchronized to the remote cluster, we allow the remote cluster includes this metadata. The local and remote clusters may be shared a global zk.
> createLocalTopicOnly=true means this creation request comes from a cluster, not user client, we can continue to create the partitioned topic if the metadata is existing on the remote cluster.
> The pulsar has been covered this changed by existing test.
Could you explain the flow of the issue that you solved? Just like follows
- There are two clusters `[c1, c2]` enabled Geo-Replication `c1 -> c2`.
- `c2` has partitioned metadata, `c1` has not
- I failed to take the next step.....
> The previous code only checked whether the topic was a partition, without verifying its partitioned metadata. The updated logic now checks the partition status and its associated metadata, which is why you are seeing this additional check.
The original code was based on such an assumption: if a partition exists, the partitioned metadata should exist. I really do not know what issue you want to solve đŸ˜‚
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Allow recreation of partitioned topic after metadata loss [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on PR #24225:
URL: https://github.com/apache/pulsar/pull/24225#issuecomment-2835203719
> `org.apache.pulsar.broker.namespace.NamespaceService#checkTopicExists` was enhanced in #24118 and #24154. This PR depends on those improvements, as those branches may not include the necessary changes yet.
In that case, it's better to remove the release labels until the condition has been addressed. The reason for this is that release labels are used to track cherry-picking of individual PRs. This PR cannot be cherry-picked and therefore it shouldn't contain the label until it's ready. I use [some scripts to track cherry-picking across branches](https://pulsar.apache.org/contribute/maintenance-process/#cherry-picking-cli-tooling) using this assumption. This PR would pop up in the cherry-picking queue each time if it's labeled.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Allow recreation of partitioned topic after metadata loss [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on PR #24225:
URL: https://github.com/apache/pulsar/pull/24225#issuecomment-2835025474
> This PR depends on the `checkTopicExists` introduced in #24118 and #24154, please be cautious when cherry-picking it to branch-3.0 and branch-4.0.
@nodece Could we remove the release labels in that case?
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Allow recreation of partitioned topic after metadata loss [pulsar]
Posted by "wangchao316 (via GitHub)" <gi...@apache.org>.
wangchao316 commented on PR #24225:
URL: https://github.com/apache/pulsar/pull/24225#issuecomment-2840648388
hello , I am a new developer for pulsar, our product start to use pulsar, which replace kafka in serverless env. I can learn your pr. I have a little question. why does the actual topic partitions may still exist, but the partitioned metadata is missing? Human error deletion or does not full delete?
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Allow recreation of partitioned topic after metadata loss [pulsar]
Posted by "wangchao316 (via GitHub)" <gi...@apache.org>.
wangchao316 commented on PR #24225:
URL: https://github.com/apache/pulsar/pull/24225#issuecomment-2840926659
> > why does the actual topic partitions may still exist, but the partitioned metadata is missing? Human error deletion or does not full delete?
>
> Good catch! This PR description has been updated, please check.
thanks, I understand.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Allow recreation of partitioned topic after metadata loss [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #24225:
URL: https://github.com/apache/pulsar/pull/24225#discussion_r2085876990
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java:
##########
@@ -596,12 +598,32 @@ protected void internalCreatePartitionedTopic(AsyncResponse asyncResponse, int n
}
})
.thenCompose(__ -> checkTopicExistsAsync(topicName))
- .thenAccept(exists -> {
- if (exists) {
- log.warn("[{}] Failed to create already existing topic {}", clientAppId(), topicName);
- throw new RestException(Status.CONFLICT, "This topic already exists");
+ .thenAccept(topicExistsInfo -> {
+ try {
+ if (topicExistsInfo.isExists()) {
+ if (topicExistsInfo.getTopicType().equals(TopicType.NON_PARTITIONED)
+ || (topicExistsInfo.getTopicType().equals(TopicType.PARTITIONED)
+ && !createLocalTopicOnly)) {
Review Comment:
Your purpose does not seem to relate to the logic that continuously attempts to create a partitioned topic when `topicExistsInfo.getTopicType().equals(TopicType.PARTITIONED) && !createLocalTopicOnly`
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Allow recreation of partitioned topic after metadata loss [pulsar]
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece commented on code in PR #24225:
URL: https://github.com/apache/pulsar/pull/24225#discussion_r2081300105
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java:
##########
@@ -596,12 +598,32 @@ protected void internalCreatePartitionedTopic(AsyncResponse asyncResponse, int n
}
})
.thenCompose(__ -> checkTopicExistsAsync(topicName))
- .thenAccept(exists -> {
- if (exists) {
- log.warn("[{}] Failed to create already existing topic {}", clientAppId(), topicName);
- throw new RestException(Status.CONFLICT, "This topic already exists");
+ .thenAccept(topicExistsInfo -> {
+ try {
+ if (topicExistsInfo.isExists()) {
+ if (topicExistsInfo.getTopicType().equals(TopicType.NON_PARTITIONED)
+ || (topicExistsInfo.getTopicType().equals(TopicType.PARTITIONED)
+ && !createLocalTopicOnly)) {
Review Comment:
Ping @poorbarcode
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Allow recreation of partitioned topic after metadata loss [pulsar]
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece commented on code in PR #24225:
URL: https://github.com/apache/pulsar/pull/24225#discussion_r2086270383
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java:
##########
@@ -596,12 +598,32 @@ protected void internalCreatePartitionedTopic(AsyncResponse asyncResponse, int n
}
})
.thenCompose(__ -> checkTopicExistsAsync(topicName))
- .thenAccept(exists -> {
- if (exists) {
- log.warn("[{}] Failed to create already existing topic {}", clientAppId(), topicName);
- throw new RestException(Status.CONFLICT, "This topic already exists");
+ .thenAccept(topicExistsInfo -> {
+ try {
+ if (topicExistsInfo.isExists()) {
+ if (topicExistsInfo.getTopicType().equals(TopicType.NON_PARTITIONED)
+ || (topicExistsInfo.getTopicType().equals(TopicType.PARTITIONED)
+ && !createLocalTopicOnly)) {
Review Comment:
@poorbarcode
> The motivation of this PR is to create missed partitioned metadata if only partitions exist.
This is correct.
> but the purpose of the current modification is that it continuously creates the topic even though the partitioned metadata exists, right?
- When `createLocalTopicOnly = false` (the default), and the partitioned metadata already exists on the local cluster, the broker rejects the creation request — this is expected and correct behavior.
- When `createLocalTopicOnly = true`, the creation logic runs on the remote cluster as part of GEO replication. In this case, if the metadata already exists, we continue to create the partition on the remote cluster.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Allow recreation of partitioned topic after metadata loss [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #24225:
URL: https://github.com/apache/pulsar/pull/24225#discussion_r2086457591
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java:
##########
@@ -596,12 +598,32 @@ protected void internalCreatePartitionedTopic(AsyncResponse asyncResponse, int n
}
})
.thenCompose(__ -> checkTopicExistsAsync(topicName))
- .thenAccept(exists -> {
- if (exists) {
- log.warn("[{}] Failed to create already existing topic {}", clientAppId(), topicName);
- throw new RestException(Status.CONFLICT, "This topic already exists");
+ .thenAccept(topicExistsInfo -> {
+ try {
+ if (topicExistsInfo.isExists()) {
+ if (topicExistsInfo.getTopicType().equals(TopicType.NON_PARTITIONED)
+ || (topicExistsInfo.getTopicType().equals(TopicType.PARTITIONED)
+ && !createLocalTopicOnly)) {
Review Comment:
> When createLocalTopicOnly = true, the creation logic runs on the remote cluster as part of GEO replication. In this case, if the metadata already exists, we continue to create the partition on the remote cluster.
Regarding this case, please call create missed partitions instead of create partitioned metadata
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Allow recreation of partitioned topic after metadata loss [pulsar]
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece commented on code in PR #24225:
URL: https://github.com/apache/pulsar/pull/24225#discussion_r2085908115
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java:
##########
@@ -596,12 +598,32 @@ protected void internalCreatePartitionedTopic(AsyncResponse asyncResponse, int n
}
})
.thenCompose(__ -> checkTopicExistsAsync(topicName))
- .thenAccept(exists -> {
- if (exists) {
- log.warn("[{}] Failed to create already existing topic {}", clientAppId(), topicName);
- throw new RestException(Status.CONFLICT, "This topic already exists");
+ .thenAccept(topicExistsInfo -> {
+ try {
+ if (topicExistsInfo.isExists()) {
+ if (topicExistsInfo.getTopicType().equals(TopicType.NON_PARTITIONED)
+ || (topicExistsInfo.getTopicType().equals(TopicType.PARTITIONED)
+ && !createLocalTopicOnly)) {
Review Comment:
I'd suggest using an AI tool to summarize all the comments and descriptions in this PR. It can help piece together the context and motivation more effectively. I'm confident it will make the purpose of this change much easier to grasp.
The existing test can cover this change, but I forgot this test name.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Allow recreation of partitioned topic after metadata loss [pulsar]
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece commented on PR #24225:
URL: https://github.com/apache/pulsar/pull/24225#issuecomment-2840696481
> why does the actual topic partitions may still exist, but the partitioned metadata is missing? Human error deletion or does not full delete?
Good catch! This PR description has been updated, please check.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Allow recreation of partitioned topic after metadata loss [pulsar]
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece commented on code in PR #24225:
URL: https://github.com/apache/pulsar/pull/24225#discussion_r2072293154
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java:
##########
@@ -596,12 +598,32 @@ protected void internalCreatePartitionedTopic(AsyncResponse asyncResponse, int n
}
})
.thenCompose(__ -> checkTopicExistsAsync(topicName))
- .thenAccept(exists -> {
- if (exists) {
- log.warn("[{}] Failed to create already existing topic {}", clientAppId(), topicName);
- throw new RestException(Status.CONFLICT, "This topic already exists");
+ .thenAccept(topicExistsInfo -> {
+ try {
+ if (topicExistsInfo.isExists()) {
+ if (topicExistsInfo.getTopicType().equals(TopicType.NON_PARTITIONED)
+ || (topicExistsInfo.getTopicType().equals(TopicType.PARTITIONED)
+ && !createLocalTopicOnly)) {
Review Comment:
@poorbarcode When a partitioned topic will be synchronized to the remote cluster, we allow the remote cluster includes this metadata. The local and remote clusters may be shared a global zk.
createLocalTopicOnly=true means this creation request comes from a cluster, not user client, we can continue to create the partitioned topic if the metadata is existing.
The pulsar has been covered this changed by existing 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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Allow recreation of partitioned topic after metadata loss [pulsar]
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece commented on code in PR #24225:
URL: https://github.com/apache/pulsar/pull/24225#discussion_r2068253835
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java:
##########
@@ -596,12 +598,32 @@ protected void internalCreatePartitionedTopic(AsyncResponse asyncResponse, int n
}
})
.thenCompose(__ -> checkTopicExistsAsync(topicName))
- .thenAccept(exists -> {
- if (exists) {
- log.warn("[{}] Failed to create already existing topic {}", clientAppId(), topicName);
- throw new RestException(Status.CONFLICT, "This topic already exists");
+ .thenAccept(topicExistsInfo -> {
+ try {
+ if (topicExistsInfo.isExists()) {
+ if (topicExistsInfo.getTopicType().equals(TopicType.NON_PARTITIONED)
+ || (topicExistsInfo.getTopicType().equals(TopicType.PARTITIONED)
+ && !createLocalTopicOnly)) {
Review Comment:
@poorbarcode
> The current login in the PR attempts to create a partitioned topic if `createLocalTopicOnly` is `true`, either using global ZK or not.
This is correct. The `createLocalTopicOnly=true` is used to prevent the cluster doesn't synchronizing this topic to the remote cluster.
> I think the correct action should be as follows:
>
> * If users want to create a partitioned topic on the remote cluster, they should call the creation command on the remote cluster. Do not call the creation command on the local cluster.
I remember that if the GEO replication is enabled at the namespace level, the topic will be synchronized to the remote/local cluster, so like:
- `createLocalTopicOnly=true`:
- Local/remote cluster synchronizes the topic to the remote/local cluster.
- `createLocalTopicOnly=false`
- Local/remote cluster only creates the topic locally, which doesn't synchronize this topic to the remote/local cluster.
> * If you want to expand partitions, you should call `pulsar-admin update-partitioned-topic`. Do not call the creation command.
Correct, I didn't change this log. When the metadata exists, `This topic already exists` will be thrown.
Could you check the following code? It would be helpful for you:
https://github.com/apache/pulsar/blob/v4.0.4/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java#L615-L623
https://github.com/apache/pulsar/blob/v4.0.4/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java#L760-L765
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Allow recreation of partitioned topic after metadata loss [pulsar]
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece commented on code in PR #24225:
URL: https://github.com/apache/pulsar/pull/24225#discussion_r2068253835
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java:
##########
@@ -596,12 +598,32 @@ protected void internalCreatePartitionedTopic(AsyncResponse asyncResponse, int n
}
})
.thenCompose(__ -> checkTopicExistsAsync(topicName))
- .thenAccept(exists -> {
- if (exists) {
- log.warn("[{}] Failed to create already existing topic {}", clientAppId(), topicName);
- throw new RestException(Status.CONFLICT, "This topic already exists");
+ .thenAccept(topicExistsInfo -> {
+ try {
+ if (topicExistsInfo.isExists()) {
+ if (topicExistsInfo.getTopicType().equals(TopicType.NON_PARTITIONED)
+ || (topicExistsInfo.getTopicType().equals(TopicType.PARTITIONED)
+ && !createLocalTopicOnly)) {
Review Comment:
@poorbarcode
> The current login in the PR attempts to create a partitioned topic if `createLocalTopicOnly` is `true`, either using global ZK or not.
This is correct. The `createLocalTopicOnly=true` is used to prevent the cluster doesn't synchronizing this topic to the remote cluster.
> I think the correct action should be as follows:
>
> * If users want to create a partitioned topic on the remote cluster, they should call the creation command on the remote cluster. Do not call the creation command on the local cluster.
I remember that if the GEO replication is enabled at the namespace level, the topic will be synchronized to the remote/local cluster, so like:
- `createLocalTopicOnly=true`:
- Local/remote cluster only creates the topic locally, which doesn't synchronize this topic to the remote/local cluster.
- `createLocalTopicOnly=false`(default behavior)
- Local/remote cluster synchronizes the topic to the remote/local cluster.
> * If you want to expand partitions, you should call `pulsar-admin update-partitioned-topic`. Do not call the creation command.
Correct, I didn't change this log. When the metadata exists, `This topic already exists` will be thrown.
Could you check the following code? It would be helpful for you:
https://github.com/apache/pulsar/blob/v4.0.4/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java#L615-L623
https://github.com/apache/pulsar/blob/v4.0.4/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java#L760-L765
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Allow recreation of partitioned topic after metadata loss [pulsar]
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece commented on PR #24225:
URL: https://github.com/apache/pulsar/pull/24225#issuecomment-2835136000
> > This PR depends on the `checkTopicExists` introduced in #24118 and #24154, please be cautious when cherry-picking it to branch-3.0 and branch-4.0.
>
> @nodece Could we remove the release labels in that case?
This is a bugfix. I suggest we cherry-pick this PR to the branch-3.0 and branch-4.0.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Allow recreation of partitioned topic after metadata loss [pulsar]
Posted by "crossoverJie (via GitHub)" <gi...@apache.org>.
crossoverJie commented on code in PR #24225:
URL: https://github.com/apache/pulsar/pull/24225#discussion_r2067950792
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest.java:
##########
@@ -3694,4 +3695,22 @@ public void testPermissionsAllowAclChangesOnNonExistentTopics() {
pulsar.getConfiguration().setAllowAclChangesOnNonExistentTopics(false);
}
}
+
+ @Test
+ public void testRecreatePartitionedTopicAfterMetadataLoss()
+ throws PulsarAdminException, ExecutionException, InterruptedException {
+ String namespace = "prop-xyz/ns1/";
+ final String random = UUID.randomUUID().toString();
+ final String topic = "persistent://" + namespace + random;
+ admin.topics().createPartitionedTopic(topic, 5);
+
+ // Delete the topic metadata.
+ pulsar.getPulsarResources().getNamespaceResources().getPartitionedTopicResources()
+ .deletePartitionedTopicAsync(TopicName.get(topic)).get();
+ List<String> partitionedTopicList = admin.topics().getPartitionedTopicList(namespace);
+ assertThat(partitionedTopicList).doesNotContain(topic);
+
+ // Create the partitioned topic again.
+ admin.topics().createPartitionedTopic(topic, 5);
Review Comment:
Why check the metadata completeness again?
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Allow recreation of partitioned topic after metadata loss [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #24225:
URL: https://github.com/apache/pulsar/pull/24225#discussion_r2067967483
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest.java:
##########
@@ -3694,4 +3695,22 @@ public void testPermissionsAllowAclChangesOnNonExistentTopics() {
pulsar.getConfiguration().setAllowAclChangesOnNonExistentTopics(false);
}
}
+
+ @Test
+ public void testRecreatePartitionedTopicAfterMetadataLoss()
+ throws PulsarAdminException, ExecutionException, InterruptedException {
+ String namespace = "prop-xyz/ns1/";
+ final String random = UUID.randomUUID().toString();
+ final String topic = "persistent://" + namespace + random;
+ admin.topics().createPartitionedTopic(topic, 5);
+
+ // Delete the topic metadata.
+ pulsar.getPulsarResources().getNamespaceResources().getPartitionedTopicResources()
+ .deletePartitionedTopicAsync(TopicName.get(topic)).get();
+ List<String> partitionedTopicList = admin.topics().getPartitionedTopicList(namespace);
+ assertThat(partitionedTopicList).doesNotContain(topic);
+
+ // Create the partitioned topic again.
+ admin.topics().createPartitionedTopic(topic, 5);
Review Comment:
He removed the partitioned metadata at line `3708` to mock an issue that someone did the wrong operation on the metadata store.
After that, he can reproduce the scenario he described in the Motivation section.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Allow recreation of partitioned topic after metadata loss [pulsar]
Posted by "liangyepianzhou (via GitHub)" <gi...@apache.org>.
liangyepianzhou commented on code in PR #24225:
URL: https://github.com/apache/pulsar/pull/24225#discussion_r2085784116
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java:
##########
@@ -1412,6 +1412,10 @@ public CompletableFuture<List<String>> getOwnedTopicListForNamespaceBundle(Names
* Check topic exists( partitioned or non-partitioned ).
*/
public CompletableFuture<TopicExistsInfo> checkTopicExists(TopicName topic) {
Review Comment:
OK, Let me submit a PR 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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Allow recreation of partitioned topic after metadata loss [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #24225:
URL: https://github.com/apache/pulsar/pull/24225#discussion_r2072252144
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java:
##########
@@ -596,12 +598,32 @@ protected void internalCreatePartitionedTopic(AsyncResponse asyncResponse, int n
}
})
.thenCompose(__ -> checkTopicExistsAsync(topicName))
- .thenAccept(exists -> {
- if (exists) {
- log.warn("[{}] Failed to create already existing topic {}", clientAppId(), topicName);
- throw new RestException(Status.CONFLICT, "This topic already exists");
+ .thenAccept(topicExistsInfo -> {
+ try {
+ if (topicExistsInfo.isExists()) {
+ if (topicExistsInfo.getTopicType().equals(TopicType.NON_PARTITIONED)
+ || (topicExistsInfo.getTopicType().equals(TopicType.PARTITIONED)
+ && !createLocalTopicOnly)) {
Review Comment:
I think I did not explain my question clearly. What I want to ask is as follows
> Could you explain what partitioned topic exists && !createLocalTopicOnly mean?
The original behavior is that users will get a Conflict error if the topic already exists when calling a topic creation, see https://github.com/apache/pulsar/blob/v4.0.4/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java#L598-L604
But you changed the behavior in the current PR, the check will pass when "partitioned topic exists && !createLocalTopicOnly", I want to know why
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Allow recreation of partitioned topic after metadata loss [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #24225:
URL: https://github.com/apache/pulsar/pull/24225#discussion_r2068135805
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java:
##########
@@ -596,12 +598,32 @@ protected void internalCreatePartitionedTopic(AsyncResponse asyncResponse, int n
}
})
.thenCompose(__ -> checkTopicExistsAsync(topicName))
- .thenAccept(exists -> {
- if (exists) {
- log.warn("[{}] Failed to create already existing topic {}", clientAppId(), topicName);
- throw new RestException(Status.CONFLICT, "This topic already exists");
+ .thenAccept(topicExistsInfo -> {
+ try {
+ if (topicExistsInfo.isExists()) {
+ if (topicExistsInfo.getTopicType().equals(TopicType.NON_PARTITIONED)
+ || (topicExistsInfo.getTopicType().equals(TopicType.PARTITIONED)
+ && !createLocalTopicOnly)) {
Review Comment:
> When the local and remote clusters share the global ZooKeeper, we already allow the partitioned metadata to exist.
The current login in the PR attempts to create a partitioned topic if `createLocalTopicOnly` is `true`, either using global ZK or not.
I think the correct action should be as follows:
- If users want to create a partitioned topic on the remote cluster, they should call the creation command on the remote cluster. Do not call the creation command on the local cluster.
- If you want to expand partitions, you should call `pulsar-admin update-partitioned-topic`. Do not call the creation command.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Allow recreation of partitioned topic after metadata loss [pulsar]
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece commented on code in PR #24225:
URL: https://github.com/apache/pulsar/pull/24225#discussion_r2072311103
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java:
##########
@@ -596,12 +598,32 @@ protected void internalCreatePartitionedTopic(AsyncResponse asyncResponse, int n
}
})
.thenCompose(__ -> checkTopicExistsAsync(topicName))
- .thenAccept(exists -> {
- if (exists) {
- log.warn("[{}] Failed to create already existing topic {}", clientAppId(), topicName);
- throw new RestException(Status.CONFLICT, "This topic already exists");
+ .thenAccept(topicExistsInfo -> {
+ try {
+ if (topicExistsInfo.isExists()) {
+ if (topicExistsInfo.getTopicType().equals(TopicType.NON_PARTITIONED)
+ || (topicExistsInfo.getTopicType().equals(TopicType.PARTITIONED)
+ && !createLocalTopicOnly)) {
Review Comment:
The previous code only checked whether the topic was a partition, without verifying its partitioned metadata. The updated logic now checks the partition status and its associated metadata, which is why you are seeing this additional check.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Allow recreation of partitioned topic after metadata loss [pulsar]
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece commented on PR #24225:
URL: https://github.com/apache/pulsar/pull/24225#issuecomment-2834842203
This PR depends on the `checkTopicExists` introduced in #24118 and #24154, please be cautious when cherry-picking it to branch-3.0 and branch-4.0.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Allow recreation of partitioned topic after metadata loss [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #24225:
URL: https://github.com/apache/pulsar/pull/24225#discussion_r2067970290
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java:
##########
@@ -596,12 +598,32 @@ protected void internalCreatePartitionedTopic(AsyncResponse asyncResponse, int n
}
})
.thenCompose(__ -> checkTopicExistsAsync(topicName))
- .thenAccept(exists -> {
- if (exists) {
- log.warn("[{}] Failed to create already existing topic {}", clientAppId(), topicName);
- throw new RestException(Status.CONFLICT, "This topic already exists");
+ .thenAccept(topicExistsInfo -> {
+ try {
+ if (topicExistsInfo.isExists()) {
+ if (topicExistsInfo.getTopicType().equals(TopicType.NON_PARTITIONED)
+ || (topicExistsInfo.getTopicType().equals(TopicType.PARTITIONED)
+ && !createLocalTopicOnly)) {
Review Comment:
Could you explain what `partitioned topic exists && !createLocalTopicOnly` mean?
I think the correct behavior should be that : once the partitioned topic exists, the creation should fail
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Allow recreation of partitioned topic after metadata loss [pulsar]
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece commented on code in PR #24225:
URL: https://github.com/apache/pulsar/pull/24225#discussion_r2072293154
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java:
##########
@@ -596,12 +598,32 @@ protected void internalCreatePartitionedTopic(AsyncResponse asyncResponse, int n
}
})
.thenCompose(__ -> checkTopicExistsAsync(topicName))
- .thenAccept(exists -> {
- if (exists) {
- log.warn("[{}] Failed to create already existing topic {}", clientAppId(), topicName);
- throw new RestException(Status.CONFLICT, "This topic already exists");
+ .thenAccept(topicExistsInfo -> {
+ try {
+ if (topicExistsInfo.isExists()) {
+ if (topicExistsInfo.getTopicType().equals(TopicType.NON_PARTITIONED)
+ || (topicExistsInfo.getTopicType().equals(TopicType.PARTITIONED)
+ && !createLocalTopicOnly)) {
Review Comment:
@poorbarcode When a partitioned topic will be synchronized to the remote cluster, we allow the remote cluster includes this metadata. The local and remote clusters may be shared a global zk.
The pulsar has been covered this changed by existing 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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Allow recreation of partitioned topic after metadata loss [pulsar]
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece commented on code in PR #24225:
URL: https://github.com/apache/pulsar/pull/24225#discussion_r2084618240
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java:
##########
@@ -1412,6 +1412,10 @@ public CompletableFuture<List<String>> getOwnedTopicListForNamespaceBundle(Names
* Check topic exists( partitioned or non-partitioned ).
*/
public CompletableFuture<TopicExistsInfo> checkTopicExists(TopicName topic) {
Review Comment:
I did not change this method name.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Allow recreation of partitioned topic after metadata loss [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #24225:
URL: https://github.com/apache/pulsar/pull/24225#discussion_r2085876990
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java:
##########
@@ -596,12 +598,32 @@ protected void internalCreatePartitionedTopic(AsyncResponse asyncResponse, int n
}
})
.thenCompose(__ -> checkTopicExistsAsync(topicName))
- .thenAccept(exists -> {
- if (exists) {
- log.warn("[{}] Failed to create already existing topic {}", clientAppId(), topicName);
- throw new RestException(Status.CONFLICT, "This topic already exists");
+ .thenAccept(topicExistsInfo -> {
+ try {
+ if (topicExistsInfo.isExists()) {
+ if (topicExistsInfo.getTopicType().equals(TopicType.NON_PARTITIONED)
+ || (topicExistsInfo.getTopicType().equals(TopicType.PARTITIONED)
+ && !createLocalTopicOnly)) {
Review Comment:
Your purpose does not seem to relate to the logic that continuously attempts to create a partitioned topic when partitioned metadata already exists and the request parameter `createLocalTopicOnly` is `false`
@lhotari @crossoverJie @wangchao316 I have discussed with @nodece for a long time, but I still do not know the relationship between the PR's purpose and the change I mentioned here. Since you have approved the PR, could you explain the details?
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Allow recreation of partitioned topic after metadata loss [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #24225:
URL: https://github.com/apache/pulsar/pull/24225#discussion_r2086219638
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java:
##########
@@ -596,12 +598,32 @@ protected void internalCreatePartitionedTopic(AsyncResponse asyncResponse, int n
}
})
.thenCompose(__ -> checkTopicExistsAsync(topicName))
- .thenAccept(exists -> {
- if (exists) {
- log.warn("[{}] Failed to create already existing topic {}", clientAppId(), topicName);
- throw new RestException(Status.CONFLICT, "This topic already exists");
+ .thenAccept(topicExistsInfo -> {
+ try {
+ if (topicExistsInfo.isExists()) {
+ if (topicExistsInfo.getTopicType().equals(TopicType.NON_PARTITIONED)
+ || (topicExistsInfo.getTopicType().equals(TopicType.PARTITIONED)
+ && !createLocalTopicOnly)) {
Review Comment:
> Once such a partition is mistakenly created as a non-partitioned topic, it is no longer possible to convert it into a partitioned topic, because Pulsar does not allow overwriting existing topics. This results in effective metadata loss. A similar issue can occur in geo-replication scenarios, where only the individual partition topics are replicated, but the partitioned topic metadata is not.
> Moreover, if the partitioned topic metadata is lost, running `pulsar-admin topics create-partitioned-topic` fails to recreate it:
> ```
> > pulsar-admin topics create-partitioned-topic test-partitioned-topic -p 4
> 2025-04-25T19:45:14,373+0800 [AsyncHttpClient-7-2] WARN org.apache.pulsar.client.admin.internal.BaseResource - [http://localhost:8080/admin/v2/persistent/public/default/test-partitioned-topic/partitions?createLocalTopicOnly=false] Failed to perform http put request: javax.ws.rs.ClientErrorException: HTTP 409 {"reason":"This topic already exists"}
> This topic already exists
> ```
> In this case, the partition still exist, but the partitioned topic metadata is missing—causing a `409 Conflict` when attempting to recreate it.
The motivation of this PR is to create missed partitioned metadata if only partitions exist, but the purpose of the current modification is that it continuously creates the topic even though the partitioned metadata exists, right?
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Allow recreation of partitioned topic after metadata loss [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #24225:
URL: https://github.com/apache/pulsar/pull/24225#discussion_r2072252144
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java:
##########
@@ -596,12 +598,32 @@ protected void internalCreatePartitionedTopic(AsyncResponse asyncResponse, int n
}
})
.thenCompose(__ -> checkTopicExistsAsync(topicName))
- .thenAccept(exists -> {
- if (exists) {
- log.warn("[{}] Failed to create already existing topic {}", clientAppId(), topicName);
- throw new RestException(Status.CONFLICT, "This topic already exists");
+ .thenAccept(topicExistsInfo -> {
+ try {
+ if (topicExistsInfo.isExists()) {
+ if (topicExistsInfo.getTopicType().equals(TopicType.NON_PARTITIONED)
+ || (topicExistsInfo.getTopicType().equals(TopicType.PARTITIONED)
+ && !createLocalTopicOnly)) {
Review Comment:
I think I did not explain my question clearly. What I want to ask is as follows
> Could you explain what partitioned topic exists && !createLocalTopicOnly mean?
The original behavior is that users will get a Rest Conflict Error if the topic already exists when calling a topic creation, see https://github.com/apache/pulsar/blob/v4.0.4/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java#L598-L604
But you changed the behavior in the current PR, the check will pass when "partitioned topic exists && !createLocalTopicOnly", I want to know why
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java:
##########
@@ -596,12 +598,32 @@ protected void internalCreatePartitionedTopic(AsyncResponse asyncResponse, int n
}
})
.thenCompose(__ -> checkTopicExistsAsync(topicName))
- .thenAccept(exists -> {
- if (exists) {
- log.warn("[{}] Failed to create already existing topic {}", clientAppId(), topicName);
- throw new RestException(Status.CONFLICT, "This topic already exists");
+ .thenAccept(topicExistsInfo -> {
+ try {
+ if (topicExistsInfo.isExists()) {
+ if (topicExistsInfo.getTopicType().equals(TopicType.NON_PARTITIONED)
+ || (topicExistsInfo.getTopicType().equals(TopicType.PARTITIONED)
+ && !createLocalTopicOnly)) {
Review Comment:
I think I did not explain my question clearly. What I want to ask is as follows
> Could you explain what partitioned topic exists && !createLocalTopicOnly mean?
The original behavior is that users will get a HTTP Conflict Error if the topic already exists when calling a topic creation, see https://github.com/apache/pulsar/blob/v4.0.4/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java#L598-L604
But you changed the behavior in the current PR, the check will pass when "partitioned topic exists && !createLocalTopicOnly", I want to know why
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Allow recreation of partitioned topic after metadata loss [pulsar]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #24225:
URL: https://github.com/apache/pulsar/pull/24225#issuecomment-2838263593
## [Codecov](https://app.codecov.io/gh/apache/pulsar/pull/24225?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
Attention: Patch coverage is `93.93939%` with `2 lines` in your changes missing coverage. Please review.
> Project coverage is 74.26%. Comparing base [(`bbc6224`)](https://app.codecov.io/gh/apache/pulsar/commit/bbc62245c5ddba1de4b1e7cee4ab49334bc36277?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) to head [(`98a08b4`)](https://app.codecov.io/gh/apache/pulsar/commit/98a08b43d7e58162612c7ae01e8b23773a08dc08?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
> Report is 1052 commits behind head on master.
| [Files with missing lines](https://app.codecov.io/gh/apache/pulsar/pull/24225?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
|---|---|---|
| [.../org/apache/pulsar/broker/admin/AdminResource.java](https://app.codecov.io/gh/apache/pulsar/pull/24225?src=pr&el=tree&filepath=pulsar-broker%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fbroker%2Fadmin%2FAdminResource.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWJyb2tlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2Jyb2tlci9hZG1pbi9BZG1pblJlc291cmNlLmphdmE=) | 92.85% | [0 Missing and 2 partials :warning: ](https://app.codecov.io/gh/apache/pulsar/pull/24225?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
<details><summary>Additional details and impacted files</summary>
[](https://app.codecov.io/gh/apache/pulsar/pull/24225?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
```diff
@@ Coverage Diff @@
## master #24225 +/- ##
============================================
+ Coverage 73.57% 74.26% +0.69%
+ Complexity 32624 32129 -495
============================================
Files 1877 1866 -11
Lines 139502 144802 +5300
Branches 15299 16540 +1241
============================================
+ Hits 102638 107542 +4904
+ Misses 28908 28771 -137
- Partials 7956 8489 +533
```
| [Flag](https://app.codecov.io/gh/apache/pulsar/pull/24225/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [inttests](https://app.codecov.io/gh/apache/pulsar/pull/24225/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `26.73% <57.57%> (+2.15%)` | :arrow_up: |
| [systests](https://app.codecov.io/gh/apache/pulsar/pull/24225/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `23.27% <30.30%> (-1.06%)` | :arrow_down: |
| [unittests](https://app.codecov.io/gh/apache/pulsar/pull/24225/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `73.74% <93.93%> (+0.90%)` | :arrow_up: |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#carryforward-flags-in-the-pull-request-comment) to find out more.
| [Files with missing lines](https://app.codecov.io/gh/apache/pulsar/pull/24225?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [...pulsar/broker/admin/impl/PersistentTopicsBase.java](https://app.codecov.io/gh/apache/pulsar/pull/24225?src=pr&el=tree&filepath=pulsar-broker%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fbroker%2Fadmin%2Fimpl%2FPersistentTopicsBase.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWJyb2tlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2Jyb2tlci9hZG1pbi9pbXBsL1BlcnNpc3RlbnRUb3BpY3NCYXNlLmphdmE=) | `70.09% <100.00%> (+4.64%)` | :arrow_up: |
| [...ache/pulsar/broker/namespace/NamespaceService.java](https://app.codecov.io/gh/apache/pulsar/pull/24225?src=pr&el=tree&filepath=pulsar-broker%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fbroker%2Fnamespace%2FNamespaceService.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWJyb2tlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2Jyb2tlci9uYW1lc3BhY2UvTmFtZXNwYWNlU2VydmljZS5qYXZh) | `73.82% <100.00%> (+1.58%)` | :arrow_up: |
| [.../org/apache/pulsar/broker/admin/AdminResource.java](https://app.codecov.io/gh/apache/pulsar/pull/24225?src=pr&el=tree&filepath=pulsar-broker%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fbroker%2Fadmin%2FAdminResource.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWJyb2tlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2Jyb2tlci9hZG1pbi9BZG1pblJlc291cmNlLmphdmE=) | `76.64% <92.85%> (-0.99%)` | :arrow_down: |
... and [1081 files with indirect coverage changes](https://app.codecov.io/gh/apache/pulsar/pull/24225/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
</details>
<details><summary> :rocket: New features to boost your workflow: </summary>
- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Allow recreation of partitioned topic after metadata loss [pulsar]
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece commented on code in PR #24225:
URL: https://github.com/apache/pulsar/pull/24225#discussion_r2087166758
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java:
##########
@@ -635,6 +657,26 @@ && pulsar().getConfig().isCreateTopicToRemoteClusterForReplication()) {
});
}
+ private CompletableFuture<Integer> getMaxPartitionIndex(TopicName topicName) {
+ if (topicName.getDomain() == TopicDomain.persistent) {
+ AtomicInteger maxIndex = new AtomicInteger(-1);
+ return getPulsarResources().getTopicResources().listPersistentTopicsAsync(topicName.getNamespaceObject())
+ .thenApply(list -> {
+ for (String s : list) {
+ TopicName item = TopicName.get(s);
+ if (item.isPartitioned() && item.getPartitionedTopicName()
+ .startsWith(topicName.getPartitionedTopicName())) {
+ if (item.getPartitionIndex() > maxIndex.get()) {
+ maxIndex.set(item.getPartitionIndex());
+ }
+ }
+ }
+ return maxIndex.get();
Review Comment:
Done.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Allow recreation of partitioned topic after metadata loss [pulsar]
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece commented on code in PR #24225:
URL: https://github.com/apache/pulsar/pull/24225#discussion_r2086270383
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java:
##########
@@ -596,12 +598,32 @@ protected void internalCreatePartitionedTopic(AsyncResponse asyncResponse, int n
}
})
.thenCompose(__ -> checkTopicExistsAsync(topicName))
- .thenAccept(exists -> {
- if (exists) {
- log.warn("[{}] Failed to create already existing topic {}", clientAppId(), topicName);
- throw new RestException(Status.CONFLICT, "This topic already exists");
+ .thenAccept(topicExistsInfo -> {
+ try {
+ if (topicExistsInfo.isExists()) {
+ if (topicExistsInfo.getTopicType().equals(TopicType.NON_PARTITIONED)
+ || (topicExistsInfo.getTopicType().equals(TopicType.PARTITIONED)
+ && !createLocalTopicOnly)) {
Review Comment:
@poorbarcode
> The motivation of this PR is to create missed partitioned metadata if only partitions exist.
This is correct.
> but the purpose of the current modification is that it continuously creates the topic even though the partitioned metadata exists, right?
- When `createLocalTopicOnly = false` (the default), and the partitioned metadata already exists, the broker rejects the creation request — this is expected and correct behavior.
- When `createLocalTopicOnly = true`, the creation logic runs on the remote cluster as part of GEO replication. In this case, if the metadata already exists, we continue to create the partition on the remote cluster.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Allow recreation of partitioned topic after metadata loss [pulsar]
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece commented on code in PR #24225:
URL: https://github.com/apache/pulsar/pull/24225#discussion_r2086270383
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java:
##########
@@ -596,12 +598,32 @@ protected void internalCreatePartitionedTopic(AsyncResponse asyncResponse, int n
}
})
.thenCompose(__ -> checkTopicExistsAsync(topicName))
- .thenAccept(exists -> {
- if (exists) {
- log.warn("[{}] Failed to create already existing topic {}", clientAppId(), topicName);
- throw new RestException(Status.CONFLICT, "This topic already exists");
+ .thenAccept(topicExistsInfo -> {
+ try {
+ if (topicExistsInfo.isExists()) {
+ if (topicExistsInfo.getTopicType().equals(TopicType.NON_PARTITIONED)
+ || (topicExistsInfo.getTopicType().equals(TopicType.PARTITIONED)
+ && !createLocalTopicOnly)) {
Review Comment:
@poorbarcode
> The motivation of this PR is to create missed partitioned metadata if only partitions exist.
This is correct.
> but the purpose of the current modification is that it continuously creates the topic even though the partitioned metadata exists, right?
- When `createLocalTopicOnly = false` (the default), and the partitioned metadata already exists, the broker rejects the creation request — this is expected and correct behavior.
- When `createLocalTopicOnly = true`, the creation logic runs on the remote cluster as part of GEO replication. In this case, if the metadata already exists, we continue to create the partition on the remote cluster.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Allow recreation of partitioned topic after metadata loss [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #24225:
URL: https://github.com/apache/pulsar/pull/24225#discussion_r2087237846
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java:
##########
@@ -725,26 +766,16 @@ protected Map<String, CompletableFuture<Void>> internalCreatePartitionedTopicToR
/**
* Check the exists topics contains the given topic.
- * Since there are topic partitions and non-partitioned topics in Pulsar, must ensure both partitions
- * and non-partitioned topics are not duplicated. So, if compare with a partition name, we should compare
- * to the partitioned name of this partition.
+ * Since there are topic partitions and non-partitioned topics in Pulsar. This check ensures that
+ * there is no duplication between them. Partition *instances* (e.g., topic-partition-0)
+ * are not considered individually — only the base partitioned topic name is checked.
+ *
+ * <p>Note: Be sure to recycle the {@link TopicExistsInfo} after it is no longer needed.</p>
*
* @param topicName given topic name
*/
- protected CompletableFuture<Boolean> checkTopicExistsAsync(TopicName topicName) {
- return pulsar().getNamespaceService().getListOfTopics(topicName.getNamespaceObject(),
- CommandGetTopicsOfNamespace.Mode.ALL)
- .thenCompose(topics -> {
- boolean exists = false;
- for (String topic : topics) {
- if (topicName.getPartitionedTopicName().equals(
- TopicName.get(topic).getPartitionedTopicName())) {
- exists = true;
- break;
- }
- }
- return CompletableFuture.completedFuture(exists);
- });
+ protected CompletableFuture<TopicExistsInfo> checkTopicExistsAsync(TopicName topicName) {
+ return pulsar().getNamespaceService().checkTopicExists(topicName);
Review Comment:
this `checkTopicExists` is deprecated in NamespaceService. There's a method called `checkTopicExistsAsync` which I guess is intended to be used 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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Allow recreation of partitioned topic after metadata loss [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #24225:
URL: https://github.com/apache/pulsar/pull/24225#discussion_r2086219638
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java:
##########
@@ -596,12 +598,32 @@ protected void internalCreatePartitionedTopic(AsyncResponse asyncResponse, int n
}
})
.thenCompose(__ -> checkTopicExistsAsync(topicName))
- .thenAccept(exists -> {
- if (exists) {
- log.warn("[{}] Failed to create already existing topic {}", clientAppId(), topicName);
- throw new RestException(Status.CONFLICT, "This topic already exists");
+ .thenAccept(topicExistsInfo -> {
+ try {
+ if (topicExistsInfo.isExists()) {
+ if (topicExistsInfo.getTopicType().equals(TopicType.NON_PARTITIONED)
+ || (topicExistsInfo.getTopicType().equals(TopicType.PARTITIONED)
+ && !createLocalTopicOnly)) {
Review Comment:
The Motivation said:
> Once such a partition is mistakenly created as a non-partitioned topic, it is no longer possible to convert it into a partitioned topic, because Pulsar does not allow overwriting existing topics. This results in effective metadata loss. A similar issue can occur in geo-replication scenarios, where only the individual partition topics are replicated, but the partitioned topic metadata is not.
> Moreover, if the partitioned topic metadata is lost, running `pulsar-admin topics create-partitioned-topic` fails to recreate it:
> ```
> > pulsar-admin topics create-partitioned-topic test-partitioned-topic -p 4
> 2025-04-25T19:45:14,373+0800 [AsyncHttpClient-7-2] WARN org.apache.pulsar.client.admin.internal.BaseResource - [http://localhost:8080/admin/v2/persistent/public/default/test-partitioned-topic/partitions?createLocalTopicOnly=false] Failed to perform http put request: javax.ws.rs.ClientErrorException: HTTP 409 {"reason":"This topic already exists"}
> This topic already exists
> ```
> In this case, the partition still exist, but the partitioned topic metadata is missing—causing a `409 Conflict` when attempting to recreate it.
The motivation of this PR is to create missed partitioned metadata if only partitions exist, but the purpose of the current modification is that it continuously creates the topic even though the partitioned metadata exists, right?
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Allow recreation of partitioned topic after metadata loss [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #24225:
URL: https://github.com/apache/pulsar/pull/24225#discussion_r2087201615
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java:
##########
@@ -635,6 +656,26 @@ && pulsar().getConfig().isCreateTopicToRemoteClusterForReplication()) {
});
}
+ private CompletableFuture<Integer> getMaxPartitionIndex(TopicName topicName) {
+ if (topicName.getDomain() == TopicDomain.persistent) {
+ return getPulsarResources().getTopicResources().listPersistentTopicsAsync(topicName.getNamespaceObject())
+ .thenApply(list -> {
+ int maxIndex = -1;
+ for (String s : list) {
+ TopicName item = TopicName.get(s);
+ if (item.isPartitioned() && item.getPartitionedTopicName()
+ .startsWith(topicName.getPartitionedTopicName())) {
Review Comment:
why is this `.startsWith`? Shouldn't it be `.equals`?
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Allow recreation of partitioned topic after metadata loss [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on PR #24225:
URL: https://github.com/apache/pulsar/pull/24225#issuecomment-2835028432
@nodece Please check the test failures whether those are valid ones
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Allow recreation of partitioned topic after metadata loss [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on PR #24225:
URL: https://github.com/apache/pulsar/pull/24225#issuecomment-2835155819
> > > This PR depends on the `checkTopicExists` introduced in #24118 and #24154, please be cautious when cherry-picking it to branch-3.0 and branch-4.0.
> >
> >
> > @nodece Could we remove the release labels in that case?
>
> This is a bugfix. I suggest we cherry-pick this PR to the branch-3.0 and branch-4.0.
@nodece what does "This PR depends on the `checkTopicExists` introduced in #24118 and #24154, please be cautious when cherry-picking it to branch-3.0 and branch-4.0." mean?
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Allow recreation of partitioned topic after metadata loss [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #24225:
URL: https://github.com/apache/pulsar/pull/24225#discussion_r2068139831
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java:
##########
@@ -596,12 +598,32 @@ protected void internalCreatePartitionedTopic(AsyncResponse asyncResponse, int n
}
})
.thenCompose(__ -> checkTopicExistsAsync(topicName))
- .thenAccept(exists -> {
- if (exists) {
- log.warn("[{}] Failed to create already existing topic {}", clientAppId(), topicName);
- throw new RestException(Status.CONFLICT, "This topic already exists");
+ .thenAccept(topicExistsInfo -> {
+ try {
+ if (topicExistsInfo.isExists()) {
+ if (topicExistsInfo.getTopicType().equals(TopicType.NON_PARTITIONED)
+ || (topicExistsInfo.getTopicType().equals(TopicType.PARTITIONED)
+ && !createLocalTopicOnly)) {
Review Comment:
Could you explain the purpose of this change?
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Allow recreation of partitioned topic after metadata loss [pulsar]
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece commented on code in PR #24225:
URL: https://github.com/apache/pulsar/pull/24225#discussion_r2072293154
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java:
##########
@@ -596,12 +598,32 @@ protected void internalCreatePartitionedTopic(AsyncResponse asyncResponse, int n
}
})
.thenCompose(__ -> checkTopicExistsAsync(topicName))
- .thenAccept(exists -> {
- if (exists) {
- log.warn("[{}] Failed to create already existing topic {}", clientAppId(), topicName);
- throw new RestException(Status.CONFLICT, "This topic already exists");
+ .thenAccept(topicExistsInfo -> {
+ try {
+ if (topicExistsInfo.isExists()) {
+ if (topicExistsInfo.getTopicType().equals(TopicType.NON_PARTITIONED)
+ || (topicExistsInfo.getTopicType().equals(TopicType.PARTITIONED)
+ && !createLocalTopicOnly)) {
Review Comment:
@poorbarcode When a partitioned topic will be synchronized to the remote cluster, we allow the remote cluster includes this metadata. The local and remote clusters may be shared a global zk.
createLocalTopicOnly=true means this creation request comes from a cluster, not user client, we can continue to create the partitioned topic if the metadata is existing on the remote cluster.
The pulsar has been covered this changed by existing 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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Allow recreation of partitioned topic after metadata loss [pulsar]
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece commented on code in PR #24225:
URL: https://github.com/apache/pulsar/pull/24225#discussion_r2086469597
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java:
##########
@@ -596,12 +598,32 @@ protected void internalCreatePartitionedTopic(AsyncResponse asyncResponse, int n
}
})
.thenCompose(__ -> checkTopicExistsAsync(topicName))
- .thenAccept(exists -> {
- if (exists) {
- log.warn("[{}] Failed to create already existing topic {}", clientAppId(), topicName);
- throw new RestException(Status.CONFLICT, "This topic already exists");
+ .thenAccept(topicExistsInfo -> {
+ try {
+ if (topicExistsInfo.isExists()) {
+ if (topicExistsInfo.getTopicType().equals(TopicType.NON_PARTITIONED)
+ || (topicExistsInfo.getTopicType().equals(TopicType.PARTITIONED)
+ && !createLocalTopicOnly)) {
Review Comment:
Please stay on topic — partitions and metadata are not the same thing. It seems you haven't looked at the `createLocalTopicOnly` logic. I've already explained this part, but you still seem confused and haven't referred to the code either.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Allow recreation of partitioned topic after metadata loss [pulsar]
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece commented on code in PR #24225:
URL: https://github.com/apache/pulsar/pull/24225#discussion_r2087072938
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java:
##########
@@ -596,12 +598,32 @@ protected void internalCreatePartitionedTopic(AsyncResponse asyncResponse, int n
}
})
.thenCompose(__ -> checkTopicExistsAsync(topicName))
- .thenAccept(exists -> {
- if (exists) {
- log.warn("[{}] Failed to create already existing topic {}", clientAppId(), topicName);
- throw new RestException(Status.CONFLICT, "This topic already exists");
+ .thenAccept(topicExistsInfo -> {
+ try {
+ if (topicExistsInfo.isExists()) {
+ if (topicExistsInfo.getTopicType().equals(TopicType.NON_PARTITIONED)
+ || (topicExistsInfo.getTopicType().equals(TopicType.PARTITIONED)
+ && !createLocalTopicOnly)) {
Review Comment:
When this value is true, which means this request comes from the other cluster, so my check is correct.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Allow recreation of partitioned topic after metadata loss [pulsar]
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece commented on code in PR #24225:
URL: https://github.com/apache/pulsar/pull/24225#discussion_r2085860230
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/AdminResource.java:
##########
@@ -596,12 +598,32 @@ protected void internalCreatePartitionedTopic(AsyncResponse asyncResponse, int n
}
})
.thenCompose(__ -> checkTopicExistsAsync(topicName))
- .thenAccept(exists -> {
- if (exists) {
- log.warn("[{}] Failed to create already existing topic {}", clientAppId(), topicName);
- throw new RestException(Status.CONFLICT, "This topic already exists");
+ .thenAccept(topicExistsInfo -> {
+ try {
+ if (topicExistsInfo.isExists()) {
+ if (topicExistsInfo.getTopicType().equals(TopicType.NON_PARTITIONED)
+ || (topicExistsInfo.getTopicType().equals(TopicType.PARTITIONED)
+ && !createLocalTopicOnly)) {
Review Comment:
@poorbarcode
I've clarified the motivation in the Motivation section. The goal is to address the inconsistency where a partition topic can exist even if the partitioned metadata is missing—something that can happen due to metadata corruption or external manipulation.
Could you explain the flow of the issue that you solved? Just like follows
- There are two clusters `[c1, c2]` enabled Geo-Replication `c1 -> c2`.
- `c2` has partitioned metadata, `c1` has not
- I failed to take the next step.....
```
pulsar-admin topics create-partitioned-topic ci-topic -p 4
# On c1:
# - create metadata
# - create partition 0-4
# - sync this topic to the c2
# On c2:
# - create metadata if not exits
# - create partition 0-4
```
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Allow recreation of partitioned topic after metadata loss [pulsar]
Posted by "liangyepianzhou (via GitHub)" <gi...@apache.org>.
liangyepianzhou commented on code in PR #24225:
URL: https://github.com/apache/pulsar/pull/24225#discussion_r2084329995
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java:
##########
@@ -1412,6 +1412,10 @@ public CompletableFuture<List<String>> getOwnedTopicListForNamespaceBundle(Names
* Check topic exists( partitioned or non-partitioned ).
*/
public CompletableFuture<TopicExistsInfo> checkTopicExists(TopicName topic) {
Review Comment:
Mark the old method as deprecated and point to the new method checkTopicExistsAsync
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org