You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "XComp (via GitHub)" <gi...@apache.org> on 2023/06/21 15:27:29 UTC
[GitHub] [flink] XComp opened a new pull request, #22844: [WIP][FLINK-31784][runtime] Adds multi-component support to DefaultLeaderElectionService
XComp opened a new pull request, #22844:
URL: https://github.com/apache/flink/pull/22844
* https://github.com/apache/flink/pull/21742
* https://github.com/apache/flink/pull/22379
* https://github.com/apache/flink/pull/22422
* https://github.com/apache/flink/pull/22380
* https://github.com/apache/flink/pull/22623
* https://github.com/apache/flink/pull/22384
* https://github.com/apache/flink/pull/22390
* https://github.com/apache/flink/pull/22404
* https://github.com/apache/flink/pull/22601
* https://github.com/apache/flink/pull/22642
* https://github.com/apache/flink/pull/22640
* https://github.com/apache/flink/pull/22656
* https://github.com/apache/flink/pull/22828
* https://github.com/apache/flink/pull/22830
* https://github.com/apache/flink/pull/22829
* https://github.com/apache/flink/pull/22661
* === THIS PR === FLINK-31784
## What is the purpose of the change
Replacing the `leaderContender` and `contenderID` fields in `DefaultLeaderElectionService` with a `leaderContenderRegistry` that allows adding multiple contenders.
## Brief change log
* Removes `DefaultLeaderElectionService` fields `leaderContender` and `contenderID`
* Adds `DefaultLeaderElectionService#leaderContenderRegistry`
* Makes `MultpleComponentLeaderElectionDriver.Listener` interface methods first-class citizens and let the `LeaderElectionEventHandler` interface call those
* Introduces `contenderID` in `EmbeddedLeaderElection` to allow logging
* Removes `LeaderContender#getDescription()`
## Verifying this change
* Added multi-component support test scenarios to `DefaultLeaderElectionServiceTest`
* Copied relevant tests over from `DefaultMultipleComponentLeaderElectionServiceTest`
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): no
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: no
- The serializers: no
- The runtime per-record code paths (performance sensitive): no
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: yes
- The S3 file system connector: no
## Documentation
- Does this pull request introduce a new feature? no
- If yes, how is the feature documented? not applicable
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] XComp commented on a diff in pull request #22844: [FLINK-31784][runtime] Adds multi-component support to DefaultLeaderElectionService
Posted by "XComp (via GitHub)" <gi...@apache.org>.
XComp commented on code in PR #22844:
URL: https://github.com/apache/flink/pull/22844#discussion_r1251550146
##########
flink-runtime/src/main/java/org/apache/flink/runtime/leaderelection/DefaultLeaderElectionService.java:
##########
@@ -425,59 +425,72 @@ private void onRevokeLeadershipInternal() {
}
@GuardedBy("lock")
- private void notifyLeaderContenderOfLeadershipLoss() {
+ private void notifyLeaderContenderOfLeadershipLoss(
+ String contenderID, LeaderContender leaderContender) {
Preconditions.checkState(
leaderContender != null,
"The LeaderContender should be always set when calling this method.");
- if (confirmedLeaderInformation.isEmpty()) {
+ if (!confirmedLeaderInformation.hasLeaderInformation(contenderID)) {
LOG.debug(
"Revoking leadership to contender {} while a previous leadership grant wasn't confirmed, yet.",
- leaderContender.getDescription());
+ contenderID);
} else {
LOG.debug(
"Revoking leadership to contender {} for {}.",
- leaderContender.getDescription(),
- LeaderElectionUtils.convertToString(confirmedLeaderInformation));
+ contenderID,
+ LeaderElectionUtils.convertToString(
+ confirmedLeaderInformation.forContenderIDOrEmpty(contenderID)));
}
- confirmedLeaderInformation = LeaderInformation.empty();
+ confirmedLeaderInformation =
+ LeaderInformationRegister.clear(confirmedLeaderInformation, contenderID);
leaderContender.revokeLeadership();
}
@Override
public void onLeaderInformationChange(LeaderInformation leaderInformation) {
- runInLeaderEventThread(() -> onLeaderInformationChangeInternal(leaderInformation));
+ leaderContenderRegistry
+ .keySet()
+ .forEach(
+ contenderID ->
+ notifyLeaderInformationChange(contenderID, leaderInformation));
}
@GuardedBy("lock")
- private void onLeaderInformationChangeInternal(LeaderInformation leaderInformation) {
- if (leaderContender != null) {
- LOG.trace(
- "Leader node changed while {} is the leader with {}. New leader information {}.",
- leaderContender.getDescription(),
- LeaderElectionUtils.convertToString(confirmedLeaderInformation),
- LeaderElectionUtils.convertToString(leaderInformation));
- if (!confirmedLeaderInformation.isEmpty()) {
- final LeaderInformation confirmedLeaderInfo = this.confirmedLeaderInformation;
- if (leaderInformation.isEmpty()) {
- LOG.debug(
- "Writing leader information by {} since the external storage is empty.",
- leaderContender.getDescription());
- leaderElectionDriver.publishLeaderInformation(contenderID, confirmedLeaderInfo);
- } else if (!leaderInformation.equals(confirmedLeaderInfo)) {
- // the data field does not correspond to the expected leader information
- LOG.debug(
- "Correcting leader information by {}.",
- leaderContender.getDescription());
- leaderElectionDriver.publishLeaderInformation(contenderID, confirmedLeaderInfo);
- }
- }
- } else {
- LOG.debug(
- "Ignoring change notification since the {} has already been stopped.",
- leaderElectionDriver);
- }
+ private void onLeaderInformationChangeInternal(LeaderInformationRegister leaderInformation) {
+ leaderContenderRegistry.forEach(
+ (contenderID, leaderContender) -> {
+ final LeaderInformation newLeaderInformationForContender =
+ leaderInformation
+ .forContenderID(contenderID)
+ .orElse(LeaderInformation.empty());
+ final LeaderInformation confirmedLeaderInformationForContender =
+ confirmedLeaderInformation.forContenderIDOrEmpty(contenderID);
+
+ if (confirmedLeaderInformationForContender.isEmpty()) {
+ LOG.trace(
+ "Leader information changed while there's no confirmation available by the contender for contender ID '{}' for this session. Changed leader information {} will be ignored.",
+ contenderID,
+ LeaderElectionUtils.convertToString(
+ newLeaderInformationForContender));
Review Comment:
I also noticed that the `notifyLeaderInformationChange` method is actually implemented in a wrong way: The `LeaderInformationRegister` is used in the wrong way. I have to get back to that one. :thinking:
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] XComp commented on a diff in pull request #22844: [FLINK-31784][runtime] Adds multi-component support to DefaultLeaderElectionService
Posted by "XComp (via GitHub)" <gi...@apache.org>.
XComp commented on code in PR #22844:
URL: https://github.com/apache/flink/pull/22844#discussion_r1251550146
##########
flink-runtime/src/main/java/org/apache/flink/runtime/leaderelection/DefaultLeaderElectionService.java:
##########
@@ -425,59 +425,72 @@ private void onRevokeLeadershipInternal() {
}
@GuardedBy("lock")
- private void notifyLeaderContenderOfLeadershipLoss() {
+ private void notifyLeaderContenderOfLeadershipLoss(
+ String contenderID, LeaderContender leaderContender) {
Preconditions.checkState(
leaderContender != null,
"The LeaderContender should be always set when calling this method.");
- if (confirmedLeaderInformation.isEmpty()) {
+ if (!confirmedLeaderInformation.hasLeaderInformation(contenderID)) {
LOG.debug(
"Revoking leadership to contender {} while a previous leadership grant wasn't confirmed, yet.",
- leaderContender.getDescription());
+ contenderID);
} else {
LOG.debug(
"Revoking leadership to contender {} for {}.",
- leaderContender.getDescription(),
- LeaderElectionUtils.convertToString(confirmedLeaderInformation));
+ contenderID,
+ LeaderElectionUtils.convertToString(
+ confirmedLeaderInformation.forContenderIDOrEmpty(contenderID)));
}
- confirmedLeaderInformation = LeaderInformation.empty();
+ confirmedLeaderInformation =
+ LeaderInformationRegister.clear(confirmedLeaderInformation, contenderID);
leaderContender.revokeLeadership();
}
@Override
public void onLeaderInformationChange(LeaderInformation leaderInformation) {
- runInLeaderEventThread(() -> onLeaderInformationChangeInternal(leaderInformation));
+ leaderContenderRegistry
+ .keySet()
+ .forEach(
+ contenderID ->
+ notifyLeaderInformationChange(contenderID, leaderInformation));
}
@GuardedBy("lock")
- private void onLeaderInformationChangeInternal(LeaderInformation leaderInformation) {
- if (leaderContender != null) {
- LOG.trace(
- "Leader node changed while {} is the leader with {}. New leader information {}.",
- leaderContender.getDescription(),
- LeaderElectionUtils.convertToString(confirmedLeaderInformation),
- LeaderElectionUtils.convertToString(leaderInformation));
- if (!confirmedLeaderInformation.isEmpty()) {
- final LeaderInformation confirmedLeaderInfo = this.confirmedLeaderInformation;
- if (leaderInformation.isEmpty()) {
- LOG.debug(
- "Writing leader information by {} since the external storage is empty.",
- leaderContender.getDescription());
- leaderElectionDriver.publishLeaderInformation(contenderID, confirmedLeaderInfo);
- } else if (!leaderInformation.equals(confirmedLeaderInfo)) {
- // the data field does not correspond to the expected leader information
- LOG.debug(
- "Correcting leader information by {}.",
- leaderContender.getDescription());
- leaderElectionDriver.publishLeaderInformation(contenderID, confirmedLeaderInfo);
- }
- }
- } else {
- LOG.debug(
- "Ignoring change notification since the {} has already been stopped.",
- leaderElectionDriver);
- }
+ private void onLeaderInformationChangeInternal(LeaderInformationRegister leaderInformation) {
+ leaderContenderRegistry.forEach(
+ (contenderID, leaderContender) -> {
+ final LeaderInformation newLeaderInformationForContender =
+ leaderInformation
+ .forContenderID(contenderID)
+ .orElse(LeaderInformation.empty());
+ final LeaderInformation confirmedLeaderInformationForContender =
+ confirmedLeaderInformation.forContenderIDOrEmpty(contenderID);
+
+ if (confirmedLeaderInformationForContender.isEmpty()) {
+ LOG.trace(
+ "Leader information changed while there's no confirmation available by the contender for contender ID '{}' for this session. Changed leader information {} will be ignored.",
+ contenderID,
+ LeaderElectionUtils.convertToString(
+ newLeaderInformationForContender));
Review Comment:
I also noticed that the `notifyLeaderInformationChange` method is actually implemented in a wrong way: The `LeaderInformationRegister` is used in the wrong way. I have to get back to that one. :thinking:
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] XComp commented on a diff in pull request #22844: [FLINK-31784][runtime] Adds multi-component support to DefaultLeaderElectionService
Posted by "XComp (via GitHub)" <gi...@apache.org>.
XComp commented on code in PR #22844:
URL: https://github.com/apache/flink/pull/22844#discussion_r1251551929
##########
flink-runtime/src/test/java/org/apache/flink/runtime/leaderelection/DefaultLeaderElectionServiceTest.java:
##########
@@ -349,22 +380,42 @@ void testProperCleanupOnLeaderElectionCloseWhenHoldingTheLeadership() throws Exc
final UUID leaderSessionID = UUID.randomUUID();
grantLeadership(leaderSessionID);
- assertThat(testingContender.getLeaderSessionID()).isNotNull();
- assertThat(leaderElectionService.getLeaderSessionID(contenderID))
- .isEqualTo(leaderSessionID);
- assertThat(storedLeaderInformation.get().forContenderID(contenderID))
- .hasValue(LeaderInformation.known(leaderSessionID, TEST_URL));
+ applyToBothContenderContexts(
+ ctx -> {
+ assertThat(ctx.contender.getLeaderSessionID())
+ .isEqualTo(leaderSessionID);
+ assertThat(
+ leaderElectionService.getLeaderSessionID(
+ ctx.contenderID))
+ .isEqualTo(leaderSessionID);
+
+ assertThat(
+ leaderElectionService.getLeaderSessionID(
+ ctx.contenderID))
+ .isEqualTo(leaderSessionID);
+
+ assertThat(
+ storedLeaderInformation
+ .get()
+ .forContenderID(ctx.contenderID))
+ .hasValue(
+ LeaderInformation.known(
+ leaderSessionID, ctx.address));
Review Comment:
yeah, I tried to reduce it already by providing the context members in the callback function instead of the context struct itself. That way, we would avoid having `ctx.` in the code. But it' doesn't help as much as I hoped.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] XComp commented on a diff in pull request #22844: [FLINK-31784][runtime] Adds multi-component support to DefaultLeaderElectionService
Posted by "XComp (via GitHub)" <gi...@apache.org>.
XComp commented on code in PR #22844:
URL: https://github.com/apache/flink/pull/22844#discussion_r1251031367
##########
flink-runtime/src/main/java/org/apache/flink/runtime/leaderelection/DefaultLeaderElectionService.java:
##########
@@ -189,66 +179,62 @@ protected void register(String contenderID, LeaderContender contender) throws Ex
checkNotNull(contender, "Contender must not be null.");
synchronized (lock) {
- Preconditions.checkState(
- leaderContender == null,
- "Only one LeaderContender is allowed to be registered to this service.");
- Preconditions.checkState(
- this.contenderID == null, "The contenderID is only allowed to be set once.");
Preconditions.checkState(
running,
"The DefaultLeaderElectionService should have established a connection to the backend before it's started.");
- leaderContender = contender;
- this.contenderID = contenderID;
+ leaderContenderRegistry.put(contenderID, contender);
LOG.info(
"LeaderContender {} has been registered for {}.",
- contender.getDescription(),
+ contenderID,
leaderElectionDriver);
if (issuedLeaderSessionID != null) {
// notifying the LeaderContender shouldn't happen in the contender's main thread
runInLeaderEventThread(
- () -> notifyLeaderContenderOfLeadership(issuedLeaderSessionID));
+ () ->
+ notifyLeaderContenderOfLeadership(
+ contenderID, issuedLeaderSessionID));
}
}
}
@Override
protected final void remove(String contenderID) {
synchronized (lock) {
- if (this.contenderID == null) {
+ if (!leaderContenderRegistry.containsKey(contenderID)) {
LOG.debug(
"The stop procedure was called on an already stopped DefaultLeaderElectionService instance. No action necessary.");
Review Comment:
You're right - I will update the message. We're not really "stopping" here anymore. That's some log message that survived the refactoring and should be rephrased.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] XComp commented on pull request #22844: [FLINK-31784][runtime] Adds multi-component support to DefaultLeaderElectionService
Posted by "XComp (via GitHub)" <gi...@apache.org>.
XComp commented on PR #22844:
URL: https://github.com/apache/flink/pull/22844#issuecomment-1614818534
Force-pushed a rebase to the most-recent version of [FLINK-31783](https://issues.apache.org/jira/browse/FLINK-31783) to include `master` which appears to be more stable now
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] zentol commented on a diff in pull request #22844: [FLINK-31784][runtime] Adds multi-component support to DefaultLeaderElectionService
Posted by "zentol (via GitHub)" <gi...@apache.org>.
zentol commented on code in PR #22844:
URL: https://github.com/apache/flink/pull/22844#discussion_r1251908493
##########
flink-runtime/src/main/java/org/apache/flink/runtime/leaderelection/DefaultLeaderElectionService.java:
##########
@@ -425,59 +425,72 @@ private void onRevokeLeadershipInternal() {
}
@GuardedBy("lock")
- private void notifyLeaderContenderOfLeadershipLoss() {
+ private void notifyLeaderContenderOfLeadershipLoss(
+ String contenderID, LeaderContender leaderContender) {
Preconditions.checkState(
leaderContender != null,
"The LeaderContender should be always set when calling this method.");
- if (confirmedLeaderInformation.isEmpty()) {
+ if (!confirmedLeaderInformation.hasLeaderInformation(contenderID)) {
LOG.debug(
"Revoking leadership to contender {} while a previous leadership grant wasn't confirmed, yet.",
- leaderContender.getDescription());
+ contenderID);
} else {
LOG.debug(
"Revoking leadership to contender {} for {}.",
- leaderContender.getDescription(),
- LeaderElectionUtils.convertToString(confirmedLeaderInformation));
+ contenderID,
+ LeaderElectionUtils.convertToString(
+ confirmedLeaderInformation.forContenderIDOrEmpty(contenderID)));
}
- confirmedLeaderInformation = LeaderInformation.empty();
+ confirmedLeaderInformation =
+ LeaderInformationRegister.clear(confirmedLeaderInformation, contenderID);
leaderContender.revokeLeadership();
}
@Override
public void onLeaderInformationChange(LeaderInformation leaderInformation) {
- runInLeaderEventThread(() -> onLeaderInformationChangeInternal(leaderInformation));
+ leaderContenderRegistry
+ .keySet()
+ .forEach(
+ contenderID ->
+ notifyLeaderInformationChange(contenderID, leaderInformation));
}
@GuardedBy("lock")
- private void onLeaderInformationChangeInternal(LeaderInformation leaderInformation) {
- if (leaderContender != null) {
- LOG.trace(
- "Leader node changed while {} is the leader with {}. New leader information {}.",
- leaderContender.getDescription(),
- LeaderElectionUtils.convertToString(confirmedLeaderInformation),
- LeaderElectionUtils.convertToString(leaderInformation));
- if (!confirmedLeaderInformation.isEmpty()) {
- final LeaderInformation confirmedLeaderInfo = this.confirmedLeaderInformation;
- if (leaderInformation.isEmpty()) {
- LOG.debug(
- "Writing leader information by {} since the external storage is empty.",
- leaderContender.getDescription());
- leaderElectionDriver.publishLeaderInformation(contenderID, confirmedLeaderInfo);
- } else if (!leaderInformation.equals(confirmedLeaderInfo)) {
- // the data field does not correspond to the expected leader information
- LOG.debug(
- "Correcting leader information by {}.",
- leaderContender.getDescription());
- leaderElectionDriver.publishLeaderInformation(contenderID, confirmedLeaderInfo);
- }
- }
- } else {
- LOG.debug(
- "Ignoring change notification since the {} has already been stopped.",
- leaderElectionDriver);
- }
+ private void onLeaderInformationChangeInternal(LeaderInformationRegister leaderInformation) {
+ leaderContenderRegistry.forEach(
+ (contenderID, leaderContender) -> {
+ final LeaderInformation newLeaderInformationForContender =
+ leaderInformation
+ .forContenderID(contenderID)
+ .orElse(LeaderInformation.empty());
+ final LeaderInformation confirmedLeaderInformationForContender =
+ confirmedLeaderInformation.forContenderIDOrEmpty(contenderID);
+
+ if (confirmedLeaderInformationForContender.isEmpty()) {
+ LOG.trace(
+ "Leader information changed while there's no confirmation available by the contender for contender ID '{}' for this session. Changed leader information {} will be ignored.",
+ contenderID,
+ LeaderElectionUtils.convertToString(
+ newLeaderInformationForContender));
Review Comment:
> But thinking about it: We might want to write the empty LeaderInformation to the HA backend in that case to avoid that monitoring services pick up the changed (false) LeaderInformation from the HA backend while the confirmLeadership call is delayed. thinking
On the flip-side, we don't know if the leader is ready yet to accept requests yet? (because for example the dispatcher is still starting) :thinking:
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] XComp commented on a diff in pull request #22844: [FLINK-31784][runtime] Adds multi-component support to DefaultLeaderElectionService
Posted by "XComp (via GitHub)" <gi...@apache.org>.
XComp commented on code in PR #22844:
URL: https://github.com/apache/flink/pull/22844#discussion_r1252041691
##########
flink-runtime/src/main/java/org/apache/flink/runtime/leaderelection/DefaultLeaderElectionService.java:
##########
@@ -189,66 +179,62 @@ protected void register(String contenderID, LeaderContender contender) throws Ex
checkNotNull(contender, "Contender must not be null.");
synchronized (lock) {
- Preconditions.checkState(
- leaderContender == null,
- "Only one LeaderContender is allowed to be registered to this service.");
- Preconditions.checkState(
- this.contenderID == null, "The contenderID is only allowed to be set once.");
Preconditions.checkState(
running,
"The DefaultLeaderElectionService should have established a connection to the backend before it's started.");
- leaderContender = contender;
- this.contenderID = contenderID;
+ leaderContenderRegistry.put(contenderID, contender);
Review Comment:
> Side-question: Are there any safeguards that there is at most 1 contender for a given leadership role?
This is mostly relevant for testing of course; but what would happen if 2 JMs would use the same LES? They'd both become leader?
That's essentially the safe-guard you were talking about in your [general review comment](https://github.com/apache/flink/pull/22844#pullrequestreview-1511266413)
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] zentol commented on a diff in pull request #22844: [FLINK-31784][runtime] Adds multi-component support to DefaultLeaderElectionService
Posted by "zentol (via GitHub)" <gi...@apache.org>.
zentol commented on code in PR #22844:
URL: https://github.com/apache/flink/pull/22844#discussion_r1250941228
##########
flink-runtime/src/main/java/org/apache/flink/runtime/leaderelection/DefaultLeaderElectionService.java:
##########
@@ -189,66 +179,62 @@ protected void register(String contenderID, LeaderContender contender) throws Ex
checkNotNull(contender, "Contender must not be null.");
synchronized (lock) {
- Preconditions.checkState(
- leaderContender == null,
- "Only one LeaderContender is allowed to be registered to this service.");
- Preconditions.checkState(
- this.contenderID == null, "The contenderID is only allowed to be set once.");
Preconditions.checkState(
running,
"The DefaultLeaderElectionService should have established a connection to the backend before it's started.");
- leaderContender = contender;
- this.contenderID = contenderID;
+ leaderContenderRegistry.put(contenderID, contender);
LOG.info(
"LeaderContender {} has been registered for {}.",
- contender.getDescription(),
+ contenderID,
leaderElectionDriver);
if (issuedLeaderSessionID != null) {
// notifying the LeaderContender shouldn't happen in the contender's main thread
runInLeaderEventThread(
- () -> notifyLeaderContenderOfLeadership(issuedLeaderSessionID));
+ () ->
+ notifyLeaderContenderOfLeadership(
+ contenderID, issuedLeaderSessionID));
}
}
}
@Override
protected final void remove(String contenderID) {
synchronized (lock) {
- if (this.contenderID == null) {
+ if (!leaderContenderRegistry.containsKey(contenderID)) {
LOG.debug(
"The stop procedure was called on an already stopped DefaultLeaderElectionService instance. No action necessary.");
Review Comment:
update message to include the contender and clarify the the LES may still be running?
##########
flink-runtime/src/main/java/org/apache/flink/runtime/leaderelection/DefaultLeaderElectionService.java:
##########
@@ -189,66 +179,62 @@ protected void register(String contenderID, LeaderContender contender) throws Ex
checkNotNull(contender, "Contender must not be null.");
synchronized (lock) {
- Preconditions.checkState(
- leaderContender == null,
- "Only one LeaderContender is allowed to be registered to this service.");
- Preconditions.checkState(
- this.contenderID == null, "The contenderID is only allowed to be set once.");
Preconditions.checkState(
running,
"The DefaultLeaderElectionService should have established a connection to the backend before it's started.");
- leaderContender = contender;
- this.contenderID = contenderID;
+ leaderContenderRegistry.put(contenderID, contender);
Review Comment:
check that the previous value was null?
##########
flink-runtime/src/test/java/org/apache/flink/runtime/leaderelection/DefaultLeaderElectionServiceTest.java:
##########
@@ -932,38 +1064,71 @@ void runTestWithManuallyTriggeredEvents(
void runTest(RunnableWithException testMethod, ExecutorService leaderEventOperationExecutor)
throws Exception {
- try {
- leaderElectionService =
- new DefaultLeaderElectionService(
- driverFactory,
- DefaultLeaderElectionServiceTest.this.fatalErrorHandlerExtension
- .getTestingFatalErrorHandler(),
- leaderEventOperationExecutor);
+ try (final DefaultLeaderElectionService localLeaderElectionService =
+ new DefaultLeaderElectionService(
+ driverFactory,
+ DefaultLeaderElectionServiceTest.this.fatalErrorHandlerExtension
+ .getTestingFatalErrorHandler(),
+ leaderEventOperationExecutor)) {
+ leaderElectionService = localLeaderElectionService;
leaderElectionService.startLeaderElectionBackend();
testingLeaderElectionDriver = driverFactory.assertAndGetOnlyCreatedDriver();
- leaderElection = leaderElectionService.createLeaderElection(contenderID);
- testingContender = new TestingContender(TEST_URL, leaderElection);
- testingContender.startLeaderElection();
+ try (final ContenderContext localContenderContext0 =
+ ContenderContext.create(0, leaderElectionService);
+ final ContenderContext localContenderContext1 =
+ ContenderContext.create(1, leaderElectionService)) {
+ this.contenderContext0 = localContenderContext0;
+ this.contenderContext1 = localContenderContext1;
Review Comment:
Why are we creating 2 contenders now? Purely to have all tests also apply to cases with multiple contenders?
##########
flink-runtime/src/test/java/org/apache/flink/runtime/leaderelection/LeaderInformationRegisterTest.java:
##########
@@ -0,0 +1,244 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.leaderelection;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.UUID;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+class LeaderInformationRegisterTest {
+
+ @Test
+ void testOfWithKnownLeaderInformation() {
+ final String contenderID = "contender-id";
+ final LeaderInformation leaderInformation =
+ LeaderInformation.known(UUID.randomUUID(), "address");
+
+ final LeaderInformationRegister testInstance =
+ LeaderInformationRegister.of(contenderID, leaderInformation);
+ assertThat(testInstance.getRegisteredContenderIDs()).containsExactly(contenderID);
+ assertThat(testInstance.forContenderID(contenderID)).hasValue(leaderInformation);
+ }
+
+ @Test
+ void testOfWithEmptyLeaderInformation() {
+ final String contenderID = "contender-id";
+ final LeaderInformationRegister testInstance =
+ LeaderInformationRegister.of(contenderID, LeaderInformation.empty());
+
+ assertThat(testInstance.getRegisteredContenderIDs()).isEmpty();
+ assertThat(testInstance.forContenderID(contenderID)).isNotPresent();
+ }
+
+ @Test
+ void testMerge() {
+ final String contenderID = "contender-id";
+ final LeaderInformation leaderInformation =
+ LeaderInformation.known(UUID.randomUUID(), "address");
+
+ final String newContenderID = "new-contender-id";
+ final LeaderInformation newLeaderInformation =
+ LeaderInformation.known(UUID.randomUUID(), "new-address");
+
+ final LeaderInformationRegister initialRegister =
+ LeaderInformationRegister.of(contenderID, leaderInformation);
+
+ final LeaderInformationRegister newRegister =
+ LeaderInformationRegister.merge(
+ initialRegister, newContenderID, newLeaderInformation);
+
+ assertThat(newRegister).isNotSameAs(initialRegister);
+ assertThat(newRegister.getRegisteredContenderIDs())
+ .containsExactlyInAnyOrder(contenderID, newContenderID);
+ assertThat(newRegister.forContenderID(contenderID)).hasValue(leaderInformation);
+ assertThat(newRegister.forContenderID(newContenderID)).hasValue(newLeaderInformation);
+ }
+
+ @Test
+ void testMergeEmptyLeaderInformation() {
+ final String contenderID = "contender-id";
+ final LeaderInformation leaderInformation =
+ LeaderInformation.known(UUID.randomUUID(), "address");
+
+ final String newContenderID = "new-contender-id";
+
+ final LeaderInformationRegister initialRegister =
+ LeaderInformationRegister.of(contenderID, leaderInformation);
+
+ final LeaderInformationRegister newRegister =
+ LeaderInformationRegister.merge(
+ initialRegister, newContenderID, LeaderInformation.empty());
+
+ assertThat(newRegister).isNotSameAs(initialRegister);
+ assertThat(newRegister.getRegisteredContenderIDs()).containsExactly(contenderID);
+ assertThat(newRegister.forContenderID(newContenderID)).isNotPresent();
+ }
+
+ @Test
+ void testMergeEmptyLeaderInformationForExistingContenderID() {
Review Comment:
for completeness a test where we overwrite existing leader information with new (non-empty) leader information would be nice.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/leaderelection/DefaultLeaderElectionService.java:
##########
@@ -503,43 +516,53 @@ private void runInLeaderEventThread(Runnable callback) {
private void forwardErrorToLeaderContender(Throwable t) {
synchronized (lock) {
- if (leaderContender == null) {
+ if (leaderContenderRegistry.isEmpty()) {
fallbackErrorHandler.onFatalError(t);
return;
}
- if (t instanceof LeaderElectionException) {
- leaderContender.handleError((LeaderElectionException) t);
- } else {
- leaderContender.handleError(new LeaderElectionException(t));
- }
+ leaderContenderRegistry
+ .values()
+ .forEach(
+ leaderContender -> {
+ if (leaderContender == null) {
Review Comment:
:thinking: When can the value of the registry map be null? Even if that were possible (which I dont think it is), then we'd logging this message once per contender.
##########
flink-runtime/src/test/java/org/apache/flink/runtime/leaderelection/DefaultLeaderElectionServiceTest.java:
##########
@@ -349,22 +380,42 @@ void testProperCleanupOnLeaderElectionCloseWhenHoldingTheLeadership() throws Exc
final UUID leaderSessionID = UUID.randomUUID();
grantLeadership(leaderSessionID);
- assertThat(testingContender.getLeaderSessionID()).isNotNull();
- assertThat(leaderElectionService.getLeaderSessionID(contenderID))
- .isEqualTo(leaderSessionID);
- assertThat(storedLeaderInformation.get().forContenderID(contenderID))
- .hasValue(LeaderInformation.known(leaderSessionID, TEST_URL));
+ applyToBothContenderContexts(
+ ctx -> {
+ assertThat(ctx.contender.getLeaderSessionID())
+ .isEqualTo(leaderSessionID);
+ assertThat(
+ leaderElectionService.getLeaderSessionID(
+ ctx.contenderID))
+ .isEqualTo(leaderSessionID);
+
+ assertThat(
+ leaderElectionService.getLeaderSessionID(
+ ctx.contenderID))
+ .isEqualTo(leaderSessionID);
+
+ assertThat(
+ storedLeaderInformation
+ .get()
+ .forContenderID(ctx.contenderID))
+ .hasValue(
+ LeaderInformation.known(
+ leaderSessionID, ctx.address));
Review Comment:
look at that magnificent formatting magic, going from 5 lines 18.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/leaderelection/DefaultLeaderElectionService.java:
##########
@@ -425,59 +425,72 @@ private void onRevokeLeadershipInternal() {
}
@GuardedBy("lock")
- private void notifyLeaderContenderOfLeadershipLoss() {
+ private void notifyLeaderContenderOfLeadershipLoss(
+ String contenderID, LeaderContender leaderContender) {
Preconditions.checkState(
leaderContender != null,
"The LeaderContender should be always set when calling this method.");
- if (confirmedLeaderInformation.isEmpty()) {
+ if (!confirmedLeaderInformation.hasLeaderInformation(contenderID)) {
LOG.debug(
"Revoking leadership to contender {} while a previous leadership grant wasn't confirmed, yet.",
- leaderContender.getDescription());
+ contenderID);
} else {
LOG.debug(
"Revoking leadership to contender {} for {}.",
- leaderContender.getDescription(),
- LeaderElectionUtils.convertToString(confirmedLeaderInformation));
+ contenderID,
+ LeaderElectionUtils.convertToString(
+ confirmedLeaderInformation.forContenderIDOrEmpty(contenderID)));
}
- confirmedLeaderInformation = LeaderInformation.empty();
+ confirmedLeaderInformation =
+ LeaderInformationRegister.clear(confirmedLeaderInformation, contenderID);
leaderContender.revokeLeadership();
}
@Override
public void onLeaderInformationChange(LeaderInformation leaderInformation) {
- runInLeaderEventThread(() -> onLeaderInformationChangeInternal(leaderInformation));
+ leaderContenderRegistry
+ .keySet()
+ .forEach(
+ contenderID ->
+ notifyLeaderInformationChange(contenderID, leaderInformation));
}
@GuardedBy("lock")
- private void onLeaderInformationChangeInternal(LeaderInformation leaderInformation) {
- if (leaderContender != null) {
- LOG.trace(
- "Leader node changed while {} is the leader with {}. New leader information {}.",
- leaderContender.getDescription(),
- LeaderElectionUtils.convertToString(confirmedLeaderInformation),
- LeaderElectionUtils.convertToString(leaderInformation));
- if (!confirmedLeaderInformation.isEmpty()) {
- final LeaderInformation confirmedLeaderInfo = this.confirmedLeaderInformation;
- if (leaderInformation.isEmpty()) {
- LOG.debug(
- "Writing leader information by {} since the external storage is empty.",
- leaderContender.getDescription());
- leaderElectionDriver.publishLeaderInformation(contenderID, confirmedLeaderInfo);
- } else if (!leaderInformation.equals(confirmedLeaderInfo)) {
- // the data field does not correspond to the expected leader information
- LOG.debug(
- "Correcting leader information by {}.",
- leaderContender.getDescription());
- leaderElectionDriver.publishLeaderInformation(contenderID, confirmedLeaderInfo);
- }
- }
- } else {
- LOG.debug(
- "Ignoring change notification since the {} has already been stopped.",
- leaderElectionDriver);
- }
+ private void onLeaderInformationChangeInternal(LeaderInformationRegister leaderInformation) {
+ leaderContenderRegistry.forEach(
+ (contenderID, leaderContender) -> {
+ final LeaderInformation newLeaderInformationForContender =
+ leaderInformation
+ .forContenderID(contenderID)
+ .orElse(LeaderInformation.empty());
+ final LeaderInformation confirmedLeaderInformationForContender =
+ confirmedLeaderInformation.forContenderIDOrEmpty(contenderID);
+
+ if (confirmedLeaderInformationForContender.isEmpty()) {
+ LOG.trace(
+ "Leader information changed while there's no confirmation available by the contender for contender ID '{}' for this session. Changed leader information {} will be ignored.",
+ contenderID,
+ LeaderElectionUtils.convertToString(
+ newLeaderInformationForContender));
Review Comment:
I have trouble understanding this situation. A contender got leadership, but hasn't confirmed it yet? Shouldn't we then put the leadership loss on the wire anyway? When is the contender informed about the leadership loss otherwise?
##########
flink-runtime/src/main/java/org/apache/flink/runtime/leaderelection/DefaultLeaderElectionService.java:
##########
@@ -349,15 +342,17 @@ protected boolean hasLeadership(String contenderID, UUID leaderSessionId) {
@Nullable
public UUID getLeaderSessionID(String contenderID) {
synchronized (lock) {
- return contenderID.equals(this.contenderID)
- ? confirmedLeaderInformation.getLeaderSessionID()
+ return leaderContenderRegistry.containsKey(contenderID)
+ ? confirmedLeaderInformation
+ .forContenderIDOrEmpty(contenderID)
Review Comment:
This method names looks weird; maybe rename to `forContenderIdOrEmpty`?
##########
flink-runtime/src/test/java/org/apache/flink/runtime/leaderelection/DefaultLeaderElectionServiceTest.java:
##########
@@ -597,14 +676,30 @@ void testOnLeaderInformationChangeIsIgnoredAfterLeaderElectionBeingStop() throws
() -> {
grantLeadership();
- assertThat(storedLeaderInformation.get().forContenderID(contenderID))
- .isPresent();
+ applyToBothContenderContexts(
+ ctx ->
+ assertThat(
+ storedLeaderInformation
+ .get()
+ .forContenderID(
+ ctx.contenderID))
+ .isPresent());
+
+ contenderContext0.leaderElection.close();
Review Comment:
why isn't this done within `applyToBothContenderContexts`? it runs the same logic for both contenders, no?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] XComp commented on pull request #22844: [FLINK-31784][runtime] Adds multi-component support to DefaultLeaderElectionService
Posted by "XComp (via GitHub)" <gi...@apache.org>.
XComp commented on PR #22844:
URL: https://github.com/apache/flink/pull/22844#issuecomment-1620269151
I addressed your changes. I will go ahead and do another force-push to fix the conflicts.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] XComp commented on a diff in pull request #22844: [FLINK-31784][runtime] Adds multi-component support to DefaultLeaderElectionService
Posted by "XComp (via GitHub)" <gi...@apache.org>.
XComp commented on code in PR #22844:
URL: https://github.com/apache/flink/pull/22844#discussion_r1252040025
##########
flink-runtime/src/main/java/org/apache/flink/runtime/leaderelection/DefaultLeaderElectionService.java:
##########
@@ -425,59 +425,72 @@ private void onRevokeLeadershipInternal() {
}
@GuardedBy("lock")
- private void notifyLeaderContenderOfLeadershipLoss() {
+ private void notifyLeaderContenderOfLeadershipLoss(
+ String contenderID, LeaderContender leaderContender) {
Preconditions.checkState(
leaderContender != null,
"The LeaderContender should be always set when calling this method.");
- if (confirmedLeaderInformation.isEmpty()) {
+ if (!confirmedLeaderInformation.hasLeaderInformation(contenderID)) {
LOG.debug(
"Revoking leadership to contender {} while a previous leadership grant wasn't confirmed, yet.",
- leaderContender.getDescription());
+ contenderID);
} else {
LOG.debug(
"Revoking leadership to contender {} for {}.",
- leaderContender.getDescription(),
- LeaderElectionUtils.convertToString(confirmedLeaderInformation));
+ contenderID,
+ LeaderElectionUtils.convertToString(
+ confirmedLeaderInformation.forContenderIDOrEmpty(contenderID)));
}
- confirmedLeaderInformation = LeaderInformation.empty();
+ confirmedLeaderInformation =
+ LeaderInformationRegister.clear(confirmedLeaderInformation, contenderID);
leaderContender.revokeLeadership();
}
@Override
public void onLeaderInformationChange(LeaderInformation leaderInformation) {
- runInLeaderEventThread(() -> onLeaderInformationChangeInternal(leaderInformation));
+ leaderContenderRegistry
+ .keySet()
+ .forEach(
+ contenderID ->
+ notifyLeaderInformationChange(contenderID, leaderInformation));
}
@GuardedBy("lock")
- private void onLeaderInformationChangeInternal(LeaderInformation leaderInformation) {
- if (leaderContender != null) {
- LOG.trace(
- "Leader node changed while {} is the leader with {}. New leader information {}.",
- leaderContender.getDescription(),
- LeaderElectionUtils.convertToString(confirmedLeaderInformation),
- LeaderElectionUtils.convertToString(leaderInformation));
- if (!confirmedLeaderInformation.isEmpty()) {
- final LeaderInformation confirmedLeaderInfo = this.confirmedLeaderInformation;
- if (leaderInformation.isEmpty()) {
- LOG.debug(
- "Writing leader information by {} since the external storage is empty.",
- leaderContender.getDescription());
- leaderElectionDriver.publishLeaderInformation(contenderID, confirmedLeaderInfo);
- } else if (!leaderInformation.equals(confirmedLeaderInfo)) {
- // the data field does not correspond to the expected leader information
- LOG.debug(
- "Correcting leader information by {}.",
- leaderContender.getDescription());
- leaderElectionDriver.publishLeaderInformation(contenderID, confirmedLeaderInfo);
- }
- }
- } else {
- LOG.debug(
- "Ignoring change notification since the {} has already been stopped.",
- leaderElectionDriver);
- }
+ private void onLeaderInformationChangeInternal(LeaderInformationRegister leaderInformation) {
+ leaderContenderRegistry.forEach(
+ (contenderID, leaderContender) -> {
+ final LeaderInformation newLeaderInformationForContender =
+ leaderInformation
+ .forContenderID(contenderID)
+ .orElse(LeaderInformation.empty());
+ final LeaderInformation confirmedLeaderInformationForContender =
+ confirmedLeaderInformation.forContenderIDOrEmpty(contenderID);
+
+ if (confirmedLeaderInformationForContender.isEmpty()) {
+ LOG.trace(
+ "Leader information changed while there's no confirmation available by the contender for contender ID '{}' for this session. Changed leader information {} will be ignored.",
+ contenderID,
+ LeaderElectionUtils.convertToString(
+ newLeaderInformationForContender));
Review Comment:
> On the flip-side, we don't know if the leader is ready yet to accept requests yet? (because for example the dispatcher is still starting) thinking
Yes, that would only be the case after the `confirmLeadership` method call finally happened that would set the `confirmedLeaderInformation` within the service and in the HA backend.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] XComp commented on pull request #22844: [FLINK-31784][runtime] Adds multi-component support to DefaultLeaderElectionService
Posted by "XComp (via GitHub)" <gi...@apache.org>.
XComp commented on PR #22844:
URL: https://github.com/apache/flink/pull/22844#issuecomment-1620584126
Thanks for the approval. I squashed the commits and rebased to most-recent `master` for a final CI run.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] XComp commented on pull request #22844: [FLINK-31784][runtime] Adds multi-component support to DefaultLeaderElectionService
Posted by "XComp (via GitHub)" <gi...@apache.org>.
XComp commented on PR #22844:
URL: https://github.com/apache/flink/pull/22844#issuecomment-1613481762
Force-pushed a rebase to the most-recent version of FLINK-31783 to include master with the fix for FLINK-32490
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] XComp commented on a diff in pull request #22844: [FLINK-31784][runtime] Adds multi-component support to DefaultLeaderElectionService
Posted by "XComp (via GitHub)" <gi...@apache.org>.
XComp commented on code in PR #22844:
URL: https://github.com/apache/flink/pull/22844#discussion_r1251576723
##########
flink-runtime/src/test/java/org/apache/flink/runtime/leaderelection/DefaultLeaderElectionServiceTest.java:
##########
@@ -597,14 +676,30 @@ void testOnLeaderInformationChangeIsIgnoredAfterLeaderElectionBeingStop() throws
() -> {
grantLeadership();
- assertThat(storedLeaderInformation.get().forContenderID(contenderID))
- .isPresent();
+ applyToBothContenderContexts(
+ ctx ->
+ assertThat(
+ storedLeaderInformation
+ .get()
+ .forContenderID(
+ ctx.contenderID))
+ .isPresent());
+
+ contenderContext0.leaderElection.close();
Review Comment:
Because it checked also the other contender which the test expects to be corrected after the first close call. But I noticed that this isn't the right behavior: We shouldn't touch the LeaderInformation for a contender that wasn't mentioned in the `notifyLeaderInformationChange` call. I have to revisit the behavior of the leaderinformation change event handling
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot commented on pull request #22844: [WIP][FLINK-31784][runtime] Adds multi-component support to DefaultLeaderElectionService
Posted by "flinkbot (via GitHub)" <gi...@apache.org>.
flinkbot commented on PR #22844:
URL: https://github.com/apache/flink/pull/22844#issuecomment-1601072224
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "506a625457f00f4c99d8ccc6a8bef17aeb4a5749",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "506a625457f00f4c99d8ccc6a8bef17aeb4a5749",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 506a625457f00f4c99d8ccc6a8bef17aeb4a5749 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] XComp commented on pull request #22844: [FLINK-31784][runtime] Adds multi-component support to DefaultLeaderElectionService
Posted by "XComp (via GitHub)" <gi...@apache.org>.
XComp commented on PR #22844:
URL: https://github.com/apache/flink/pull/22844#issuecomment-1613331340
@zentol I prepared the PR - feel free to pick it for a review.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] XComp commented on pull request #22844: [FLINK-31784][runtime] Adds multi-component support to DefaultLeaderElectionService
Posted by "XComp (via GitHub)" <gi...@apache.org>.
XComp commented on PR #22844:
URL: https://github.com/apache/flink/pull/22844#issuecomment-1614873661
and another one after I rebased FLINK-31783 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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] XComp commented on a diff in pull request #22844: [FLINK-31784][runtime] Adds multi-component support to DefaultLeaderElectionService
Posted by "XComp (via GitHub)" <gi...@apache.org>.
XComp commented on code in PR #22844:
URL: https://github.com/apache/flink/pull/22844#discussion_r1251577433
##########
flink-runtime/src/test/java/org/apache/flink/runtime/leaderelection/DefaultLeaderElectionServiceTest.java:
##########
@@ -932,38 +1064,71 @@ void runTestWithManuallyTriggeredEvents(
void runTest(RunnableWithException testMethod, ExecutorService leaderEventOperationExecutor)
throws Exception {
- try {
- leaderElectionService =
- new DefaultLeaderElectionService(
- driverFactory,
- DefaultLeaderElectionServiceTest.this.fatalErrorHandlerExtension
- .getTestingFatalErrorHandler(),
- leaderEventOperationExecutor);
+ try (final DefaultLeaderElectionService localLeaderElectionService =
+ new DefaultLeaderElectionService(
+ driverFactory,
+ DefaultLeaderElectionServiceTest.this.fatalErrorHandlerExtension
+ .getTestingFatalErrorHandler(),
+ leaderEventOperationExecutor)) {
+ leaderElectionService = localLeaderElectionService;
leaderElectionService.startLeaderElectionBackend();
testingLeaderElectionDriver = driverFactory.assertAndGetOnlyCreatedDriver();
- leaderElection = leaderElectionService.createLeaderElection(contenderID);
- testingContender = new TestingContender(TEST_URL, leaderElection);
- testingContender.startLeaderElection();
+ try (final ContenderContext localContenderContext0 =
+ ContenderContext.create(0, leaderElectionService);
+ final ContenderContext localContenderContext1 =
+ ContenderContext.create(1, leaderElectionService)) {
+ this.contenderContext0 = localContenderContext0;
+ this.contenderContext1 = localContenderContext1;
Review Comment:
Yes, the purpose was to have the `DefaultLeaderElectionServiceTest` providing functionality for multiple contenders
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] XComp commented on a diff in pull request #22844: [FLINK-31784][runtime] Adds multi-component support to DefaultLeaderElectionService
Posted by "XComp (via GitHub)" <gi...@apache.org>.
XComp commented on code in PR #22844:
URL: https://github.com/apache/flink/pull/22844#discussion_r1251536978
##########
flink-runtime/src/main/java/org/apache/flink/runtime/leaderelection/DefaultLeaderElectionService.java:
##########
@@ -425,59 +425,72 @@ private void onRevokeLeadershipInternal() {
}
@GuardedBy("lock")
- private void notifyLeaderContenderOfLeadershipLoss() {
+ private void notifyLeaderContenderOfLeadershipLoss(
+ String contenderID, LeaderContender leaderContender) {
Preconditions.checkState(
leaderContender != null,
"The LeaderContender should be always set when calling this method.");
- if (confirmedLeaderInformation.isEmpty()) {
+ if (!confirmedLeaderInformation.hasLeaderInformation(contenderID)) {
LOG.debug(
"Revoking leadership to contender {} while a previous leadership grant wasn't confirmed, yet.",
- leaderContender.getDescription());
+ contenderID);
} else {
LOG.debug(
"Revoking leadership to contender {} for {}.",
- leaderContender.getDescription(),
- LeaderElectionUtils.convertToString(confirmedLeaderInformation));
+ contenderID,
+ LeaderElectionUtils.convertToString(
+ confirmedLeaderInformation.forContenderIDOrEmpty(contenderID)));
}
- confirmedLeaderInformation = LeaderInformation.empty();
+ confirmedLeaderInformation =
+ LeaderInformationRegister.clear(confirmedLeaderInformation, contenderID);
leaderContender.revokeLeadership();
}
@Override
public void onLeaderInformationChange(LeaderInformation leaderInformation) {
- runInLeaderEventThread(() -> onLeaderInformationChangeInternal(leaderInformation));
+ leaderContenderRegistry
+ .keySet()
+ .forEach(
+ contenderID ->
+ notifyLeaderInformationChange(contenderID, leaderInformation));
}
@GuardedBy("lock")
- private void onLeaderInformationChangeInternal(LeaderInformation leaderInformation) {
- if (leaderContender != null) {
- LOG.trace(
- "Leader node changed while {} is the leader with {}. New leader information {}.",
- leaderContender.getDescription(),
- LeaderElectionUtils.convertToString(confirmedLeaderInformation),
- LeaderElectionUtils.convertToString(leaderInformation));
- if (!confirmedLeaderInformation.isEmpty()) {
- final LeaderInformation confirmedLeaderInfo = this.confirmedLeaderInformation;
- if (leaderInformation.isEmpty()) {
- LOG.debug(
- "Writing leader information by {} since the external storage is empty.",
- leaderContender.getDescription());
- leaderElectionDriver.publishLeaderInformation(contenderID, confirmedLeaderInfo);
- } else if (!leaderInformation.equals(confirmedLeaderInfo)) {
- // the data field does not correspond to the expected leader information
- LOG.debug(
- "Correcting leader information by {}.",
- leaderContender.getDescription());
- leaderElectionDriver.publishLeaderInformation(contenderID, confirmedLeaderInfo);
- }
- }
- } else {
- LOG.debug(
- "Ignoring change notification since the {} has already been stopped.",
- leaderElectionDriver);
- }
+ private void onLeaderInformationChangeInternal(LeaderInformationRegister leaderInformation) {
+ leaderContenderRegistry.forEach(
+ (contenderID, leaderContender) -> {
+ final LeaderInformation newLeaderInformationForContender =
+ leaderInformation
+ .forContenderID(contenderID)
+ .orElse(LeaderInformation.empty());
+ final LeaderInformation confirmedLeaderInformationForContender =
+ confirmedLeaderInformation.forContenderIDOrEmpty(contenderID);
+
+ if (confirmedLeaderInformationForContender.isEmpty()) {
+ LOG.trace(
+ "Leader information changed while there's no confirmation available by the contender for contender ID '{}' for this session. Changed leader information {} will be ignored.",
+ contenderID,
+ LeaderElectionUtils.convertToString(
+ newLeaderInformationForContender));
Review Comment:
The interface methods `onLeaderInformationChange` (for the `LeaderElectionEventHandler`) and `notifyLeaderInformationChange`/`notifyAllKnownLeaderInformation` (for the `MultipleComponentLeaderElectionDriver.Listener`) are only called if the current service has leadership acquired and informs the service that some other process updated the connection info (w/o having the right to do so). In that case, the leader has to sent its confirmed information to the HA backend to overwrite the data again. That's what's happening here.
I understand the code in a way for your case, that there is no confirmedInformation, yet, but the leadership was granted, i.e. there is a `confirmLeadership` call on the way which, as a consequence, will write the confirmed leadership to the HA backend. I guess, that this is why there was no data written to the HA backend in that case (even in the past).
But thinking about it: We might want to write the empty LeaderInformation to the HA backend in that case to avoid that monitoring services pick up the changed (false) LeaderInformation from the HA backend while the `confirmLeadership` call is delayed. :thinking:
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] XComp commented on pull request #22844: [FLINK-31784][runtime] Adds multi-component support to DefaultLeaderElectionService
Posted by "XComp (via GitHub)" <gi...@apache.org>.
XComp commented on PR #22844:
URL: https://github.com/apache/flink/pull/22844#issuecomment-1612584018
I rebased the PR to most-recent version of PR https://github.com/apache/flink/pull/22661 (FLINK-31783) and added an additional commit with more test coverage
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] XComp merged pull request #22844: [FLINK-31784][runtime] Adds multi-component support to DefaultLeaderElectionService
Posted by "XComp (via GitHub)" <gi...@apache.org>.
XComp merged PR #22844:
URL: https://github.com/apache/flink/pull/22844
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] XComp commented on a diff in pull request #22844: [FLINK-31784][runtime] Adds multi-component support to DefaultLeaderElectionService
Posted by "XComp (via GitHub)" <gi...@apache.org>.
XComp commented on code in PR #22844:
URL: https://github.com/apache/flink/pull/22844#discussion_r1252040025
##########
flink-runtime/src/main/java/org/apache/flink/runtime/leaderelection/DefaultLeaderElectionService.java:
##########
@@ -425,59 +425,72 @@ private void onRevokeLeadershipInternal() {
}
@GuardedBy("lock")
- private void notifyLeaderContenderOfLeadershipLoss() {
+ private void notifyLeaderContenderOfLeadershipLoss(
+ String contenderID, LeaderContender leaderContender) {
Preconditions.checkState(
leaderContender != null,
"The LeaderContender should be always set when calling this method.");
- if (confirmedLeaderInformation.isEmpty()) {
+ if (!confirmedLeaderInformation.hasLeaderInformation(contenderID)) {
LOG.debug(
"Revoking leadership to contender {} while a previous leadership grant wasn't confirmed, yet.",
- leaderContender.getDescription());
+ contenderID);
} else {
LOG.debug(
"Revoking leadership to contender {} for {}.",
- leaderContender.getDescription(),
- LeaderElectionUtils.convertToString(confirmedLeaderInformation));
+ contenderID,
+ LeaderElectionUtils.convertToString(
+ confirmedLeaderInformation.forContenderIDOrEmpty(contenderID)));
}
- confirmedLeaderInformation = LeaderInformation.empty();
+ confirmedLeaderInformation =
+ LeaderInformationRegister.clear(confirmedLeaderInformation, contenderID);
leaderContender.revokeLeadership();
}
@Override
public void onLeaderInformationChange(LeaderInformation leaderInformation) {
- runInLeaderEventThread(() -> onLeaderInformationChangeInternal(leaderInformation));
+ leaderContenderRegistry
+ .keySet()
+ .forEach(
+ contenderID ->
+ notifyLeaderInformationChange(contenderID, leaderInformation));
}
@GuardedBy("lock")
- private void onLeaderInformationChangeInternal(LeaderInformation leaderInformation) {
- if (leaderContender != null) {
- LOG.trace(
- "Leader node changed while {} is the leader with {}. New leader information {}.",
- leaderContender.getDescription(),
- LeaderElectionUtils.convertToString(confirmedLeaderInformation),
- LeaderElectionUtils.convertToString(leaderInformation));
- if (!confirmedLeaderInformation.isEmpty()) {
- final LeaderInformation confirmedLeaderInfo = this.confirmedLeaderInformation;
- if (leaderInformation.isEmpty()) {
- LOG.debug(
- "Writing leader information by {} since the external storage is empty.",
- leaderContender.getDescription());
- leaderElectionDriver.publishLeaderInformation(contenderID, confirmedLeaderInfo);
- } else if (!leaderInformation.equals(confirmedLeaderInfo)) {
- // the data field does not correspond to the expected leader information
- LOG.debug(
- "Correcting leader information by {}.",
- leaderContender.getDescription());
- leaderElectionDriver.publishLeaderInformation(contenderID, confirmedLeaderInfo);
- }
- }
- } else {
- LOG.debug(
- "Ignoring change notification since the {} has already been stopped.",
- leaderElectionDriver);
- }
+ private void onLeaderInformationChangeInternal(LeaderInformationRegister leaderInformation) {
+ leaderContenderRegistry.forEach(
+ (contenderID, leaderContender) -> {
+ final LeaderInformation newLeaderInformationForContender =
+ leaderInformation
+ .forContenderID(contenderID)
+ .orElse(LeaderInformation.empty());
+ final LeaderInformation confirmedLeaderInformationForContender =
+ confirmedLeaderInformation.forContenderIDOrEmpty(contenderID);
+
+ if (confirmedLeaderInformationForContender.isEmpty()) {
+ LOG.trace(
+ "Leader information changed while there's no confirmation available by the contender for contender ID '{}' for this session. Changed leader information {} will be ignored.",
+ contenderID,
+ LeaderElectionUtils.convertToString(
+ newLeaderInformationForContender));
Review Comment:
> On the flip-side, we don't know if the leader is ready yet to accept requests yet? (because for example the dispatcher is still starting) thinking
Yes, that would only be the case after the `confirmLeadership` method call finally happened that would set the `confirmedLeaderInformation` within the service and in the HA backend.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] XComp commented on a diff in pull request #22844: [FLINK-31784][runtime] Adds multi-component support to DefaultLeaderElectionService
Posted by "XComp (via GitHub)" <gi...@apache.org>.
XComp commented on code in PR #22844:
URL: https://github.com/apache/flink/pull/22844#discussion_r1251060414
##########
flink-runtime/src/main/java/org/apache/flink/runtime/leaderelection/DefaultLeaderElectionService.java:
##########
@@ -503,43 +516,53 @@ private void runInLeaderEventThread(Runnable callback) {
private void forwardErrorToLeaderContender(Throwable t) {
synchronized (lock) {
- if (leaderContender == null) {
+ if (leaderContenderRegistry.isEmpty()) {
fallbackErrorHandler.onFatalError(t);
return;
}
- if (t instanceof LeaderElectionException) {
- leaderContender.handleError((LeaderElectionException) t);
- } else {
- leaderContender.handleError(new LeaderElectionException(t));
- }
+ leaderContenderRegistry
+ .values()
+ .forEach(
+ leaderContender -> {
+ if (leaderContender == null) {
Review Comment:
:facepalm:
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] XComp commented on a diff in pull request #22844: [FLINK-31784][runtime] Adds multi-component support to DefaultLeaderElectionService
Posted by "XComp (via GitHub)" <gi...@apache.org>.
XComp commented on code in PR #22844:
URL: https://github.com/apache/flink/pull/22844#discussion_r1252042488
##########
flink-runtime/src/main/java/org/apache/flink/runtime/leaderelection/DefaultLeaderElectionService.java:
##########
@@ -425,59 +425,72 @@ private void onRevokeLeadershipInternal() {
}
@GuardedBy("lock")
- private void notifyLeaderContenderOfLeadershipLoss() {
+ private void notifyLeaderContenderOfLeadershipLoss(
+ String contenderID, LeaderContender leaderContender) {
Preconditions.checkState(
leaderContender != null,
"The LeaderContender should be always set when calling this method.");
- if (confirmedLeaderInformation.isEmpty()) {
+ if (!confirmedLeaderInformation.hasLeaderInformation(contenderID)) {
LOG.debug(
"Revoking leadership to contender {} while a previous leadership grant wasn't confirmed, yet.",
- leaderContender.getDescription());
+ contenderID);
} else {
LOG.debug(
"Revoking leadership to contender {} for {}.",
- leaderContender.getDescription(),
- LeaderElectionUtils.convertToString(confirmedLeaderInformation));
+ contenderID,
+ LeaderElectionUtils.convertToString(
+ confirmedLeaderInformation.forContenderIDOrEmpty(contenderID)));
}
- confirmedLeaderInformation = LeaderInformation.empty();
+ confirmedLeaderInformation =
+ LeaderInformationRegister.clear(confirmedLeaderInformation, contenderID);
leaderContender.revokeLeadership();
}
@Override
public void onLeaderInformationChange(LeaderInformation leaderInformation) {
- runInLeaderEventThread(() -> onLeaderInformationChangeInternal(leaderInformation));
+ leaderContenderRegistry
+ .keySet()
+ .forEach(
+ contenderID ->
+ notifyLeaderInformationChange(contenderID, leaderInformation));
}
@GuardedBy("lock")
- private void onLeaderInformationChangeInternal(LeaderInformation leaderInformation) {
- if (leaderContender != null) {
- LOG.trace(
- "Leader node changed while {} is the leader with {}. New leader information {}.",
- leaderContender.getDescription(),
- LeaderElectionUtils.convertToString(confirmedLeaderInformation),
- LeaderElectionUtils.convertToString(leaderInformation));
- if (!confirmedLeaderInformation.isEmpty()) {
- final LeaderInformation confirmedLeaderInfo = this.confirmedLeaderInformation;
- if (leaderInformation.isEmpty()) {
- LOG.debug(
- "Writing leader information by {} since the external storage is empty.",
- leaderContender.getDescription());
- leaderElectionDriver.publishLeaderInformation(contenderID, confirmedLeaderInfo);
- } else if (!leaderInformation.equals(confirmedLeaderInfo)) {
- // the data field does not correspond to the expected leader information
- LOG.debug(
- "Correcting leader information by {}.",
- leaderContender.getDescription());
- leaderElectionDriver.publishLeaderInformation(contenderID, confirmedLeaderInfo);
- }
- }
- } else {
- LOG.debug(
- "Ignoring change notification since the {} has already been stopped.",
- leaderElectionDriver);
- }
+ private void onLeaderInformationChangeInternal(LeaderInformationRegister leaderInformation) {
+ leaderContenderRegistry.forEach(
+ (contenderID, leaderContender) -> {
+ final LeaderInformation newLeaderInformationForContender =
+ leaderInformation
+ .forContenderID(contenderID)
+ .orElse(LeaderInformation.empty());
+ final LeaderInformation confirmedLeaderInformationForContender =
+ confirmedLeaderInformation.forContenderIDOrEmpty(contenderID);
+
+ if (confirmedLeaderInformationForContender.isEmpty()) {
+ LOG.trace(
+ "Leader information changed while there's no confirmation available by the contender for contender ID '{}' for this session. Changed leader information {} will be ignored.",
+ contenderID,
+ LeaderElectionUtils.convertToString(
+ newLeaderInformationForContender));
Review Comment:
I added the `publishLeaderInformation` call for the case you described.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] XComp commented on pull request #22844: [FLINK-31784][runtime] Adds multi-component support to DefaultLeaderElectionService
Posted by "XComp (via GitHub)" <gi...@apache.org>.
XComp commented on PR #22844:
URL: https://github.com/apache/flink/pull/22844#issuecomment-1617671478
Force-pushed a rebase to `master` after FLINK-31783 was merged.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] XComp commented on pull request #22844: [FLINK-31784][runtime] Adds multi-component support to DefaultLeaderElectionService
Posted by "XComp (via GitHub)" <gi...@apache.org>.
XComp commented on PR #22844:
URL: https://github.com/apache/flink/pull/22844#issuecomment-1613323731
Force-pushed a rebase to the most-recent version of FLINK-31783 to include `master` with the fix for FLINK-32478
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org