[Issue 13854][Broker]Fix call sync method in async rest api for internalExpireMessagesForA… #13905

AnonHxy · 2022-01-23T03:50:02Z

Master Issue: #13854

Motivation

See Issue: #13854

Modifications

Verifying this change

Make sure that the change passes the CI checks.

Does this pull request potentially affect one of the following parts:

If yes was chosen, please highlight the changes

Dependencies (does it add or upgrade a dependency): (no)
The public API: (no)
The schema: (no)
The default values of configurations: (no)
The wire protocol: (no)
The rest endpoints: (no)
The admin cli options: (no)
Anything that affects deployment: (no)

Documentation

no-need-doc

liudezhi2098 · 2022-01-23T11:27:05Z

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java

@@ -3445,6 +3449,92 @@ private void internalExpireMessagesByTimestampForSinglePartition(String subName,
        }
    }

+    private CompletableFuture<Void> internalExpireMessagesByTimestampForSinglePartitionAsync(String subName,


conflict with #13880

comment in #13880

@liudezhi2098 Any difference between these two?

Both PR's modify this method，warning: There may be conflict.

Oh, I think we'd better close the new one and apply the fix from the old one.

Ok, I can rebase the code after #13880 merging to master to fix the conflict

Jason918 · 2022-01-23T11:44:11Z

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java

                        }
-
-                        asyncResponse.resume(Response.noContent().build());
+                    }).exceptionally(ex -> {


Should this exceptionally in Line 1863?

This exceptionally can handle getPartitionedTopicMetadataAsync exceptions if topicName.isPartitioned() return false

This exceptionally can handle getPartitionedTopicMetadataAsync exceptions if topicName.isPartitioned() return false

Yes, but exceptionally is also needed there for validateGlobalNamespaceOwnership

Jason918 · 2022-01-23T11:49:00Z

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java


-        asyncResponse.resume(Response.noContent().build());
+                return FutureUtil.waitForAll(futures).thenRun(() -> {


We can use exceptionally here to handle the exceptions. No need for the exception in Line 1876

Jason918 · 2022-01-23T11:53:55Z

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java

@@ -3445,6 +3449,92 @@ private void internalExpireMessagesByTimestampForSinglePartition(String subName,
        }
    }

+    private CompletableFuture<Void> internalExpireMessagesByTimestampForSinglePartitionAsync(String subName,


@liudezhi2098 Any difference between these two?

Jason918 · 2022-01-23T11:57:22Z

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java

+        return future.thenCompose(__ -> {
+            // If the topic name is a partition name, no need to get partition topic metadata again
+            if (!topicName.isPartitioned()) {
+                getPartitionedTopicMetadataAsync(topicName, authoritative, false)


Yes, there miss return

Jason918 · 2022-01-23T11:59:29Z

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java

+                            log.error("[{}] {} {} {}", clientAppId(), msg, topicName, subName);
+                            throw new IllegalStateException(msg);
+                        } else {
+                            return doInternalExpireMessagesByTimestampForSinglePartitionAsync(subName,


We can return a CompletableFuture.completedFuture(null) if this check is passed to avoid add another method.

If (!topicName.isPartitioned()) && metadata.partitions == 0 , we should also doInternalExpireMessagesByTimestampForSinglePartitionAsync

codelipenghui · 2022-01-24T08:26:33Z

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java

                        }
-
-                        asyncResponse.resume(Response.noContent().build());
+                    }).exceptionally(ex -> {


codelipenghui · 2022-01-24T08:29:48Z

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java

+            .thenCompose(__ -> getTopicReferenceAsync(topicName))
+            .thenCompose(t -> {
+                PersistentTopic topic = (PersistentTopic) t;
+                final AtomicReference<Throwable> exception = new AtomicReference<>();


Why do we need this line? You can get the exception from the futures from line 1879 directly.

looks the same as #13905 (comment)? I will update there code

codelipenghui · 2022-01-24T08:30:59Z

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java

@@ -3445,6 +3449,92 @@ private void internalExpireMessagesByTimestampForSinglePartition(String subName,
        }
    }

+    private CompletableFuture<Void> internalExpireMessagesByTimestampForSinglePartitionAsync(String subName,


Oh, I think we'd better close the new one and apply the fix from the old one.

…llSubscriptions

AnonHxy · 2022-01-25T07:15:31Z

/pulsarbot run-failure-checks

github-actions bot assigned AnonHxy Jan 23, 2022

github-actions bot added the doc-not-needed Your PR changes do not impact docs label Jan 23, 2022

AnonHxy force-pushed the fix_call_sync_call_in_async_internalExpireMessagesForAllSubscriptions branch 2 times, most recently from 943f4d5 to bd06261 Compare January 23, 2022 04:27

AnonHxy changed the title ~~[WIP][Issue 13854][Broker]Fix call sync method in async rest api for internalExpireMessagesForA…~~ [Issue 13854][Broker]Fix call sync method in async rest api for internalExpireMessagesForA… Jan 23, 2022

AnonHxy changed the title ~~[Issue 13854][Broker]Fix call sync method in async rest api for internalExpireMessagesForA…~~ [WIP][Issue 13854][Broker]Fix call sync method in async rest api for internalExpireMessagesForA… Jan 23, 2022

AnonHxy force-pushed the fix_call_sync_call_in_async_internalExpireMessagesForAllSubscriptions branch from bd06261 to 44451b6 Compare January 23, 2022 11:10

AnonHxy changed the title ~~[WIP][Issue 13854][Broker]Fix call sync method in async rest api for internalExpireMessagesForA…~~ [Issue 13854][Broker]Fix call sync method in async rest api for internalExpireMessagesForA… Jan 23, 2022

liudezhi2098 reviewed Jan 23, 2022

View reviewed changes

Jason918 mentioned this pull request Jan 23, 2022

[Broker] PersistentTopicsBase has many async method call sync method. #13854

Closed

26 tasks

Jason918 reviewed Jan 23, 2022

View reviewed changes

AnonHxy force-pushed the fix_call_sync_call_in_async_internalExpireMessagesForAllSubscriptions branch from 44451b6 to 7f18dc6 Compare January 23, 2022 12:43

codelipenghui reviewed Jan 24, 2022

View reviewed changes

Fix call sync method in async rest api for internalExpireMessagesForA…

4ead34d

…llSubscriptions

AnonHxy force-pushed the fix_call_sync_call_in_async_internalExpireMessagesForAllSubscriptions branch from 7f18dc6 to 4ead34d Compare January 24, 2022 13:42

AnonHxy closed this Jan 28, 2022

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[Issue 13854][Broker]Fix call sync method in async rest api for internalExpireMessagesForA… #13905

[Issue 13854][Broker]Fix call sync method in async rest api for internalExpireMessagesForA… #13905

AnonHxy commented Jan 23, 2022 •

edited

Loading

liudezhi2098 Jan 23, 2022

AnonHxy Jan 23, 2022 •

edited

Loading

Jason918 Jan 23, 2022

liudezhi2098 Jan 23, 2022

codelipenghui Jan 24, 2022

AnonHxy Jan 24, 2022

Jason918 Jan 23, 2022

AnonHxy Jan 23, 2022

Jason918 Jan 24, 2022

codelipenghui Jan 24, 2022

AnonHxy Jan 24, 2022

Jason918 Jan 23, 2022

AnonHxy Jan 24, 2022

Jason918 Jan 23, 2022

Jason918 Jan 23, 2022

AnonHxy Jan 23, 2022

Jason918 Jan 23, 2022

AnonHxy Jan 23, 2022

codelipenghui Jan 24, 2022

codelipenghui Jan 24, 2022

AnonHxy Jan 24, 2022

codelipenghui Jan 24, 2022

AnonHxy commented Jan 25, 2022


		asyncResponse.resume(Response.noContent().build());
		return FutureUtil.waitForAll(futures).thenRun(() -> {

[Issue 13854][Broker]Fix call sync method in async rest api for internalExpireMessagesForA… #13905

[Issue 13854][Broker]Fix call sync method in async rest api for internalExpireMessagesForA… #13905

Conversation

AnonHxy commented Jan 23, 2022 • edited Loading

Motivation

Modifications

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

Choose a reason for hiding this comment

AnonHxy Jan 23, 2022 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

AnonHxy commented Jan 25, 2022

AnonHxy commented Jan 23, 2022 •

edited

Loading

AnonHxy Jan 23, 2022 •

edited

Loading