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/07/03 13:52:33 UTC
[GitHub] [flink] XComp opened a new pull request, #22935: [FLINK-32381][runtime] Replaces FatalErrorHandler by Listener.onError
XComp opened a new pull request, #22935:
URL: https://github.com/apache/flink/pull/22935
## What is the purpose of the change
The driver has two different ways to communicate with the `DefaultLeaderElectionService` right now:
(1) Through the `Listener` interface
(2) Through the `FatalErrorHandler`
This PR merges the two ways into a single communication channel through the `Listener` interface
## Brief change log
* Removed `FatalErrorHandler` field from driver and its constructor
* Adds `onError` to `Listener` interface
## Verifying this change
* `TestingLeaderElectionListener` implements `onError` and some helper methods to retrieve any outstanding error
* `TestingLeaderElectionListener` is used in tests where `FatalErrorHandler` was used beforehand
## 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 pull request #22935: [FLINK-32381][runtime] Replaces FatalErrorHandler by Listener.onError
Posted by "XComp (via GitHub)" <gi...@apache.org>.
XComp commented on PR #22935:
URL: https://github.com/apache/flink/pull/22935#issuecomment-1623716496
And I had to fix an unused import
--
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 #22935: [FLINK-32381][runtime] Replaces FatalErrorHandler by Listener.onError
Posted by "XComp (via GitHub)" <gi...@apache.org>.
XComp commented on PR #22935:
URL: https://github.com/apache/flink/pull/22935#issuecomment-1623075149
Force-pushed the rebase to most-recent version of FLINK-32419.
--
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 #22935: [FLINK-32381][runtime] Replaces FatalErrorHandler by Listener.onError
Posted by "flinkbot (via GitHub)" <gi...@apache.org>.
flinkbot commented on PR #22935:
URL: https://github.com/apache/flink/pull/22935#issuecomment-1618358680
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "96a4e0bc7a8bb696337c84845061c408377715c9",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "96a4e0bc7a8bb696337c84845061c408377715c9",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 96a4e0bc7a8bb696337c84845061c408377715c9 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 a diff in pull request #22935: [FLINK-32381][runtime] Replaces FatalErrorHandler by Listener.onError
Posted by "XComp (via GitHub)" <gi...@apache.org>.
XComp commented on code in PR #22935:
URL: https://github.com/apache/flink/pull/22935#discussion_r1254407986
##########
flink-runtime/src/test/java/org/apache/flink/runtime/leaderelection/ZooKeeperLeaderElectionTest.java:
##########
@@ -409,20 +408,19 @@ void testExceptionForwarding() throws Exception {
final TestingLeaderElectionListener electionEventHandler =
new TestingLeaderElectionListener();
- final TestingFatalErrorHandler fatalErrorHandler =
- testingFatalErrorHandlerResource.getTestingFatalErrorHandler();
try (ZooKeeperLeaderElectionDriver leaderElectionDriver =
- createAndInitLeaderElectionDriver(
- client, electionEventHandler, fatalErrorHandler)) {
+ createAndInitLeaderElectionDriver(client, electionEventHandler)) {
electionEventHandler.await(LeaderElectionEvent.IsLeaderEvent.class);
leaderElectionDriver.publishLeaderInformation(
CONTENDER_ID, LeaderInformation.known(UUID.randomUUID(), "some-address"));
- assertThat(fatalErrorHandler.getErrorFuture()).isCompletedWithValue(testException);
+ final LeaderElectionEvent.ErrorEvent errorEvent =
+ electionEventHandler.await(LeaderElectionEvent.ErrorEvent.class);
+ assertThat(errorEvent.getError()).isEqualTo(testException);
} finally {
- fatalErrorHandler.clearError();
+ electionEventHandler.throwIfErrorEventHappened();
Review Comment:
I pushed the changes related to your recommendation.
--
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 #22935: [FLINK-32381][runtime] Replaces FatalErrorHandler by Listener.onError
Posted by "XComp (via GitHub)" <gi...@apache.org>.
XComp merged PR #22935:
URL: https://github.com/apache/flink/pull/22935
--
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 #22935: [FLINK-32381][runtime] Replaces FatalErrorHandler by Listener.onError
Posted by "XComp (via GitHub)" <gi...@apache.org>.
XComp commented on PR #22935:
URL: https://github.com/apache/flink/pull/22935#issuecomment-1623595626
Force-pushed the rebase to most-recent version of [FLINK-32419](https://issues.apache.org/jira/browse/FLINK-32419).
--
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 #22935: [FLINK-32381][runtime] Replaces FatalErrorHandler by Listener.onError
Posted by "zentol (via GitHub)" <gi...@apache.org>.
zentol commented on code in PR #22935:
URL: https://github.com/apache/flink/pull/22935#discussion_r1254190529
##########
flink-runtime/src/test/java/org/apache/flink/runtime/leaderelection/ZooKeeperLeaderElectionTest.java:
##########
@@ -409,20 +408,19 @@ void testExceptionForwarding() throws Exception {
final TestingLeaderElectionListener electionEventHandler =
new TestingLeaderElectionListener();
- final TestingFatalErrorHandler fatalErrorHandler =
- testingFatalErrorHandlerResource.getTestingFatalErrorHandler();
try (ZooKeeperLeaderElectionDriver leaderElectionDriver =
- createAndInitLeaderElectionDriver(
- client, electionEventHandler, fatalErrorHandler)) {
+ createAndInitLeaderElectionDriver(client, electionEventHandler)) {
electionEventHandler.await(LeaderElectionEvent.IsLeaderEvent.class);
leaderElectionDriver.publishLeaderInformation(
CONTENDER_ID, LeaderInformation.known(UUID.randomUUID(), "some-address"));
- assertThat(fatalErrorHandler.getErrorFuture()).isCompletedWithValue(testException);
+ final LeaderElectionEvent.ErrorEvent errorEvent =
+ electionEventHandler.await(LeaderElectionEvent.ErrorEvent.class);
+ assertThat(errorEvent.getError()).isEqualTo(testException);
} finally {
- fatalErrorHandler.clearError();
+ electionEventHandler.throwIfErrorEventHappened();
Review Comment:
Or even better, an unchecked exception?
--
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 #22935: [FLINK-32381][runtime] Replaces FatalErrorHandler by Listener.onError
Posted by "zentol (via GitHub)" <gi...@apache.org>.
zentol commented on code in PR #22935:
URL: https://github.com/apache/flink/pull/22935#discussion_r1254766401
##########
flink-runtime/src/test/java/org/apache/flink/runtime/leaderelection/TestingLeaderElectionListener.java:
##########
@@ -61,6 +66,19 @@ private void put(LeaderElectionEvent leaderElectionEvent) {
}
}
+ public <T> T assertNextEvent(Class<T> expectedEventClass) throws InterruptedException {
+ final LeaderElectionEvent leaderElectionEvent = leaderElectionEvents.take();
+
+ if (expectedEventClass.isAssignableFrom(leaderElectionEvent.getClass())) {
+ return expectedEventClass.cast(leaderElectionEvent);
+ }
+
+ throw new AssertionError(
+ String.format(
+ "The next event didn't match the expected event type %s.",
+ expectedEventClass.getSimpleName()));
Review Comment:
```
assertThat(leaderElectionEvent.getClass).isInstanceOf(expectedEventClass);
return <cast>
```
--
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 #22935: [FLINK-32381][runtime] Replaces FatalErrorHandler by Listener.onError
Posted by "XComp (via GitHub)" <gi...@apache.org>.
XComp commented on code in PR #22935:
URL: https://github.com/apache/flink/pull/22935#discussion_r1254003907
##########
flink-runtime/src/test/java/org/apache/flink/runtime/leaderelection/ZooKeeperLeaderElectionTest.java:
##########
@@ -409,20 +408,19 @@ void testExceptionForwarding() throws Exception {
final TestingLeaderElectionListener electionEventHandler =
new TestingLeaderElectionListener();
- final TestingFatalErrorHandler fatalErrorHandler =
- testingFatalErrorHandlerResource.getTestingFatalErrorHandler();
try (ZooKeeperLeaderElectionDriver leaderElectionDriver =
- createAndInitLeaderElectionDriver(
- client, electionEventHandler, fatalErrorHandler)) {
+ createAndInitLeaderElectionDriver(client, electionEventHandler)) {
electionEventHandler.await(LeaderElectionEvent.IsLeaderEvent.class);
leaderElectionDriver.publishLeaderInformation(
CONTENDER_ID, LeaderInformation.known(UUID.randomUUID(), "some-address"));
- assertThat(fatalErrorHandler.getErrorFuture()).isCompletedWithValue(testException);
+ final LeaderElectionEvent.ErrorEvent errorEvent =
+ electionEventHandler.await(LeaderElectionEvent.ErrorEvent.class);
+ assertThat(errorEvent.getError()).isEqualTo(testException);
} finally {
- fatalErrorHandler.clearError();
+ electionEventHandler.throwIfErrorEventHappened();
Review Comment:
This change will cause all the test methods to change there signature from throwing `Throwable` instead of `Exception`. But that's what we want because we also would want to forward Errors, not only Exceptions.
--
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 #22935: [FLINK-32381][runtime] Replaces FatalErrorHandler by Listener.onError
Posted by "XComp (via GitHub)" <gi...@apache.org>.
XComp commented on PR #22935:
URL: https://github.com/apache/flink/pull/22935#issuecomment-1624163237
Thanks for the approval. I force-pushed a rebase to `master` (with FLINK-32419 being included in `master`) for a final CI run. I addressed your comment accordingly.
--
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 #22935: [FLINK-32381][runtime] Replaces FatalErrorHandler by Listener.onError
Posted by "XComp (via GitHub)" <gi...@apache.org>.
XComp commented on code in PR #22935:
URL: https://github.com/apache/flink/pull/22935#discussion_r1254358231
##########
flink-runtime/src/test/java/org/apache/flink/runtime/leaderelection/ZooKeeperLeaderElectionTest.java:
##########
@@ -409,20 +408,19 @@ void testExceptionForwarding() throws Exception {
final TestingLeaderElectionListener electionEventHandler =
new TestingLeaderElectionListener();
- final TestingFatalErrorHandler fatalErrorHandler =
- testingFatalErrorHandlerResource.getTestingFatalErrorHandler();
try (ZooKeeperLeaderElectionDriver leaderElectionDriver =
- createAndInitLeaderElectionDriver(
- client, electionEventHandler, fatalErrorHandler)) {
+ createAndInitLeaderElectionDriver(client, electionEventHandler)) {
electionEventHandler.await(LeaderElectionEvent.IsLeaderEvent.class);
leaderElectionDriver.publishLeaderInformation(
CONTENDER_ID, LeaderInformation.known(UUID.randomUUID(), "some-address"));
- assertThat(fatalErrorHandler.getErrorFuture()).isCompletedWithValue(testException);
+ final LeaderElectionEvent.ErrorEvent errorEvent =
+ electionEventHandler.await(LeaderElectionEvent.ErrorEvent.class);
+ assertThat(errorEvent.getError()).isEqualTo(testException);
} finally {
- fatalErrorHandler.clearError();
+ electionEventHandler.throwIfErrorEventHappened();
Review Comment:
that's a good point :+1:
--
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 #22935: [FLINK-32381][runtime] Replaces FatalErrorHandler by Listener.onError
Posted by "XComp (via GitHub)" <gi...@apache.org>.
XComp commented on PR #22935:
URL: https://github.com/apache/flink/pull/22935#issuecomment-1623665936
Force-pushed rebase to fix the checkstyle issue that was introduced in the previous version of FLINK-32419
--
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 #22935: [FLINK-32381][runtime] Replaces FatalErrorHandler by Listener.onError
Posted by "zentol (via GitHub)" <gi...@apache.org>.
zentol commented on code in PR #22935:
URL: https://github.com/apache/flink/pull/22935#discussion_r1254189035
##########
flink-runtime/src/test/java/org/apache/flink/runtime/leaderelection/ZooKeeperLeaderElectionTest.java:
##########
@@ -409,20 +408,19 @@ void testExceptionForwarding() throws Exception {
final TestingLeaderElectionListener electionEventHandler =
new TestingLeaderElectionListener();
- final TestingFatalErrorHandler fatalErrorHandler =
- testingFatalErrorHandlerResource.getTestingFatalErrorHandler();
try (ZooKeeperLeaderElectionDriver leaderElectionDriver =
- createAndInitLeaderElectionDriver(
- client, electionEventHandler, fatalErrorHandler)) {
+ createAndInitLeaderElectionDriver(client, electionEventHandler)) {
electionEventHandler.await(LeaderElectionEvent.IsLeaderEvent.class);
leaderElectionDriver.publishLeaderInformation(
CONTENDER_ID, LeaderInformation.known(UUID.randomUUID(), "some-address"));
- assertThat(fatalErrorHandler.getErrorFuture()).isCompletedWithValue(testException);
+ final LeaderElectionEvent.ErrorEvent errorEvent =
+ electionEventHandler.await(LeaderElectionEvent.ErrorEvent.class);
+ assertThat(errorEvent.getError()).isEqualTo(testException);
} finally {
- fatalErrorHandler.clearError();
+ electionEventHandler.throwIfErrorEventHappened();
Review Comment:
> But that's what we want because we also would want to forward Errors, not only Exceptions.
Could we just wrap everything into an exception? :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