[improve][broker] defer the ownership checks if the owner is inactive (ExtensibleLoadManager) #21811
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
…ip (ExtensibleLoadManager)
heesung-sn
force-pushed
the
activeOwner
branch
from
6e42ab0
to
c43e55f
Compare
Comment on lines
503
to
504
| ownerAfterDeferred -> ownerAfterDeferred == null ? Optional.empty() | ||
| : Optional.of(ownerAfterDeferred)))) |
There was a problem hiding this comment.
Can we simplify this to Optional.ofNullable?
There was a problem hiding this comment.
thanks. updated.
Comment on lines
1255
to
1256
| var override = getOverrideInactiveBrokerStateData( | ||
| orphanData, selectedBroker, inactiveBroker); |
There was a problem hiding this comment.
Suggested change
| var override = getOverrideInactiveBrokerStateData( | |
| orphanData, selectedBroker, inactiveBroker); | |
| var override = getOverrideInactiveBrokerStateData(orphanData, selectedBroker, inactiveBroker); |
There was a problem hiding this comment.
thanks. updated.
Comment on lines
143
to
145
| } catch (Throwable e) { | ||
| return false; | ||
| } |
There was a problem hiding this comment.
Can we remove this and add ignoreExceptions to the Awaitility chain above, similar to instances below?
| @@ -259,9 +281,11 @@ public void testStopBroker() throws Exception { | |||
| } | |||
| } | |||
|
|
|||
| String broker1 = admin.lookups().lookupTopic(topicName); | |||
| Awaitility.waitAtMost(60, TimeUnit.SECONDS).ignoreExceptions().untilAsserted(() -> { | |||
There was a problem hiding this comment.
The wait period here is larger than the test timeout itself.
Suggested change
| Awaitility.waitAtMost(60, TimeUnit.SECONDS).ignoreExceptions().untilAsserted(() -> { | |
| Awaitility.waitAtMost(40, TimeUnit.SECONDS).ignoreExceptions().untilAsserted(() -> { |
There was a problem hiding this comment.
Increased the test time, instead.
| @@ -380,8 +404,7 @@ public void testIsolationPolicy() throws Exception { | |||
| fail(); | |||
| } catch (Exception ex) { | |||
| log.error("Failed to lookup topic: ", ex); | |||
| assertThat(ex.getMessage()).containsAnyOf("Failed to look up a broker", | |||
| "Failed to select the new owner broker for bundle"); | |||
| assertThat(ex.getMessage()).containsAnyOf("Failed to select the new owner broker for bundle"); | |||
There was a problem hiding this comment.
Nit:
Suggested change
| assertThat(ex.getMessage()).containsAnyOf("Failed to select the new owner broker for bundle"); | |
| assertThat(ex.getMessage()).contains("Failed to select the new owner broker for bundle"); |
Comment on lines
1565
to
1568
| public void testActiveGetOwner() | ||
| throws IllegalAccessException, ExecutionException, InterruptedException, TimeoutException { | ||
|
|
||
|
|
There was a problem hiding this comment.
Suggested change
| public void testActiveGetOwner() | |
| throws IllegalAccessException, ExecutionException, InterruptedException, TimeoutException { | |
| public void testActiveGetOwner() throws Exception { |
There was a problem hiding this comment.
Thanks. updated.
Comment on lines
1589
to
1598
| try { | ||
| channel1.getOwnerAsync(bundle).get(); | ||
| fail(); | ||
| } catch (Exception e) { | ||
| if (e.getCause() instanceof TimeoutException) { | ||
| // expected | ||
| } else { | ||
| fail(); | ||
| } | ||
| } |
There was a problem hiding this comment.
This reads a bit more clearly IMO, but I haven't seen it widely used in our tests:
Suggested change
| try { | |
| channel1.getOwnerAsync(bundle).get(); | |
| fail(); | |
| } catch (Exception e) { | |
| if (e.getCause() instanceof TimeoutException) { | |
| // expected | |
| } else { | |
| fail(); | |
| } | |
| } | |
| var ex = expectThrows(ExecutionException.class, () -> channel1.getOwnerAsync(bundle).get()); | |
| assertTrue(ex.getCause() instanceof TimeoutException); |
There was a problem hiding this comment.
thanks. updated.
| doReturn(CompletableFuture.completedFuture(Optional.empty())) | ||
| .when(loadManager).selectAsync(any(), any()); | ||
| var leaderChannel = channel1; | ||
| String leader = channel1.getChannelOwnerAsync().get(2, TimeUnit.SECONDS).get(); |
There was a problem hiding this comment.
Nit: can we rename this to leader1, to stay consistent with the naming scheme?
There was a problem hiding this comment.
thanks. updated.
...ava/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannelImpl.java
Show resolved
Hide resolved
Comment on lines
+1566
to
+1567
|
|
||
|
|
Demogorgon314
approved these changes
Dec 29, 2023
Codecov ReportAttention:
Additional details and impacted files@@ Coverage Diff @@
## master #21811 +/- ##
============================================
- Coverage 73.58% 73.54% -0.04%
- Complexity 32254 32278 +24
============================================
Files 1858 1858
Lines 138021 138110 +89
Branches 15111 15137 +26
============================================
+ Hits 101558 101579 +21
- Misses 28606 28670 +64
- Partials 7857 7861 +4
Flags with carried forward coverage won't be shown. Click here to find out more.
|
heesung-sn
added a commit
to heesung-sn/pulsar
that referenced
this pull request
Jan 4, 2024
… (ExtensibleLoadManager) (apache#21811)
Technoboy-
pushed a commit
that referenced
this pull request
Jan 16, 2024
… (ExtensibleLoadManager) (#21811)
2 tasks
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Motivation
When a broker shutdown or gets killed, the leader broker will get notified and soon reassign the orphan bundle ownerships. Meanwhile, the ownership lookup to these orphan bundles should be deferred til the new brokers own them. This will further reduce the lookup request retries when brokers are shutdown or get killed.
Modifications
ServiceUnitStateChannelImpl.getOwnerAsync, additionally, check if the current owner broker is active or not in the broker registry. If active, return the current owner. Otherwise, defer the ownership lookup requests til the new owner owns it.Free.writeLoadReportOnZookeeperwriteResourceQuotasToZooKeeperVerifying this change
Added a test case
testActiveGetOwnerto cover this.Does this pull request potentially affect one of the following parts:
If the box was checked, please highlight the changes
Documentation
docdoc-requireddoc-not-neededdoc-completeMatching PR in forked repository
PR in forked repository: heesung-sn#58