You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "RocMarshal (via GitHub)" <gi...@apache.org> on 2023/02/22 13:49:20 UTC
[GitHub] [flink] RocMarshal opened a new pull request, #21999: [FLINK-29816][streaming] Fix SteamTask skip exception in userfunction of ProcessWindowFunction during Steam.restore
RocMarshal opened a new pull request, #21999:
URL: https://github.com/apache/flink/pull/21999
<!--
*Thank you very much for contributing to Apache Flink - we are happy that you want to help us improve Flink. To help the community review your contribution in the best possible way, please go through the checklist below, which will get the contribution into a shape in which it can be best reviewed.*
*Please understand that we do not do this to make contributions to Flink a hassle. In order to uphold a high standard of quality for code contributions, while at the same time managing a large number of contributions, we need contributors to prepare the contributions well, and give reviewers enough contextual information for the review. Please also understand that contributions that do not follow this guide will take longer to review and thus typically be picked up with lower priority by the community.*
## Contribution Checklist
- Make sure that the pull request corresponds to a [JIRA issue](https://issues.apache.org/jira/projects/FLINK/issues). Exceptions are made for typos in JavaDoc or documentation files, which need no JIRA issue.
- Name the pull request in the form "[FLINK-XXXX] [component] Title of the pull request", where *FLINK-XXXX* should be replaced by the actual issue number. Skip *component* if you are unsure about which is the best component.
Typo fixes that have no associated JIRA issue should be named following this pattern: `[hotfix] [docs] Fix typo in event time introduction` or `[hotfix] [javadocs] Expand JavaDoc for PuncuatedWatermarkGenerator`.
- Fill out the template below to describe the changes contributed by the pull request. That will give reviewers the context they need to do the review.
- Make sure that the change passes the automated tests, i.e., `mvn clean verify` passes. You can set up Azure Pipelines CI to do that following [this guide](https://cwiki.apache.org/confluence/display/FLINK/Azure+Pipelines#AzurePipelines-Tutorial:SettingupAzurePipelinesforaforkoftheFlinkrepository).
- Each pull request should address only one issue, not mix up code from multiple issues.
- Each commit in the pull request has a meaningful commit message (including the JIRA id)
- Once all items of the checklist are addressed, remove the above text and this checklist, leaving only the filled out template below.
**(The sections below can be removed for hotfixes of typos)**
-->
## What is the purpose of the change
*(For example: This pull request makes task deployment go through the blob server, rather than through RPC. That way we avoid re-transferring them on each deployment (during recovery).)*
## Brief change log
Fix SteamTask skip exception in userfunction of ProcessWindowFunction during Steam.restore
## Verifying this change
Please make sure both new and modified tests in this PR follows the conventions defined in our code quality guide: https://flink.apache.org/contributing/code-style-and-quality-common.html#testing
*(Please pick either of the following options)*
This change added tests and can be verified as follows:
- Manually verified locally
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): (yes / **no**)
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: (yes / **no**)
- The serializers: (yes / **no** / don't know)
- The runtime per-record code paths (performance sensitive): (yes / **no** / don't know)
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: (yes / **no** / don't know)
- The S3 file system connector: (yes / **no** / don't know)
## Documentation
- Does this pull request introduce a new feature? (yes / **no**)
- If yes, how is the feature documented? (**not applicable** / docs / JavaDocs / not documented)
--
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 #21999: [FLINK-29816][streaming] Fix SteamTask skip exception in userfunction of ProcessWindowFunction during Steam.restore
Posted by "flinkbot (via GitHub)" <gi...@apache.org>.
flinkbot commented on PR #21999:
URL: https://github.com/apache/flink/pull/21999#issuecomment-1440054899
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "7f36733ac4e13768107d7772c8ebdb541f62ef4a",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "7f36733ac4e13768107d7772c8ebdb541f62ef4a",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 7f36733ac4e13768107d7772c8ebdb541f62ef4a 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] RocMarshal commented on a diff in pull request #21999: [FLINK-29816][streaming] Fix the bug that StreamTask doesn't handle exception during restoring
Posted by "RocMarshal (via GitHub)" <gi...@apache.org>.
RocMarshal commented on code in PR #21999:
URL: https://github.com/apache/flink/pull/21999#discussion_r1116679594
##########
flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTest.java:
##########
@@ -368,17 +368,19 @@ void testCleanUpExceptionSuppressing() throws Exception {
.setupOutputForSingletonOperatorChain(new FailingTwiceOperator())
.build()) {
- assertThatThrownBy(
- () -> {
- testHarness.processElement(new StreamRecord<>("Doesn't matter", 0));
- throw new RuntimeException(
- "Expected an exception but ran successfully");
- })
- .isInstanceOf(ExpectedTestException.class);
+ try {
+ testHarness.processElement(new StreamRecord<>("Doesn't matter", 0));
+ throw new RuntimeException("Expected an exception but ran successfully");
+ } catch (Exception ex) {
+ ExceptionUtils.assertThrowable(ex, ExpectedTestException.class);
+ }
Review Comment:
nice.
thx so much for the quick comments.
--
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] reswqa commented on pull request #21999: [FLINK-29816][streaming] Fix the bug that StreamTask doesn't handle exception during restoring
Posted by "reswqa (via GitHub)" <gi...@apache.org>.
reswqa commented on PR #21999:
URL: https://github.com/apache/flink/pull/21999#issuecomment-1441903864
If you like, you'd better migrate all the tests in `StreamTaskTest` to `Junit5` and `AssertJ` in a hotfix commit. No matter what, the newly introduced test should no longer use the `hamcrest` or `Junit` assertion according to our [Code Style and Quality Guide](https://flink.apache.org/how-to-contribute/code-style-and-quality-common/#tooling).
--
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] RocMarshal commented on a diff in pull request #21999: [FLINK-29816][streaming] Fix the bug that StreamTask doesn't handle exception during restoring
Posted by "RocMarshal (via GitHub)" <gi...@apache.org>.
RocMarshal commented on code in PR #21999:
URL: https://github.com/apache/flink/pull/21999#discussion_r1115840729
##########
flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTest.java:
##########
@@ -410,6 +410,42 @@ public CloseException() {
}
}
+ @Test
+ public void testAsyncExceptionCanBeHandledDuringRestoring() throws Exception {
+ MockEnvironment mockEnvironment = MockEnvironment.builder().build();
+ RuntimeException expectedException = new RuntimeException("RUNTIME EXCEPTION");
+
+ mockEnvironment.setExpectedExternalFailureCause(AsynchronousException.class);
+ final String expectedErrorMessage = "EXPECTED_ERROR MESSAGE";
+
+ StreamTaskITCase.NoOpStreamTask noOpStreamTask =
Review Comment:
`initThrowExceptionTask` is appropriate. a good proposal.👍
--
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] RocMarshal commented on a diff in pull request #21999: [FLINK-29816][streaming] Fix the bug that StreamTask doesn't handle exception during restoring
Posted by "RocMarshal (via GitHub)" <gi...@apache.org>.
RocMarshal commented on code in PR #21999:
URL: https://github.com/apache/flink/pull/21999#discussion_r1117840988
##########
flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTest.java:
##########
@@ -362,30 +358,35 @@ private void testSyncSavepointWithEndInput(
"savepointResult");
harness.processAll();
- Assert.assertEquals(expectEndInput, TestBoundedOneInputStreamOperator.isInputEnded());
+ assertThat(TestBoundedOneInputStreamOperator.isInputEnded()).isEqualTo(expectEndInput);
}
@Test
- public void testCleanUpExceptionSuppressing() throws Exception {
+ void testCleanUpExceptionSuppressing() throws Exception {
try (StreamTaskMailboxTestHarness<String> testHarness =
new StreamTaskMailboxTestHarnessBuilder<>(OneInputStreamTask::new, STRING_TYPE_INFO)
.addInput(STRING_TYPE_INFO)
.setupOutputForSingletonOperatorChain(new FailingTwiceOperator())
.build()) {
- try {
- testHarness.processElement(new StreamRecord<>("Doesn't matter", 0));
- throw new RuntimeException("Expected an exception but ran successfully");
- } catch (Exception ex) {
- ExceptionUtils.assertThrowable(ex, ExpectedTestException.class);
- }
+ assertThatThrownBy(
+ () ->
+ testHarness.processElement(
+ new StreamRecord<>("Doesn't matter", 0)))
+ .satisfies(
+ (Consumer<Throwable>)
+ throwable ->
+ ExceptionUtils.assertThrowable(
+ throwable, ExpectedTestException.class));
- try {
- testHarness.finishProcessing();
- } catch (Exception ex) {
- // todo: checking for suppression if there are more exceptions during cleanup
- ExceptionUtils.assertThrowable(ex, FailingTwiceOperator.CloseException.class);
- }
+ // todo: checking for suppression if there are more exceptions during cleanup
+ assertThatThrownBy(testHarness::finishProcessing)
+ .satisfies(
+ (ThrowingConsumer<Throwable>)
+ throwable ->
+ ExceptionUtils.assertThrowable(
+ throwable,
+ FailingTwiceOperator.CloseException.class));
Review Comment:
Yes, the line is also my original migration line.
@1996fanrui
@reswqa described the reason why we still need the `ExceptionUtils.assertThrowable` at https://github.com/apache/flink/pull/21999#discussion_r1116493534 due to the depth of the 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] reswqa commented on a diff in pull request #21999: [FLINK-29816][streaming] Fix SteamTask skip exception in userfunction of ProcessWindowFunction during Steam.restore
Posted by "reswqa (via GitHub)" <gi...@apache.org>.
reswqa commented on code in PR #21999:
URL: https://github.com/apache/flink/pull/21999#discussion_r1114467955
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java:
##########
@@ -689,6 +695,7 @@ void restoreInternal() throws Exception {
LOG.debug("Re-restore attempt rejected.");
return;
}
+ isInitializing = true;
Review Comment:
+1, It looks a little confused.
--
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] RocMarshal commented on a diff in pull request #21999: [FLINK-29816][streaming] Fix the bug that SteamTask skip handle exception during restoring.
Posted by "RocMarshal (via GitHub)" <gi...@apache.org>.
RocMarshal commented on code in PR #21999:
URL: https://github.com/apache/flink/pull/21999#discussion_r1115242031
##########
flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskITCase.java:
##########
@@ -110,6 +118,42 @@ public void testFailInEndOfConstructor() throws Exception {
.build(EXECUTOR_RESOURCE.getExecutor()));
}
+ @Test
+ public void streamTaskAsyncExceptionHandlerHandleExceptionInInitializingDuration()
+ throws Exception {
+ MockEnvironment mockEnvironment = MockEnvironment.builder().build();
+ RuntimeException expectedException = new RuntimeException("RUNTIME EXCEPTION");
+
+ mockEnvironment.setExpectedExternalFailureCause(AsynchronousException.class);
+ final String expectedErrorMessage = "EXPECTED_ERROR MESSAGE";
+
+ NoOpStreamTask noOpStreamTask =
+ new NoOpStreamTask(mockEnvironment) {
+ @Override
+ protected void init() throws Exception {
+ super.init();
+ CompletableFuture.runAsync(
+ () -> {
+ assertFalse(this.isRunning());
+ assertTrue(this.isInitializing());
+ this.handleAsyncException(
+ expectedErrorMessage, expectedException);
+ });
+ }
Review Comment:
Should the make thread of NoOpStreamTask sleep a small duration
Or add a flag to ensure that The `runAsync` operation must run in the span (isRestoring = true, isRunning=false) ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] 1996fanrui commented on a diff in pull request #21999: [FLINK-29816][streaming] Fix the bug that StreamTask doesn't handle exception during restoring
Posted by "1996fanrui (via GitHub)" <gi...@apache.org>.
1996fanrui commented on code in PR #21999:
URL: https://github.com/apache/flink/pull/21999#discussion_r1115433852
##########
flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskITCase.java:
##########
@@ -110,6 +118,47 @@ public void testFailInEndOfConstructor() throws Exception {
.build(EXECUTOR_RESOURCE.getExecutor()));
}
+ @Test
+ public void streamTaskAsyncExceptionHandlerHandleExceptionInInitializingDuration()
+ throws Exception {
+ MockEnvironment mockEnvironment = MockEnvironment.builder().build();
+ RuntimeException expectedException = new RuntimeException("RUNTIME EXCEPTION");
+
+ mockEnvironment.setExpectedExternalFailureCause(AsynchronousException.class);
+ final String expectedErrorMessage = "EXPECTED_ERROR MESSAGE";
+
+ NoOpStreamTask noOpStreamTask =
+ new NoOpStreamTask(mockEnvironment) {
+ volatile boolean handledAsyncException = false;
+
+ @Override
+ protected void init() throws Exception {
+ super.init();
+
+ CompletableFuture.runAsync(
Review Comment:
Add a comment : `Handle async exception during restoring.`
--
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] 1996fanrui commented on a diff in pull request #21999: [FLINK-29816][streaming] Fix the bug that StreamTask doesn't handle exception during restoring
Posted by "1996fanrui (via GitHub)" <gi...@apache.org>.
1996fanrui commented on code in PR #21999:
URL: https://github.com/apache/flink/pull/21999#discussion_r1116676401
##########
flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTest.java:
##########
@@ -1103,35 +1143,41 @@ private void testFailToConfirmCheckpointMessage(Consumer<StreamTask<?, ?>> consu
* finished.
*/
@Test
- public void testCheckpointFailueOnClosedOperator() throws Throwable {
+ void testCheckpointFailueOnClosedOperator() throws Exception {
ClosingOperator<Integer> operator = new ClosingOperator<>();
StreamTaskMailboxTestHarnessBuilder<Integer> builder =
new StreamTaskMailboxTestHarnessBuilder<>(
OneInputStreamTask::new, BasicTypeInfo.INT_TYPE_INFO)
.addInput(BasicTypeInfo.INT_TYPE_INFO);
try (StreamTaskMailboxTestHarness<Integer> harness =
builder.setupOutputForSingletonOperatorChain(operator).build()) {
- // keeps the mailbox from suspending
- harness.setAutoProcess(false);
- harness.processElement(new StreamRecord<>(1));
-
- harness.streamTask.operatorChain.finishOperators(
- harness.streamTask.getActionExecutor(), StopMode.DRAIN);
- harness.streamTask.operatorChain.closeAllOperators();
- assertTrue(ClosingOperator.closed.get());
-
- harness.streamTask.triggerCheckpointOnBarrier(
- new CheckpointMetaData(1, 0),
- CheckpointOptions.forCheckpointWithDefaultLocation(),
- new CheckpointMetricsBuilder());
- } catch (Exception ex) {
- ExceptionUtils.assertThrowableWithMessage(
- ex, "OperatorChain and Task should never be closed at this point");
+ assertThatThrownBy(
+ () -> {
+ // keeps the mailbox from suspending
+ harness.setAutoProcess(false);
+ harness.processElement(new StreamRecord<>(1));
+
+ harness.streamTask.operatorChain.finishOperators(
+ harness.streamTask.getActionExecutor(), StopMode.DRAIN);
+ harness.streamTask.operatorChain.closeAllOperators();
+ assertThat(ClosingOperator.closed.get()).isTrue();
+
+ harness.streamTask.triggerCheckpointOnBarrier(
+ new CheckpointMetaData(1, 0),
+ CheckpointOptions.forCheckpointWithDefaultLocation(),
+ new CheckpointMetricsBuilder());
+ })
+ .satisfies(
+ (Consumer<Throwable>)
+ throwable ->
+ ExceptionUtils.assertThrowableWithMessage(
+ throwable,
+ "OperatorChain and Task should never be closed at this point"));
Review Comment:
```suggestion
.hasMessage("OperatorChain and Task should never be closed at this point");
```
I'm not sure, please correct me if it doesn't work.
##########
flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTest.java:
##########
@@ -1182,13 +1228,13 @@ public void testThreadInvariants() throws Throwable {
});
runningTask.invocationFuture.get();
- assertThat(
- runningTask.streamTask.getTaskClassLoader(), is(sameInstance(taskClassLoader)));
+ assertThat(runningTask.streamTask.getTaskClassLoader()).isSameAs(taskClassLoader);
+ assertThat(runningTask.streamTask.getTaskClassLoader()).isSameAs(taskClassLoader);
Review Comment:
```suggestion
assertThat(runningTask.streamTask.getTaskClassLoader()).isSameAs(taskClassLoader);
```
Is it repeated?
##########
flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTest.java:
##########
@@ -1103,35 +1143,41 @@ private void testFailToConfirmCheckpointMessage(Consumer<StreamTask<?, ?>> consu
* finished.
*/
@Test
- public void testCheckpointFailueOnClosedOperator() throws Throwable {
+ void testCheckpointFailueOnClosedOperator() throws Exception {
ClosingOperator<Integer> operator = new ClosingOperator<>();
StreamTaskMailboxTestHarnessBuilder<Integer> builder =
new StreamTaskMailboxTestHarnessBuilder<>(
OneInputStreamTask::new, BasicTypeInfo.INT_TYPE_INFO)
.addInput(BasicTypeInfo.INT_TYPE_INFO);
try (StreamTaskMailboxTestHarness<Integer> harness =
builder.setupOutputForSingletonOperatorChain(operator).build()) {
- // keeps the mailbox from suspending
- harness.setAutoProcess(false);
- harness.processElement(new StreamRecord<>(1));
-
- harness.streamTask.operatorChain.finishOperators(
- harness.streamTask.getActionExecutor(), StopMode.DRAIN);
- harness.streamTask.operatorChain.closeAllOperators();
- assertTrue(ClosingOperator.closed.get());
-
- harness.streamTask.triggerCheckpointOnBarrier(
- new CheckpointMetaData(1, 0),
- CheckpointOptions.forCheckpointWithDefaultLocation(),
- new CheckpointMetricsBuilder());
- } catch (Exception ex) {
- ExceptionUtils.assertThrowableWithMessage(
- ex, "OperatorChain and Task should never be closed at this point");
+ assertThatThrownBy(
+ () -> {
+ // keeps the mailbox from suspending
+ harness.setAutoProcess(false);
+ harness.processElement(new StreamRecord<>(1));
+
+ harness.streamTask.operatorChain.finishOperators(
+ harness.streamTask.getActionExecutor(), StopMode.DRAIN);
+ harness.streamTask.operatorChain.closeAllOperators();
+ assertThat(ClosingOperator.closed.get()).isTrue();
Review Comment:
Could these lines be moved outside of `assertThatThrownBy`?
##########
flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTest.java:
##########
@@ -1562,12 +1604,12 @@ public void testSkipRepeatCheckpointComplete() throws Exception {
CheckpointCompleteRecordOperator operator =
(CheckpointCompleteRecordOperator)
(AbstractStreamOperator<?>) testHarness.streamTask.getMainOperator();
- assertEquals(Arrays.asList(3L, 8L), operator.getNotifiedCheckpoint());
+ assertThat(Arrays.asList(3L, 8L)).isEqualTo(operator.getNotifiedCheckpoint());
Review Comment:
```suggestion
assertThat(operator.getNotifiedCheckpoint()).isEqualTo(Arrays.asList(3L, 8L));
```
##########
flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTest.java:
##########
@@ -945,12 +982,15 @@ public void testAsyncCheckpointingConcurrentCloseBeforeAcknowledge() throws Exce
// make sure that all state handles have been discarded
discardFuture.get();
- try {
- mockEnvironment.getAcknowledgeCheckpointFuture().get(10L, TimeUnit.MILLISECONDS);
- fail("The checkpoint should not get acknowledged.");
- } catch (TimeoutException expected) {
- // future should not be completed
- }
+ assertThatThrownBy(
+ () -> {
+ // future should not be completed
+ mockEnvironment
+ .getAcknowledgeCheckpointFuture()
+ .get(10L, TimeUnit.MILLISECONDS);
+ fail("The checkpoint should not get acknowledged.");
Review Comment:
```suggestion
```
Can this line be removed? I remember if it doesn't throw expected exception, `assertThatThrownBy` will fail, right?
##########
flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTest.java:
##########
@@ -1587,12 +1629,12 @@ public void testIgnoreCompleteCheckpointBeforeStartup() throws Exception {
CheckpointCompleteRecordOperator operator =
(CheckpointCompleteRecordOperator)
(AbstractStreamOperator<?>) testHarness.streamTask.getMainOperator();
- assertEquals(Arrays.asList(5L, 6L), operator.getNotifiedCheckpoint());
+ assertThat(Arrays.asList(5L, 6L)).isEqualTo(operator.getNotifiedCheckpoint());
Review Comment:
```suggestion
assertThat(operator.getNotifiedCheckpoint()).isEqualTo(Arrays.asList(5L, 6L));
```
--
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] 1996fanrui commented on a diff in pull request #21999: [FLINK-29816][streaming] Fix the bug that StreamTask doesn't handle exception during restoring
Posted by "1996fanrui (via GitHub)" <gi...@apache.org>.
1996fanrui commented on code in PR #21999:
URL: https://github.com/apache/flink/pull/21999#discussion_r1117858061
##########
flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTest.java:
##########
@@ -362,30 +358,35 @@ private void testSyncSavepointWithEndInput(
"savepointResult");
harness.processAll();
- Assert.assertEquals(expectEndInput, TestBoundedOneInputStreamOperator.isInputEnded());
+ assertThat(TestBoundedOneInputStreamOperator.isInputEnded()).isEqualTo(expectEndInput);
}
@Test
- public void testCleanUpExceptionSuppressing() throws Exception {
+ void testCleanUpExceptionSuppressing() throws Exception {
try (StreamTaskMailboxTestHarness<String> testHarness =
new StreamTaskMailboxTestHarnessBuilder<>(OneInputStreamTask::new, STRING_TYPE_INFO)
.addInput(STRING_TYPE_INFO)
.setupOutputForSingletonOperatorChain(new FailingTwiceOperator())
.build()) {
- try {
- testHarness.processElement(new StreamRecord<>("Doesn't matter", 0));
- throw new RuntimeException("Expected an exception but ran successfully");
- } catch (Exception ex) {
- ExceptionUtils.assertThrowable(ex, ExpectedTestException.class);
- }
+ assertThatThrownBy(
+ () ->
+ testHarness.processElement(
+ new StreamRecord<>("Doesn't matter", 0)))
+ .satisfies(
+ (Consumer<Throwable>)
+ throwable ->
+ ExceptionUtils.assertThrowable(
+ throwable, ExpectedTestException.class));
- try {
- testHarness.finishProcessing();
- } catch (Exception ex) {
- // todo: checking for suppression if there are more exceptions during cleanup
- ExceptionUtils.assertThrowable(ex, FailingTwiceOperator.CloseException.class);
- }
+ // todo: checking for suppression if there are more exceptions during cleanup
+ assertThatThrownBy(testHarness::finishProcessing)
+ .satisfies(
+ (ThrowingConsumer<Throwable>)
+ throwable ->
+ ExceptionUtils.assertThrowable(
+ throwable,
+ FailingTwiceOperator.CloseException.class));
Review Comment:
```suggestion
.satisfies(anyCauseMatches(FailingTwiceOperator.CloseException.class));
```
Hi @RocMarshal , thanks for your clarification, I got it.
Hi @reswqa , I found `FlinkAssertions.anyCauseMatches` is similar with `ExceptionUtils.assertThrowable`, and it can find the throwable from more deeper level, it's simpler than `ExceptionUtils.assertThrowable`.
What do you think? And please take a look these comments[1][2], they are similar with this comment.
[1] https://github.com/apache/flink/pull/21999#discussion_r1117222444
[2] https://github.com/apache/flink/pull/21999#discussion_r1117225403
--
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] reswqa commented on a diff in pull request #21999: [FLINK-29816][streaming] Fix SteamTask skip exception in userfunction of ProcessWindowFunction during Steam.restore
Posted by "reswqa (via GitHub)" <gi...@apache.org>.
reswqa commented on code in PR #21999:
URL: https://github.com/apache/flink/pull/21999#discussion_r1114477282
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java:
##########
@@ -261,6 +261,12 @@
*/
private volatile boolean isRunning;
+ /**
+ * Flag to mark the task "in operation", in which case check needs to be process the exceptions
+ * thrown in the initializing duration.
+ */
+ private volatile boolean isInitializing;
Review Comment:
Ah, I guess he just want to follow the `ExecutionState`'s initializing state, the corresponding java doc indicates that:
`Restoring last possible valid state of the task if it has it.`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] RocMarshal commented on a diff in pull request #21999: [FLINK-29816][streaming] Fix SteamTask skip exception in userfunction of ProcessWindowFunction during Steam.restore
Posted by "RocMarshal (via GitHub)" <gi...@apache.org>.
RocMarshal commented on code in PR #21999:
URL: https://github.com/apache/flink/pull/21999#discussion_r1114510855
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java:
##########
@@ -261,6 +261,12 @@
*/
private volatile boolean isRunning;
+ /**
+ * Flag to mark the task "in operation", in which case check needs to be process the exceptions
+ * thrown in the initializing duration.
+ */
+ private volatile boolean isInitializing;
Review Comment:
> Ah, I guess he just want to follow the `ExecutionState`'s initializing state, the corresponding java doc indicates that:
@akalash @reswqa yes. I described the section at the jira ticket. Execution has the state `INITIALIZING`, and the `StreamTask` does not. So I used `isInitializing`. of course, `isRestoring` is suitable for the scenario, too. CC @1996fanrui
--
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] 1996fanrui commented on a diff in pull request #21999: [FLINK-29816][streaming] Fix the bug that StreamTask doesn't handle exception during restoring
Posted by "1996fanrui (via GitHub)" <gi...@apache.org>.
1996fanrui commented on code in PR #21999:
URL: https://github.com/apache/flink/pull/21999#discussion_r1115773340
##########
flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTest.java:
##########
@@ -410,6 +410,42 @@ public CloseException() {
}
}
+ @Test
+ public void testAsyncExceptionCanBeHandledDuringRestoring() throws Exception {
Review Comment:
Thanks for the suggestion. LGTM
--
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] RocMarshal commented on a diff in pull request #21999: [FLINK-29816][streaming] Fix the bug that StreamTask doesn't handle exception during restoring
Posted by "RocMarshal (via GitHub)" <gi...@apache.org>.
RocMarshal commented on code in PR #21999:
URL: https://github.com/apache/flink/pull/21999#discussion_r1115463498
##########
flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskITCase.java:
##########
@@ -110,6 +118,47 @@ public void testFailInEndOfConstructor() throws Exception {
.build(EXECUTOR_RESOURCE.getExecutor()));
}
+ @Test
+ public void streamTaskAsyncExceptionHandlerHandleExceptionInInitializingDuration()
+ throws Exception {
+ MockEnvironment mockEnvironment = MockEnvironment.builder().build();
+ RuntimeException expectedException = new RuntimeException("RUNTIME EXCEPTION");
+
+ mockEnvironment.setExpectedExternalFailureCause(AsynchronousException.class);
+ final String expectedErrorMessage = "EXPECTED_ERROR MESSAGE";
+
+ NoOpStreamTask noOpStreamTask =
+ new NoOpStreamTask(mockEnvironment) {
+ volatile boolean handledAsyncException = false;
+
+ @Override
+ protected void init() throws Exception {
+ super.init();
+
+ CompletableFuture.runAsync(
+ () -> {
+ this.handleAsyncException(
+ expectedErrorMessage, expectedException);
+ handledAsyncException = true;
+ });
+ while (!handledAsyncException) {
+ ThreadUtils.sleep(Duration.ofMillis(10));
+ }
+ }
+ };
+ noOpStreamTask.invoke();
Review Comment:
precise~
--
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] 1996fanrui commented on pull request #21999: [FLINK-29816][streaming] Fix the bug that StreamTask doesn't handle exception during restoring
Posted by "1996fanrui (via GitHub)" <gi...@apache.org>.
1996fanrui commented on PR #21999:
URL: https://github.com/apache/flink/pull/21999#issuecomment-1442822176
Please squish these commits, and follow the rules that @reswqa mentioned before.
> This PR will eventually have two commits in the following order:
> 1.[hotfix] Migrate StreamTaskTest to Junit5 and Assertj
> 2.[[FLINK-29816](https://issues.apache.org/jira/browse/FLINK-29816)][streaming] Fix the bug that StreamTask xxxx
And note the commit order, hotfix commit should before your commit, otherwises, you need to write new test with junit4, and then upgrade it in the next commit.
--
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] reswqa commented on a diff in pull request #21999: [FLINK-29816][streaming] Fix the bug that StreamTask doesn't handle exception during restoring
Posted by "reswqa (via GitHub)" <gi...@apache.org>.
reswqa commented on code in PR #21999:
URL: https://github.com/apache/flink/pull/21999#discussion_r1116565034
##########
flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTest.java:
##########
@@ -362,30 +357,28 @@ private void testSyncSavepointWithEndInput(
"savepointResult");
harness.processAll();
- Assert.assertEquals(expectEndInput, TestBoundedOneInputStreamOperator.isInputEnded());
+ assertThat(TestBoundedOneInputStreamOperator.isInputEnded()).isEqualTo(expectEndInput);
}
@Test
- public void testCleanUpExceptionSuppressing() throws Exception {
+ void testCleanUpExceptionSuppressing() throws Exception {
try (StreamTaskMailboxTestHarness<String> testHarness =
new StreamTaskMailboxTestHarnessBuilder<>(OneInputStreamTask::new, STRING_TYPE_INFO)
.addInput(STRING_TYPE_INFO)
.setupOutputForSingletonOperatorChain(new FailingTwiceOperator())
.build()) {
- try {
- testHarness.processElement(new StreamRecord<>("Doesn't matter", 0));
- throw new RuntimeException("Expected an exception but ran successfully");
- } catch (Exception ex) {
- ExceptionUtils.assertThrowable(ex, ExpectedTestException.class);
- }
+ assertThatThrownBy(
+ () -> {
+ testHarness.processElement(new StreamRecord<>("Doesn't matter", 0));
+ throw new RuntimeException(
+ "Expected an exception but ran successfully");
+ })
+ .isInstanceOf(ExpectedTestException.class);
- try {
- testHarness.finishProcessing();
- } catch (Exception ex) {
- // todo: checking for suppression if there are more exceptions during cleanup
- ExceptionUtils.assertThrowable(ex, FailingTwiceOperator.CloseException.class);
- }
+ // todo: checking for suppression if there are more exceptions during cleanup
+ assertThatThrownBy(testHarness::finishProcessing)
+ .isInstanceOf(FailingTwiceOperator.CloseException.class);
Review Comment:
We don't need to go back to the original implementation. You can see my last comment.
--
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] 1996fanrui commented on a diff in pull request #21999: [FLINK-29816][streaming] Fix the bug that SteamTask skip handle exception during restoring.
Posted by "1996fanrui (via GitHub)" <gi...@apache.org>.
1996fanrui commented on code in PR #21999:
URL: https://github.com/apache/flink/pull/21999#discussion_r1115402015
##########
flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskITCase.java:
##########
@@ -110,6 +118,42 @@ public void testFailInEndOfConstructor() throws Exception {
.build(EXECUTOR_RESOURCE.getExecutor()));
}
+ @Test
+ public void streamTaskAsyncExceptionHandlerHandleExceptionInInitializingDuration()
+ throws Exception {
+ MockEnvironment mockEnvironment = MockEnvironment.builder().build();
+ RuntimeException expectedException = new RuntimeException("RUNTIME EXCEPTION");
+
+ mockEnvironment.setExpectedExternalFailureCause(AsynchronousException.class);
+ final String expectedErrorMessage = "EXPECTED_ERROR MESSAGE";
+
+ NoOpStreamTask noOpStreamTask =
+ new NoOpStreamTask(mockEnvironment) {
+ @Override
+ protected void init() throws Exception {
+ super.init();
+ CompletableFuture.runAsync(
+ () -> {
+ assertFalse(this.isRunning());
+ assertTrue(this.isInitializing());
+ this.handleAsyncException(
+ expectedErrorMessage, expectedException);
+ });
+ }
Review Comment:
Flink test doesn't recommend using the `sleep`, you can use the `CompletableFuture future = new CompletableFuture()`.
Main thread call the `future.get()`, and async thread calls the `future.complete(null)` after some operators done.
You can take a look this PR: https://github.com/apache/flink/pull/21304/files
--
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] RocMarshal commented on a diff in pull request #21999: [FLINK-29816][streaming] Fix the bug that StreamTask doesn't handle exception during restoring
Posted by "RocMarshal (via GitHub)" <gi...@apache.org>.
RocMarshal commented on code in PR #21999:
URL: https://github.com/apache/flink/pull/21999#discussion_r1117078618
##########
flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTest.java:
##########
@@ -362,30 +358,35 @@ private void testSyncSavepointWithEndInput(
"savepointResult");
harness.processAll();
- Assert.assertEquals(expectEndInput, TestBoundedOneInputStreamOperator.isInputEnded());
+ assertThat(TestBoundedOneInputStreamOperator.isInputEnded()).isEqualTo(expectEndInput);
}
@Test
- public void testCleanUpExceptionSuppressing() throws Exception {
+ void testCleanUpExceptionSuppressing() throws Exception {
try (StreamTaskMailboxTestHarness<String> testHarness =
new StreamTaskMailboxTestHarnessBuilder<>(OneInputStreamTask::new, STRING_TYPE_INFO)
.addInput(STRING_TYPE_INFO)
.setupOutputForSingletonOperatorChain(new FailingTwiceOperator())
.build()) {
- try {
- testHarness.processElement(new StreamRecord<>("Doesn't matter", 0));
- throw new RuntimeException("Expected an exception but ran successfully");
- } catch (Exception ex) {
- ExceptionUtils.assertThrowable(ex, ExpectedTestException.class);
- }
+ assertThatThrownBy(
+ () ->
+ testHarness.processElement(
+ new StreamRecord<>("Doesn't matter", 0)))
+ .satisfies(
+ (Consumer<Throwable>)
+ throwable ->
+ ExceptionUtils.assertThrowable(
+ throwable, ExpectedTestException.class));
- try {
- testHarness.finishProcessing();
- } catch (Exception ex) {
- // todo: checking for suppression if there are more exceptions during cleanup
- ExceptionUtils.assertThrowable(ex, FailingTwiceOperator.CloseException.class);
- }
+ // todo: checking for suppression if there are more exceptions during cleanup
+ assertThatThrownBy(testHarness::finishProcessing)
+ .satisfies(
+ (ThrowingConsumer<Throwable>)
+ throwable ->
+ ExceptionUtils.assertThrowable(
+ throwable,
+ FailingTwiceOperator.CloseException.class));
Review Comment:
<img width="1233" alt="image" src="https://user-images.githubusercontent.com/64569824/221202347-41aeb24a-2324-46cb-9985-0a3e5595faf6.png">
So, let's keep the current lines ?
--
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] 1996fanrui commented on pull request #21999: [FLINK-29816][streaming] Fix the bug that StreamTask doesn't handle exception during restoring
Posted by "1996fanrui (via GitHub)" <gi...@apache.org>.
1996fanrui commented on PR #21999:
URL: https://github.com/apache/flink/pull/21999#issuecomment-1443901443
And please squash all review commits to the first commit after addressed all comments, thanks~
--
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] 1996fanrui commented on a diff in pull request #21999: [FLINK-29816][streaming] Fix the bug that StreamTask doesn't handle exception during restoring
Posted by "1996fanrui (via GitHub)" <gi...@apache.org>.
1996fanrui commented on code in PR #21999:
URL: https://github.com/apache/flink/pull/21999#discussion_r1117858061
##########
flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTest.java:
##########
@@ -362,30 +358,35 @@ private void testSyncSavepointWithEndInput(
"savepointResult");
harness.processAll();
- Assert.assertEquals(expectEndInput, TestBoundedOneInputStreamOperator.isInputEnded());
+ assertThat(TestBoundedOneInputStreamOperator.isInputEnded()).isEqualTo(expectEndInput);
}
@Test
- public void testCleanUpExceptionSuppressing() throws Exception {
+ void testCleanUpExceptionSuppressing() throws Exception {
try (StreamTaskMailboxTestHarness<String> testHarness =
new StreamTaskMailboxTestHarnessBuilder<>(OneInputStreamTask::new, STRING_TYPE_INFO)
.addInput(STRING_TYPE_INFO)
.setupOutputForSingletonOperatorChain(new FailingTwiceOperator())
.build()) {
- try {
- testHarness.processElement(new StreamRecord<>("Doesn't matter", 0));
- throw new RuntimeException("Expected an exception but ran successfully");
- } catch (Exception ex) {
- ExceptionUtils.assertThrowable(ex, ExpectedTestException.class);
- }
+ assertThatThrownBy(
+ () ->
+ testHarness.processElement(
+ new StreamRecord<>("Doesn't matter", 0)))
+ .satisfies(
+ (Consumer<Throwable>)
+ throwable ->
+ ExceptionUtils.assertThrowable(
+ throwable, ExpectedTestException.class));
- try {
- testHarness.finishProcessing();
- } catch (Exception ex) {
- // todo: checking for suppression if there are more exceptions during cleanup
- ExceptionUtils.assertThrowable(ex, FailingTwiceOperator.CloseException.class);
- }
+ // todo: checking for suppression if there are more exceptions during cleanup
+ assertThatThrownBy(testHarness::finishProcessing)
+ .satisfies(
+ (ThrowingConsumer<Throwable>)
+ throwable ->
+ ExceptionUtils.assertThrowable(
+ throwable,
+ FailingTwiceOperator.CloseException.class));
Review Comment:
Hi @RocMarshal , thanks for your clarification, I got it.
Hi @reswqa , I found `FlinkAssertions.anyCauseMatches` is similar with `ExceptionUtils.assertThrowable`, and it can find the throwable from more deeper level, and it's better than `ExceptionUtils.assertThrowable`.
What do you think?
```suggestion
.satisfies(
(ThrowingConsumer<Throwable>)
throwable ->
ExceptionUtils.assertThrowable(
throwable,
FailingTwiceOperator.CloseException.class));
```
--
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] RocMarshal commented on a diff in pull request #21999: [FLINK-29816][streaming] Fix the bug that StreamTask doesn't handle exception during restoring
Posted by "RocMarshal (via GitHub)" <gi...@apache.org>.
RocMarshal commented on code in PR #21999:
URL: https://github.com/apache/flink/pull/21999#discussion_r1118016721
##########
flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTest.java:
##########
@@ -362,30 +358,35 @@ private void testSyncSavepointWithEndInput(
"savepointResult");
harness.processAll();
- Assert.assertEquals(expectEndInput, TestBoundedOneInputStreamOperator.isInputEnded());
+ assertThat(TestBoundedOneInputStreamOperator.isInputEnded()).isEqualTo(expectEndInput);
}
@Test
- public void testCleanUpExceptionSuppressing() throws Exception {
+ void testCleanUpExceptionSuppressing() throws Exception {
try (StreamTaskMailboxTestHarness<String> testHarness =
new StreamTaskMailboxTestHarnessBuilder<>(OneInputStreamTask::new, STRING_TYPE_INFO)
.addInput(STRING_TYPE_INFO)
.setupOutputForSingletonOperatorChain(new FailingTwiceOperator())
.build()) {
- try {
- testHarness.processElement(new StreamRecord<>("Doesn't matter", 0));
- throw new RuntimeException("Expected an exception but ran successfully");
- } catch (Exception ex) {
- ExceptionUtils.assertThrowable(ex, ExpectedTestException.class);
- }
+ assertThatThrownBy(
+ () ->
+ testHarness.processElement(
+ new StreamRecord<>("Doesn't matter", 0)))
+ .satisfies(
+ (Consumer<Throwable>)
+ throwable ->
+ ExceptionUtils.assertThrowable(
+ throwable, ExpectedTestException.class));
Review Comment:
updated.
--
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] RocMarshal commented on pull request #21999: [FLINK-29816][streaming] Fix the bug that StreamTask doesn't handle exception during restoring
Posted by "RocMarshal (via GitHub)" <gi...@apache.org>.
RocMarshal commented on PR #21999:
URL: https://github.com/apache/flink/pull/21999#issuecomment-1445562489
> @RocMarshal Would you mind creating the `backport` pull request of `release-1.17` and `release-1.16` after all reviewers approved.
@reswqa Yes. I'm glad to ~
--
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] RocMarshal commented on a diff in pull request #21999: [FLINK-29816][streaming] Fix the bug that StreamTask doesn't handle exception during restoring
Posted by "RocMarshal (via GitHub)" <gi...@apache.org>.
RocMarshal commented on code in PR #21999:
URL: https://github.com/apache/flink/pull/21999#discussion_r1116555051
##########
flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTest.java:
##########
@@ -362,30 +357,28 @@ private void testSyncSavepointWithEndInput(
"savepointResult");
harness.processAll();
- Assert.assertEquals(expectEndInput, TestBoundedOneInputStreamOperator.isInputEnded());
+ assertThat(TestBoundedOneInputStreamOperator.isInputEnded()).isEqualTo(expectEndInput);
}
@Test
- public void testCleanUpExceptionSuppressing() throws Exception {
+ void testCleanUpExceptionSuppressing() throws Exception {
try (StreamTaskMailboxTestHarness<String> testHarness =
new StreamTaskMailboxTestHarnessBuilder<>(OneInputStreamTask::new, STRING_TYPE_INFO)
.addInput(STRING_TYPE_INFO)
.setupOutputForSingletonOperatorChain(new FailingTwiceOperator())
.build()) {
- try {
- testHarness.processElement(new StreamRecord<>("Doesn't matter", 0));
- throw new RuntimeException("Expected an exception but ran successfully");
- } catch (Exception ex) {
- ExceptionUtils.assertThrowable(ex, ExpectedTestException.class);
- }
+ assertThatThrownBy(
+ () -> {
+ testHarness.processElement(new StreamRecord<>("Doesn't matter", 0));
+ throw new RuntimeException(
+ "Expected an exception but ran successfully");
+ })
+ .isInstanceOf(ExpectedTestException.class);
- try {
- testHarness.finishProcessing();
- } catch (Exception ex) {
- // todo: checking for suppression if there are more exceptions during cleanup
- ExceptionUtils.assertThrowable(ex, FailingTwiceOperator.CloseException.class);
- }
+ // todo: checking for suppression if there are more exceptions during cleanup
+ assertThatThrownBy(testHarness::finishProcessing)
+ .isInstanceOf(FailingTwiceOperator.CloseException.class);
Review Comment:
nice catch~I compared `assertThrowable`logic with `isInstanceOf `, It's indeed as your description.
I try to find the logic-equal API to replaced it but no result.
So, I rollback to the `assertThrowable` with original implementation.
Of course, I'd like to accept any suggestions helpful .
--
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] RocMarshal commented on a diff in pull request #21999: [FLINK-29816][streaming] Fix the bug that StreamTask doesn't handle exception during restoring
Posted by "RocMarshal (via GitHub)" <gi...@apache.org>.
RocMarshal commented on code in PR #21999:
URL: https://github.com/apache/flink/pull/21999#discussion_r1116555777
##########
flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTest.java:
##########
@@ -362,30 +357,28 @@ private void testSyncSavepointWithEndInput(
"savepointResult");
harness.processAll();
- Assert.assertEquals(expectEndInput, TestBoundedOneInputStreamOperator.isInputEnded());
+ assertThat(TestBoundedOneInputStreamOperator.isInputEnded()).isEqualTo(expectEndInput);
}
@Test
- public void testCleanUpExceptionSuppressing() throws Exception {
+ void testCleanUpExceptionSuppressing() throws Exception {
try (StreamTaskMailboxTestHarness<String> testHarness =
new StreamTaskMailboxTestHarnessBuilder<>(OneInputStreamTask::new, STRING_TYPE_INFO)
.addInput(STRING_TYPE_INFO)
.setupOutputForSingletonOperatorChain(new FailingTwiceOperator())
.build()) {
- try {
- testHarness.processElement(new StreamRecord<>("Doesn't matter", 0));
- throw new RuntimeException("Expected an exception but ran successfully");
- } catch (Exception ex) {
- ExceptionUtils.assertThrowable(ex, ExpectedTestException.class);
- }
+ assertThatThrownBy(
+ () -> {
+ testHarness.processElement(new StreamRecord<>("Doesn't matter", 0));
+ throw new RuntimeException(
Review Comment:
It's the old logic line from migration. but redundant.
This lines rollback to the original lines.
--
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] RocMarshal commented on a diff in pull request #21999: [FLINK-29816][streaming] Fix the bug that StreamTask doesn't handle exception during restoring
Posted by "RocMarshal (via GitHub)" <gi...@apache.org>.
RocMarshal commented on code in PR #21999:
URL: https://github.com/apache/flink/pull/21999#discussion_r1116556431
##########
flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTest.java:
##########
@@ -1085,27 +1081,27 @@ public void testNotifyCheckpointOnClosedOperator() throws Throwable {
harness.streamTask.notifyCheckpointCompleteAsync(1);
harness.streamTask.runMailboxStep();
- assertEquals(1, ClosingOperator.notified.get());
- assertFalse(ClosingOperator.closed.get());
+ assertThat(ClosingOperator.notified.get()).isOne();
Review Comment:
Thank you~. I updated it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] RocMarshal commented on a diff in pull request #21999: [FLINK-29816][streaming] Fix the bug that SteamTask skip handle exception during restoring.
Posted by "RocMarshal (via GitHub)" <gi...@apache.org>.
RocMarshal commented on code in PR #21999:
URL: https://github.com/apache/flink/pull/21999#discussion_r1114607304
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java:
##########
@@ -261,6 +261,12 @@
*/
private volatile boolean isRunning;
+ /**
+ * Flag to mark the task "in operation", in which case check needs to be process the exceptions
+ * thrown in the initializing duration.
+ */
+ private volatile boolean isInitializing;
Review Comment:
Thanks for the clarify. Maybe I got it.
We need to define or describe different states for different abstract level as accurately as possible, right?
If so, I'd prefer your description `isRestoring`, 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] RocMarshal commented on a diff in pull request #21999: [FLINK-29816][streaming] Fix the bug that StreamTask doesn't handle exception during restoring
Posted by "RocMarshal (via GitHub)" <gi...@apache.org>.
RocMarshal commented on code in PR #21999:
URL: https://github.com/apache/flink/pull/21999#discussion_r1115463498
##########
flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskITCase.java:
##########
@@ -110,6 +118,47 @@ public void testFailInEndOfConstructor() throws Exception {
.build(EXECUTOR_RESOURCE.getExecutor()));
}
+ @Test
+ public void streamTaskAsyncExceptionHandlerHandleExceptionInInitializingDuration()
+ throws Exception {
+ MockEnvironment mockEnvironment = MockEnvironment.builder().build();
+ RuntimeException expectedException = new RuntimeException("RUNTIME EXCEPTION");
+
+ mockEnvironment.setExpectedExternalFailureCause(AsynchronousException.class);
+ final String expectedErrorMessage = "EXPECTED_ERROR MESSAGE";
+
+ NoOpStreamTask noOpStreamTask =
+ new NoOpStreamTask(mockEnvironment) {
+ volatile boolean handledAsyncException = false;
+
+ @Override
+ protected void init() throws Exception {
+ super.init();
+
+ CompletableFuture.runAsync(
+ () -> {
+ this.handleAsyncException(
+ expectedErrorMessage, expectedException);
+ handledAsyncException = true;
+ });
+ while (!handledAsyncException) {
+ ThreadUtils.sleep(Duration.ofMillis(10));
+ }
+ }
+ };
+ noOpStreamTask.invoke();
Review Comment:
precise~ & updated.
--
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] RocMarshal commented on a diff in pull request #21999: [FLINK-29816][streaming] Fix SteamTask skip exception in userfunction of ProcessWindowFunction during Steam.restore
Posted by "RocMarshal (via GitHub)" <gi...@apache.org>.
RocMarshal commented on code in PR #21999:
URL: https://github.com/apache/flink/pull/21999#discussion_r1114508117
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java:
##########
@@ -261,6 +261,12 @@
*/
private volatile boolean isRunning;
+ /**
+ * Flag to mark the task "in operation", in which case check needs to be process the exceptions
Review Comment:
> 1. What is the meaning of `in operation`?
It would be better to clarify the `in operation` to `in Initializing operation` or `in restoring operation`
--
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] RocMarshal commented on pull request #21999: [FLINK-29816][streaming] Fix the bug that StreamTask doesn't handle exception during restoring
Posted by "RocMarshal (via GitHub)" <gi...@apache.org>.
RocMarshal commented on PR #21999:
URL: https://github.com/apache/flink/pull/21999#issuecomment-1441946575
> Thanks @RocMarshal. If you like, you'd better migrate all the tests in `StreamTaskTest` to `Junit5` and `AssertJ` in a hotfix commit. No matter what, the newly introduced test should no longer use the `hamcrest` or `Junit` assertion according to our [Code Style and Quality Guide](https://flink.apache.org/how-to-contribute/code-style-and-quality-common/#tooling).
Hi, @reswqa I'm fine to migrate the tests based on https://flink.apache.org/how-to-contribute/code-style-and-quality-common/#tooling and I'm familiar to the migration process.
What do you mean is that I should split the the comments-address and tests migration by different commits in the current PR ?
I just want to confirm my confusion before we start the change, because I see you caught the hotfix-commit instead of hotfix-ticket. Thx you~.
--
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] RocMarshal commented on pull request #21999: [FLINK-29816][streaming] Fix the bug that StreamTask doesn't handle exception during restoring
Posted by "RocMarshal (via GitHub)" <gi...@apache.org>.
RocMarshal commented on PR #21999:
URL: https://github.com/apache/flink/pull/21999#issuecomment-1445265696
Hi, @1996fanrui @reswqa thanks for the comments & help.
I make change based on your suggestion. PTAL if you're in free time.
CC @akalash
--
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] 1996fanrui commented on a diff in pull request #21999: [FLINK-29816][streaming] Fix the bug that StreamTask doesn't handle exception during restoring
Posted by "1996fanrui (via GitHub)" <gi...@apache.org>.
1996fanrui commented on code in PR #21999:
URL: https://github.com/apache/flink/pull/21999#discussion_r1115773340
##########
flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTest.java:
##########
@@ -410,6 +410,42 @@ public CloseException() {
}
}
+ @Test
+ public void testAsyncExceptionCanBeHandledDuringRestoring() throws Exception {
Review Comment:
LGTM
--
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] akalash commented on a diff in pull request #21999: [FLINK-29816][streaming] Fix SteamTask skip exception in userfunction of ProcessWindowFunction during Steam.restore
Posted by "akalash (via GitHub)" <gi...@apache.org>.
akalash commented on code in PR #21999:
URL: https://github.com/apache/flink/pull/21999#discussion_r1114577578
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java:
##########
@@ -261,6 +261,12 @@
*/
private volatile boolean isRunning;
+ /**
+ * Flag to mark the task "in operation", in which case check needs to be process the exceptions
+ * thrown in the initializing duration.
+ */
+ private volatile boolean isInitializing;
Review Comment:
Hm, I see, I need to think about that.
Just in case, `StreamTask` and `ExecutionState` are different levels of abstraction and we don't have to keep them the same. We can do it but I'm not sure yet that it has any advantages. We should consider that `StreamTask` has two methods `init` and `restore` where `init` is part of `restore`. The `isInitializing` follows the `restore` but it says that it follows the `initialization` this `initialization` is have nothing in common with `init` method but it relates to `ExecutionState` which relates to `Task` rather than `StreamTask`.
So I hope you understand now why I am confused.
--
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] RocMarshal commented on a diff in pull request #21999: [FLINK-29816][streaming] Fix the bug that StreamTask doesn't handle exception during restoring
Posted by "RocMarshal (via GitHub)" <gi...@apache.org>.
RocMarshal commented on code in PR #21999:
URL: https://github.com/apache/flink/pull/21999#discussion_r1116556097
##########
flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTest.java:
##########
@@ -1121,13 +1117,14 @@ private void testFailToConfirmCheckpointMessage(Consumer<StreamTask<?, ?>> consu
StreamTaskMailboxTestHarness<Integer> harness =
builder.setupOutputForSingletonOperatorChain(streamMap).build();
- try {
- consumer.accept(harness.streamTask);
- harness.streamTask.runMailboxLoop();
- fail();
- } catch (ExpectedTestException expected) {
- // expected exceptionestProcessWithUnAvailableInput
- }
+ // expected exceptionestProcessWithUnAvailableInput
+ assertThatThrownBy(
+ () -> {
+ consumer.accept(harness.streamTask);
+ harness.streamTask.runMailboxLoop();
+ fail(null);
Review Comment:
It's from the old line.
And I removed it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] 1996fanrui commented on a diff in pull request #21999: [FLINK-29816][streaming] Fix the bug that StreamTask doesn't handle exception during restoring
Posted by "1996fanrui (via GitHub)" <gi...@apache.org>.
1996fanrui commented on code in PR #21999:
URL: https://github.com/apache/flink/pull/21999#discussion_r1117222444
##########
flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTest.java:
##########
@@ -362,30 +358,35 @@ private void testSyncSavepointWithEndInput(
"savepointResult");
harness.processAll();
- Assert.assertEquals(expectEndInput, TestBoundedOneInputStreamOperator.isInputEnded());
+ assertThat(TestBoundedOneInputStreamOperator.isInputEnded()).isEqualTo(expectEndInput);
}
@Test
- public void testCleanUpExceptionSuppressing() throws Exception {
+ void testCleanUpExceptionSuppressing() throws Exception {
try (StreamTaskMailboxTestHarness<String> testHarness =
new StreamTaskMailboxTestHarnessBuilder<>(OneInputStreamTask::new, STRING_TYPE_INFO)
.addInput(STRING_TYPE_INFO)
.setupOutputForSingletonOperatorChain(new FailingTwiceOperator())
.build()) {
- try {
- testHarness.processElement(new StreamRecord<>("Doesn't matter", 0));
- throw new RuntimeException("Expected an exception but ran successfully");
- } catch (Exception ex) {
- ExceptionUtils.assertThrowable(ex, ExpectedTestException.class);
- }
+ assertThatThrownBy(
+ () ->
+ testHarness.processElement(
+ new StreamRecord<>("Doesn't matter", 0)))
+ .satisfies(
+ (Consumer<Throwable>)
+ throwable ->
+ ExceptionUtils.assertThrowable(
+ throwable, ExpectedTestException.class));
Review Comment:
```suggestion
.satisfies(anyCauseMatches(ExpectedTestException.class));
```
--
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] reswqa commented on pull request #21999: [FLINK-29816][streaming] Fix the bug that StreamTask doesn't handle exception during restoring
Posted by "reswqa (via GitHub)" <gi...@apache.org>.
reswqa commented on PR #21999:
URL: https://github.com/apache/flink/pull/21999#issuecomment-1445393395
@RocMarshal Would you mind creating the `backport` pull request of `release-1.17` and `release-1.16` after all reviewers approved.
--
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] 1996fanrui merged pull request #21999: [FLINK-29816][streaming] Fix the bug that StreamTask doesn't handle exception during restoring
Posted by "1996fanrui (via GitHub)" <gi...@apache.org>.
1996fanrui merged PR #21999:
URL: https://github.com/apache/flink/pull/21999
--
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] reswqa commented on a diff in pull request #21999: [FLINK-29816][streaming] Fix SteamTask skip exception in userfunction of ProcessWindowFunction during Steam.restore
Posted by "reswqa (via GitHub)" <gi...@apache.org>.
reswqa commented on code in PR #21999:
URL: https://github.com/apache/flink/pull/21999#discussion_r1114482598
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java:
##########
@@ -261,6 +261,12 @@
*/
private volatile boolean isRunning;
+ /**
+ * Flag to mark the task "in operation", in which case check needs to be process the exceptions
Review Comment:
It's a little obscure here
1. What is the meaning of `in operation`?
2. IMO, we'd better describe when the `StreamTask` will be in this state, rather than say that this state is for handling 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] RocMarshal commented on a diff in pull request #21999: [FLINK-29816][streaming] Fix SteamTask skip exception in userfunction of ProcessWindowFunction during Steam.restore
Posted by "RocMarshal (via GitHub)" <gi...@apache.org>.
RocMarshal commented on code in PR #21999:
URL: https://github.com/apache/flink/pull/21999#discussion_r1114532431
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java:
##########
@@ -261,6 +261,12 @@
*/
private volatile boolean isRunning;
+ /**
+ * Flag to mark the task "in operation", in which case check needs to be process the exceptions
Review Comment:
> 2\. IMO, we'd better describe when the `StreamTask` will be in this state, rather than say that this state is for handling exception.
SGTM.
--
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] 1996fanrui commented on a diff in pull request #21999: [FLINK-29816][streaming] Fix the bug that SteamTask skip handle exception during restoring.
Posted by "1996fanrui (via GitHub)" <gi...@apache.org>.
1996fanrui commented on code in PR #21999:
URL: https://github.com/apache/flink/pull/21999#discussion_r1115402015
##########
flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskITCase.java:
##########
@@ -110,6 +118,42 @@ public void testFailInEndOfConstructor() throws Exception {
.build(EXECUTOR_RESOURCE.getExecutor()));
}
+ @Test
+ public void streamTaskAsyncExceptionHandlerHandleExceptionInInitializingDuration()
+ throws Exception {
+ MockEnvironment mockEnvironment = MockEnvironment.builder().build();
+ RuntimeException expectedException = new RuntimeException("RUNTIME EXCEPTION");
+
+ mockEnvironment.setExpectedExternalFailureCause(AsynchronousException.class);
+ final String expectedErrorMessage = "EXPECTED_ERROR MESSAGE";
+
+ NoOpStreamTask noOpStreamTask =
+ new NoOpStreamTask(mockEnvironment) {
+ @Override
+ protected void init() throws Exception {
+ super.init();
+ CompletableFuture.runAsync(
+ () -> {
+ assertFalse(this.isRunning());
+ assertTrue(this.isInitializing());
+ this.handleAsyncException(
+ expectedErrorMessage, expectedException);
+ });
+ }
Review Comment:
Flink test doesn't recommend using the `sleep`, you can use the `CompletableFuture future = new CompletableFuture()`.
Main thread call the `future.wait()`, and async thread calls the `future.complete(null)` after some operators done.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] 1996fanrui commented on a diff in pull request #21999: [FLINK-29816][streaming] Fix the bug that StreamTask doesn't handle exception during restoring
Posted by "1996fanrui (via GitHub)" <gi...@apache.org>.
1996fanrui commented on code in PR #21999:
URL: https://github.com/apache/flink/pull/21999#discussion_r1116694579
##########
flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTest.java:
##########
@@ -1103,35 +1143,41 @@ private void testFailToConfirmCheckpointMessage(Consumer<StreamTask<?, ?>> consu
* finished.
*/
@Test
- public void testCheckpointFailueOnClosedOperator() throws Throwable {
+ void testCheckpointFailueOnClosedOperator() throws Exception {
ClosingOperator<Integer> operator = new ClosingOperator<>();
StreamTaskMailboxTestHarnessBuilder<Integer> builder =
new StreamTaskMailboxTestHarnessBuilder<>(
OneInputStreamTask::new, BasicTypeInfo.INT_TYPE_INFO)
.addInput(BasicTypeInfo.INT_TYPE_INFO);
try (StreamTaskMailboxTestHarness<Integer> harness =
builder.setupOutputForSingletonOperatorChain(operator).build()) {
- // keeps the mailbox from suspending
- harness.setAutoProcess(false);
- harness.processElement(new StreamRecord<>(1));
-
- harness.streamTask.operatorChain.finishOperators(
- harness.streamTask.getActionExecutor(), StopMode.DRAIN);
- harness.streamTask.operatorChain.closeAllOperators();
- assertTrue(ClosingOperator.closed.get());
-
- harness.streamTask.triggerCheckpointOnBarrier(
- new CheckpointMetaData(1, 0),
- CheckpointOptions.forCheckpointWithDefaultLocation(),
- new CheckpointMetricsBuilder());
- } catch (Exception ex) {
- ExceptionUtils.assertThrowableWithMessage(
- ex, "OperatorChain and Task should never be closed at this point");
+ assertThatThrownBy(
+ () -> {
+ // keeps the mailbox from suspending
+ harness.setAutoProcess(false);
+ harness.processElement(new StreamRecord<>(1));
+
+ harness.streamTask.operatorChain.finishOperators(
+ harness.streamTask.getActionExecutor(), StopMode.DRAIN);
+ harness.streamTask.operatorChain.closeAllOperators();
+ assertThat(ClosingOperator.closed.get()).isTrue();
+
+ harness.streamTask.triggerCheckpointOnBarrier(
+ new CheckpointMetaData(1, 0),
+ CheckpointOptions.forCheckpointWithDefaultLocation(),
+ new CheckpointMetricsBuilder());
+ })
+ .satisfies(
+ (Consumer<Throwable>)
+ throwable ->
+ ExceptionUtils.assertThrowableWithMessage(
+ throwable,
+ "OperatorChain and Task should never be closed at this point"));
Review Comment:
If it works, please check all `satisfies`.
--
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] RocMarshal commented on a diff in pull request #21999: [FLINK-29816][streaming] Fix the bug that StreamTask doesn't handle exception during restoring
Posted by "RocMarshal (via GitHub)" <gi...@apache.org>.
RocMarshal commented on code in PR #21999:
URL: https://github.com/apache/flink/pull/21999#discussion_r1117060917
##########
flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTest.java:
##########
@@ -1182,13 +1228,13 @@ public void testThreadInvariants() throws Throwable {
});
runningTask.invocationFuture.get();
- assertThat(
- runningTask.streamTask.getTaskClassLoader(), is(sameInstance(taskClassLoader)));
+ assertThat(runningTask.streamTask.getTaskClassLoader()).isSameAs(taskClassLoader);
+ assertThat(runningTask.streamTask.getTaskClassLoader()).isSameAs(taskClassLoader);
Review Comment:
yes, nice catch!
--
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] 1996fanrui commented on a diff in pull request #21999: [FLINK-29816][streaming] Fix the bug that StreamTask doesn't handle exception during restoring
Posted by "1996fanrui (via GitHub)" <gi...@apache.org>.
1996fanrui commented on code in PR #21999:
URL: https://github.com/apache/flink/pull/21999#discussion_r1117225403
##########
flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTest.java:
##########
@@ -1103,35 +1143,41 @@ private void testFailToConfirmCheckpointMessage(Consumer<StreamTask<?, ?>> consu
* finished.
*/
@Test
- public void testCheckpointFailueOnClosedOperator() throws Throwable {
+ void testCheckpointFailueOnClosedOperator() throws Exception {
ClosingOperator<Integer> operator = new ClosingOperator<>();
StreamTaskMailboxTestHarnessBuilder<Integer> builder =
new StreamTaskMailboxTestHarnessBuilder<>(
OneInputStreamTask::new, BasicTypeInfo.INT_TYPE_INFO)
.addInput(BasicTypeInfo.INT_TYPE_INFO);
try (StreamTaskMailboxTestHarness<Integer> harness =
builder.setupOutputForSingletonOperatorChain(operator).build()) {
- // keeps the mailbox from suspending
- harness.setAutoProcess(false);
- harness.processElement(new StreamRecord<>(1));
-
- harness.streamTask.operatorChain.finishOperators(
- harness.streamTask.getActionExecutor(), StopMode.DRAIN);
- harness.streamTask.operatorChain.closeAllOperators();
- assertTrue(ClosingOperator.closed.get());
-
- harness.streamTask.triggerCheckpointOnBarrier(
- new CheckpointMetaData(1, 0),
- CheckpointOptions.forCheckpointWithDefaultLocation(),
- new CheckpointMetricsBuilder());
- } catch (Exception ex) {
- ExceptionUtils.assertThrowableWithMessage(
- ex, "OperatorChain and Task should never be closed at this point");
+ assertThatThrownBy(
+ () -> {
+ // keeps the mailbox from suspending
+ harness.setAutoProcess(false);
+ harness.processElement(new StreamRecord<>(1));
+
+ harness.streamTask.operatorChain.finishOperators(
+ harness.streamTask.getActionExecutor(), StopMode.DRAIN);
+ harness.streamTask.operatorChain.closeAllOperators();
+ assertThat(ClosingOperator.closed.get()).isTrue();
+
+ harness.streamTask.triggerCheckpointOnBarrier(
+ new CheckpointMetaData(1, 0),
+ CheckpointOptions.forCheckpointWithDefaultLocation(),
+ new CheckpointMetricsBuilder());
+ })
+ .satisfies(
+ (Consumer<Throwable>)
+ throwable ->
+ ExceptionUtils.assertThrowableWithMessage(
+ throwable,
+ "OperatorChain and Task should never be closed at this point"));
Review Comment:
```suggestion
.hasRootCauseMessage("OperatorChain and Task should never be closed at this point");
```
It works.
--
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] 1996fanrui commented on a diff in pull request #21999: [FLINK-29816][streaming] Fix the bug that StreamTask doesn't handle exception during restoring
Posted by "1996fanrui (via GitHub)" <gi...@apache.org>.
1996fanrui commented on code in PR #21999:
URL: https://github.com/apache/flink/pull/21999#discussion_r1117858061
##########
flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTest.java:
##########
@@ -362,30 +358,35 @@ private void testSyncSavepointWithEndInput(
"savepointResult");
harness.processAll();
- Assert.assertEquals(expectEndInput, TestBoundedOneInputStreamOperator.isInputEnded());
+ assertThat(TestBoundedOneInputStreamOperator.isInputEnded()).isEqualTo(expectEndInput);
}
@Test
- public void testCleanUpExceptionSuppressing() throws Exception {
+ void testCleanUpExceptionSuppressing() throws Exception {
try (StreamTaskMailboxTestHarness<String> testHarness =
new StreamTaskMailboxTestHarnessBuilder<>(OneInputStreamTask::new, STRING_TYPE_INFO)
.addInput(STRING_TYPE_INFO)
.setupOutputForSingletonOperatorChain(new FailingTwiceOperator())
.build()) {
- try {
- testHarness.processElement(new StreamRecord<>("Doesn't matter", 0));
- throw new RuntimeException("Expected an exception but ran successfully");
- } catch (Exception ex) {
- ExceptionUtils.assertThrowable(ex, ExpectedTestException.class);
- }
+ assertThatThrownBy(
+ () ->
+ testHarness.processElement(
+ new StreamRecord<>("Doesn't matter", 0)))
+ .satisfies(
+ (Consumer<Throwable>)
+ throwable ->
+ ExceptionUtils.assertThrowable(
+ throwable, ExpectedTestException.class));
- try {
- testHarness.finishProcessing();
- } catch (Exception ex) {
- // todo: checking for suppression if there are more exceptions during cleanup
- ExceptionUtils.assertThrowable(ex, FailingTwiceOperator.CloseException.class);
- }
+ // todo: checking for suppression if there are more exceptions during cleanup
+ assertThatThrownBy(testHarness::finishProcessing)
+ .satisfies(
+ (ThrowingConsumer<Throwable>)
+ throwable ->
+ ExceptionUtils.assertThrowable(
+ throwable,
+ FailingTwiceOperator.CloseException.class));
Review Comment:
Hi @RocMarshal , thanks for your clarification, I got it.
Hi @reswqa , I found `FlinkAssertions.anyCauseMatches` is similar with `ExceptionUtils.assertThrowable`, and it can find the throwable from more deeper level, and it's better than `ExceptionUtils.assertThrowable`.
What do you think?
```suggestion
.satisfies(anyCauseMatches(FailingTwiceOperator.CloseException.class));
```
--
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] RocMarshal commented on a diff in pull request #21999: [FLINK-29816][streaming] Fix the bug that SteamTask skip handle exception during restoring.
Posted by "RocMarshal (via GitHub)" <gi...@apache.org>.
RocMarshal commented on code in PR #21999:
URL: https://github.com/apache/flink/pull/21999#discussion_r1115303685
##########
flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskITCase.java:
##########
@@ -110,6 +118,47 @@ public void testFailInEndOfConstructor() throws Exception {
.build(EXECUTOR_RESOURCE.getExecutor()));
}
+ @Test
+ public void streamTaskAsyncExceptionHandlerHandleExceptionInInitializingDuration()
+ throws Exception {
+ MockEnvironment mockEnvironment = MockEnvironment.builder().build();
+ RuntimeException expectedException = new RuntimeException("RUNTIME EXCEPTION");
+
+ mockEnvironment.setExpectedExternalFailureCause(AsynchronousException.class);
+ final String expectedErrorMessage = "EXPECTED_ERROR MESSAGE";
+
+ NoOpStreamTask noOpStreamTask =
+ new NoOpStreamTask(mockEnvironment) {
+ volatile boolean handledAsyncException = false;
+
+ @Override
+ protected void init() throws Exception {
+ super.init();
+
+ CompletableFuture.runAsync(
+ () -> {
+ this.handleAsyncException(
+ expectedErrorMessage, expectedException);
+ handledAsyncException = true;
+ });
+ while (!handledAsyncException) {
+ ThreadUtils.sleep(Duration.ofMillis(10));
+ }
+ }
+ };
+ noOpStreamTask.invoke();
+
+ Optional<? extends Throwable> actualExternalFailureCause =
+ mockEnvironment.getActualExternalFailureCause();
+ final Throwable actualException =
+ actualExternalFailureCause.orElseThrow(
+ () -> new AssertionError("Expected exceptional completion"));
+
+ assertThat(actualException, instanceOf(AsynchronousException.class));
+ assertThat(actualException.getMessage(), is("EXPECTED_ERROR MESSAGE"));
Review Comment:
```suggestion
assertThat(actualException.getMessage(), is(expectedErrorMessage));
```
--
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] akalash commented on a diff in pull request #21999: [FLINK-29816][streaming] Fix SteamTask skip exception in userfunction of ProcessWindowFunction during Steam.restore
Posted by "akalash (via GitHub)" <gi...@apache.org>.
akalash commented on code in PR #21999:
URL: https://github.com/apache/flink/pull/21999#discussion_r1114460469
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java:
##########
@@ -689,6 +695,7 @@ void restoreInternal() throws Exception {
LOG.debug("Re-restore attempt rejected.");
return;
}
+ isInitializing = true;
Review Comment:
I don't see where `isInitializing` is set to `false`
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java:
##########
@@ -261,6 +261,12 @@
*/
private volatile boolean isRunning;
+ /**
+ * Flag to mark the task "in operation", in which case check needs to be process the exceptions
+ * thrown in the initializing duration.
+ */
+ private volatile boolean isInitializing;
Review Comment:
Maybe `isRestoring`? I am not sure which is better name but it seems that we have run now two clear steps 'restoring' and 'invoking'('running') so maybe it is better just to have a flag 'restoring' for 'restoring' step.
--
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] RocMarshal commented on a diff in pull request #21999: [FLINK-29816][streaming] Fix the bug that StreamTask doesn't handle exception during restoring
Posted by "RocMarshal (via GitHub)" <gi...@apache.org>.
RocMarshal commented on code in PR #21999:
URL: https://github.com/apache/flink/pull/21999#discussion_r1115570034
##########
flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskITCase.java:
##########
@@ -110,6 +116,42 @@ public void testFailInEndOfConstructor() throws Exception {
.build(EXECUTOR_RESOURCE.getExecutor()));
}
+ @Test
+ public void testAsyncExceptionCanBeHandledDuringRestoring() throws Exception {
Review Comment:
@akalash Thanks for the comments. I fine to it & update it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] RocMarshal commented on pull request #21999: [FLINK-29816][streaming] Fix the bug that StreamTask doesn't handle exception during restoring
Posted by "RocMarshal (via GitHub)" <gi...@apache.org>.
RocMarshal commented on PR #21999:
URL: https://github.com/apache/flink/pull/21999#issuecomment-1441965644
> What do you mean by `hotfix-ticket`? For the current case, we do not need another jira ticket or another PR. In my opinion, this PR will eventually have two commits in the following order:
>
> 1.[hotfix] Migrate StreamTaskTest to Junit5 and Assertj
>
> 2.[[FLINK-29816](https://issues.apache.org/jira/browse/FLINK-29816)][streaming] Fix the bug that StreamTask xxxx
Thanks for the clarify, I got it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] RocMarshal commented on a diff in pull request #21999: [FLINK-29816][streaming] Fix the bug that StreamTask doesn't handle exception during restoring
Posted by "RocMarshal (via GitHub)" <gi...@apache.org>.
RocMarshal commented on code in PR #21999:
URL: https://github.com/apache/flink/pull/21999#discussion_r1115489770
##########
flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskITCase.java:
##########
@@ -110,6 +118,42 @@ public void testFailInEndOfConstructor() throws Exception {
.build(EXECUTOR_RESOURCE.getExecutor()));
}
+ @Test
+ public void streamTaskAsyncExceptionHandlerHandleExceptionInInitializingDuration()
+ throws Exception {
+ MockEnvironment mockEnvironment = MockEnvironment.builder().build();
+ RuntimeException expectedException = new RuntimeException("RUNTIME EXCEPTION");
+
+ mockEnvironment.setExpectedExternalFailureCause(AsynchronousException.class);
+ final String expectedErrorMessage = "EXPECTED_ERROR MESSAGE";
+
+ NoOpStreamTask noOpStreamTask =
+ new NoOpStreamTask(mockEnvironment) {
+ @Override
+ protected void init() throws Exception {
+ super.init();
+ CompletableFuture.runAsync(
+ () -> {
+ assertFalse(this.isRunning());
+ assertTrue(this.isInitializing());
+ this.handleAsyncException(
+ expectedErrorMessage, expectedException);
+ });
+ }
Review Comment:
> Flink test doesn't recommend using the `sleep`, you can use the `CompletableFuture future = new CompletableFuture()`.
Thank you for the reminding. It graced the lines
--
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] RocMarshal commented on a diff in pull request #21999: [FLINK-29816][streaming] Fix SteamTask skip exception in userfunction of ProcessWindowFunction during Steam.restore
Posted by "RocMarshal (via GitHub)" <gi...@apache.org>.
RocMarshal commented on code in PR #21999:
URL: https://github.com/apache/flink/pull/21999#discussion_r1114510855
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java:
##########
@@ -261,6 +261,12 @@
*/
private volatile boolean isRunning;
+ /**
+ * Flag to mark the task "in operation", in which case check needs to be process the exceptions
+ * thrown in the initializing duration.
+ */
+ private volatile boolean isInitializing;
Review Comment:
> Ah, I guess he just want to follow the `ExecutionState`'s initializing state, the corresponding java doc indicates that:
@akalash @reswqa yes. I described the section at the jira ticket. CC @1996fanrui
--
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] reswqa commented on a diff in pull request #21999: [FLINK-29816][streaming] Fix the bug that StreamTask doesn't handle exception during restoring
Posted by "reswqa (via GitHub)" <gi...@apache.org>.
reswqa commented on code in PR #21999:
URL: https://github.com/apache/flink/pull/21999#discussion_r1115754252
##########
flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTest.java:
##########
@@ -410,6 +410,42 @@ public CloseException() {
}
}
+ @Test
+ public void testAsyncExceptionCanBeHandledDuringRestoring() throws Exception {
Review Comment:
```suggestion
public void testHandleAsyncExceptionDuringRestoring() throws Exception {
```
##########
flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTest.java:
##########
@@ -410,6 +410,42 @@ public CloseException() {
}
}
+ @Test
+ public void testAsyncExceptionCanBeHandledDuringRestoring() throws Exception {
+ MockEnvironment mockEnvironment = MockEnvironment.builder().build();
+ RuntimeException expectedException = new RuntimeException("RUNTIME EXCEPTION");
+
+ mockEnvironment.setExpectedExternalFailureCause(AsynchronousException.class);
+ final String expectedErrorMessage = "EXPECTED_ERROR MESSAGE";
+
+ StreamTaskITCase.NoOpStreamTask noOpStreamTask =
Review Comment:
This is indeed not no-op, maybe should call it `initThrowExceptionTask` or other more appropriate name
##########
flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTest.java:
##########
@@ -410,6 +410,42 @@ public CloseException() {
}
}
+ @Test
+ public void testAsyncExceptionCanBeHandledDuringRestoring() throws Exception {
+ MockEnvironment mockEnvironment = MockEnvironment.builder().build();
+ RuntimeException expectedException = new RuntimeException("RUNTIME EXCEPTION");
Review Comment:
```suggestion
Throwable expectedException = new RuntimeException("RUNTIME 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] RocMarshal commented on a diff in pull request #21999: [FLINK-29816][streaming] Fix the bug that SteamTask skip handle exception during restoring.
Posted by "RocMarshal (via GitHub)" <gi...@apache.org>.
RocMarshal commented on code in PR #21999:
URL: https://github.com/apache/flink/pull/21999#discussion_r1115242031
##########
flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskITCase.java:
##########
@@ -110,6 +118,42 @@ public void testFailInEndOfConstructor() throws Exception {
.build(EXECUTOR_RESOURCE.getExecutor()));
}
+ @Test
+ public void streamTaskAsyncExceptionHandlerHandleExceptionInInitializingDuration()
+ throws Exception {
+ MockEnvironment mockEnvironment = MockEnvironment.builder().build();
+ RuntimeException expectedException = new RuntimeException("RUNTIME EXCEPTION");
+
+ mockEnvironment.setExpectedExternalFailureCause(AsynchronousException.class);
+ final String expectedErrorMessage = "EXPECTED_ERROR MESSAGE";
+
+ NoOpStreamTask noOpStreamTask =
+ new NoOpStreamTask(mockEnvironment) {
+ @Override
+ protected void init() throws Exception {
+ super.init();
+ CompletableFuture.runAsync(
+ () -> {
+ assertFalse(this.isRunning());
+ assertTrue(this.isInitializing());
+ this.handleAsyncException(
+ expectedErrorMessage, expectedException);
+ });
+ }
Review Comment:
Should we make the main thread of NoOpStreamTask sleep a small duration
or add a flag to ensure that The `runAsync` operation must run in the span (isRestoring = true, isRunning=false) ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] 1996fanrui commented on a diff in pull request #21999: [FLINK-29816][streaming] Fix the bug that StreamTask doesn't handle exception during restoring
Posted by "1996fanrui (via GitHub)" <gi...@apache.org>.
1996fanrui commented on code in PR #21999:
URL: https://github.com/apache/flink/pull/21999#discussion_r1116696715
##########
flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTest.java:
##########
@@ -362,30 +358,35 @@ private void testSyncSavepointWithEndInput(
"savepointResult");
harness.processAll();
- Assert.assertEquals(expectEndInput, TestBoundedOneInputStreamOperator.isInputEnded());
+ assertThat(TestBoundedOneInputStreamOperator.isInputEnded()).isEqualTo(expectEndInput);
}
@Test
- public void testCleanUpExceptionSuppressing() throws Exception {
+ void testCleanUpExceptionSuppressing() throws Exception {
try (StreamTaskMailboxTestHarness<String> testHarness =
new StreamTaskMailboxTestHarnessBuilder<>(OneInputStreamTask::new, STRING_TYPE_INFO)
.addInput(STRING_TYPE_INFO)
.setupOutputForSingletonOperatorChain(new FailingTwiceOperator())
.build()) {
- try {
- testHarness.processElement(new StreamRecord<>("Doesn't matter", 0));
- throw new RuntimeException("Expected an exception but ran successfully");
- } catch (Exception ex) {
- ExceptionUtils.assertThrowable(ex, ExpectedTestException.class);
- }
+ assertThatThrownBy(
+ () ->
+ testHarness.processElement(
+ new StreamRecord<>("Doesn't matter", 0)))
+ .satisfies(
+ (Consumer<Throwable>)
+ throwable ->
+ ExceptionUtils.assertThrowable(
+ throwable, ExpectedTestException.class));
- try {
- testHarness.finishProcessing();
- } catch (Exception ex) {
- // todo: checking for suppression if there are more exceptions during cleanup
- ExceptionUtils.assertThrowable(ex, FailingTwiceOperator.CloseException.class);
- }
+ // todo: checking for suppression if there are more exceptions during cleanup
+ assertThatThrownBy(testHarness::finishProcessing)
+ .satisfies(
+ (ThrowingConsumer<Throwable>)
+ throwable ->
+ ExceptionUtils.assertThrowable(
+ throwable,
+ FailingTwiceOperator.CloseException.class));
Review Comment:
```suggestion
.hasCause(FailingTwiceOperator.CloseException.class));
```
Can it work?
--
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] RocMarshal commented on a diff in pull request #21999: [FLINK-29816][streaming] Fix the bug that StreamTask doesn't handle exception during restoring
Posted by "RocMarshal (via GitHub)" <gi...@apache.org>.
RocMarshal commented on code in PR #21999:
URL: https://github.com/apache/flink/pull/21999#discussion_r1117058961
##########
flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTest.java:
##########
@@ -1103,35 +1143,41 @@ private void testFailToConfirmCheckpointMessage(Consumer<StreamTask<?, ?>> consu
* finished.
*/
@Test
- public void testCheckpointFailueOnClosedOperator() throws Throwable {
+ void testCheckpointFailueOnClosedOperator() throws Exception {
ClosingOperator<Integer> operator = new ClosingOperator<>();
StreamTaskMailboxTestHarnessBuilder<Integer> builder =
new StreamTaskMailboxTestHarnessBuilder<>(
OneInputStreamTask::new, BasicTypeInfo.INT_TYPE_INFO)
.addInput(BasicTypeInfo.INT_TYPE_INFO);
try (StreamTaskMailboxTestHarness<Integer> harness =
builder.setupOutputForSingletonOperatorChain(operator).build()) {
- // keeps the mailbox from suspending
- harness.setAutoProcess(false);
- harness.processElement(new StreamRecord<>(1));
-
- harness.streamTask.operatorChain.finishOperators(
- harness.streamTask.getActionExecutor(), StopMode.DRAIN);
- harness.streamTask.operatorChain.closeAllOperators();
- assertTrue(ClosingOperator.closed.get());
-
- harness.streamTask.triggerCheckpointOnBarrier(
- new CheckpointMetaData(1, 0),
- CheckpointOptions.forCheckpointWithDefaultLocation(),
- new CheckpointMetricsBuilder());
- } catch (Exception ex) {
- ExceptionUtils.assertThrowableWithMessage(
- ex, "OperatorChain and Task should never be closed at this point");
+ assertThatThrownBy(
+ () -> {
+ // keeps the mailbox from suspending
+ harness.setAutoProcess(false);
+ harness.processElement(new StreamRecord<>(1));
+
+ harness.streamTask.operatorChain.finishOperators(
+ harness.streamTask.getActionExecutor(), StopMode.DRAIN);
+ harness.streamTask.operatorChain.closeAllOperators();
+ assertThat(ClosingOperator.closed.get()).isTrue();
+
+ harness.streamTask.triggerCheckpointOnBarrier(
+ new CheckpointMetaData(1, 0),
+ CheckpointOptions.forCheckpointWithDefaultLocation(),
+ new CheckpointMetricsBuilder());
+ })
+ .satisfies(
+ (Consumer<Throwable>)
+ throwable ->
+ ExceptionUtils.assertThrowableWithMessage(
+ throwable,
+ "OperatorChain and Task should never be closed at this point"));
Review Comment:
I can understand what you mean. I have a try on it with failed result, So, what about keeping the current lines, please let me know what's your opinion.
<img width="1106" alt="image" src="https://user-images.githubusercontent.com/64569824/221199128-34ed3b53-1106-4b1f-b77a-5e7124fbd20c.png">
--
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] RocMarshal commented on a diff in pull request #21999: [FLINK-29816][streaming] Fix the bug that StreamTask doesn't handle exception during restoring
Posted by "RocMarshal (via GitHub)" <gi...@apache.org>.
RocMarshal commented on code in PR #21999:
URL: https://github.com/apache/flink/pull/21999#discussion_r1117840988
##########
flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTest.java:
##########
@@ -362,30 +358,35 @@ private void testSyncSavepointWithEndInput(
"savepointResult");
harness.processAll();
- Assert.assertEquals(expectEndInput, TestBoundedOneInputStreamOperator.isInputEnded());
+ assertThat(TestBoundedOneInputStreamOperator.isInputEnded()).isEqualTo(expectEndInput);
}
@Test
- public void testCleanUpExceptionSuppressing() throws Exception {
+ void testCleanUpExceptionSuppressing() throws Exception {
try (StreamTaskMailboxTestHarness<String> testHarness =
new StreamTaskMailboxTestHarnessBuilder<>(OneInputStreamTask::new, STRING_TYPE_INFO)
.addInput(STRING_TYPE_INFO)
.setupOutputForSingletonOperatorChain(new FailingTwiceOperator())
.build()) {
- try {
- testHarness.processElement(new StreamRecord<>("Doesn't matter", 0));
- throw new RuntimeException("Expected an exception but ran successfully");
- } catch (Exception ex) {
- ExceptionUtils.assertThrowable(ex, ExpectedTestException.class);
- }
+ assertThatThrownBy(
+ () ->
+ testHarness.processElement(
+ new StreamRecord<>("Doesn't matter", 0)))
+ .satisfies(
+ (Consumer<Throwable>)
+ throwable ->
+ ExceptionUtils.assertThrowable(
+ throwable, ExpectedTestException.class));
- try {
- testHarness.finishProcessing();
- } catch (Exception ex) {
- // todo: checking for suppression if there are more exceptions during cleanup
- ExceptionUtils.assertThrowable(ex, FailingTwiceOperator.CloseException.class);
- }
+ // todo: checking for suppression if there are more exceptions during cleanup
+ assertThatThrownBy(testHarness::finishProcessing)
+ .satisfies(
+ (ThrowingConsumer<Throwable>)
+ throwable ->
+ ExceptionUtils.assertThrowable(
+ throwable,
+ FailingTwiceOperator.CloseException.class));
Review Comment:
Yes, Thanks for reply. the line is also my original migration line.
@1996fanrui
@reswqa described the reason why we still need the `ExceptionUtils.assertThrowable` at https://github.com/apache/flink/pull/21999#discussion_r1116493534 due to the depth of the 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] reswqa commented on a diff in pull request #21999: [FLINK-29816][streaming] Fix the bug that StreamTask doesn't handle exception during restoring
Posted by "reswqa (via GitHub)" <gi...@apache.org>.
reswqa commented on code in PR #21999:
URL: https://github.com/apache/flink/pull/21999#discussion_r1117930702
##########
flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTest.java:
##########
@@ -362,30 +358,35 @@ private void testSyncSavepointWithEndInput(
"savepointResult");
harness.processAll();
- Assert.assertEquals(expectEndInput, TestBoundedOneInputStreamOperator.isInputEnded());
+ assertThat(TestBoundedOneInputStreamOperator.isInputEnded()).isEqualTo(expectEndInput);
}
@Test
- public void testCleanUpExceptionSuppressing() throws Exception {
+ void testCleanUpExceptionSuppressing() throws Exception {
try (StreamTaskMailboxTestHarness<String> testHarness =
new StreamTaskMailboxTestHarnessBuilder<>(OneInputStreamTask::new, STRING_TYPE_INFO)
.addInput(STRING_TYPE_INFO)
.setupOutputForSingletonOperatorChain(new FailingTwiceOperator())
.build()) {
- try {
- testHarness.processElement(new StreamRecord<>("Doesn't matter", 0));
- throw new RuntimeException("Expected an exception but ran successfully");
- } catch (Exception ex) {
- ExceptionUtils.assertThrowable(ex, ExpectedTestException.class);
- }
+ assertThatThrownBy(
+ () ->
+ testHarness.processElement(
+ new StreamRecord<>("Doesn't matter", 0)))
+ .satisfies(
+ (Consumer<Throwable>)
+ throwable ->
+ ExceptionUtils.assertThrowable(
+ throwable, ExpectedTestException.class));
- try {
- testHarness.finishProcessing();
- } catch (Exception ex) {
- // todo: checking for suppression if there are more exceptions during cleanup
- ExceptionUtils.assertThrowable(ex, FailingTwiceOperator.CloseException.class);
- }
+ // todo: checking for suppression if there are more exceptions during cleanup
+ assertThatThrownBy(testHarness::finishProcessing)
+ .satisfies(
+ (ThrowingConsumer<Throwable>)
+ throwable ->
+ ExceptionUtils.assertThrowable(
+ throwable,
+ FailingTwiceOperator.CloseException.class));
Review Comment:
Good suggestion @1996fanrui, Make sense to me.
--
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] RocMarshal commented on a diff in pull request #21999: [FLINK-29816][streaming] Fix the bug that StreamTask doesn't handle exception during restoring
Posted by "RocMarshal (via GitHub)" <gi...@apache.org>.
RocMarshal commented on code in PR #21999:
URL: https://github.com/apache/flink/pull/21999#discussion_r1117066514
##########
flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTest.java:
##########
@@ -1103,35 +1143,41 @@ private void testFailToConfirmCheckpointMessage(Consumer<StreamTask<?, ?>> consu
* finished.
*/
@Test
- public void testCheckpointFailueOnClosedOperator() throws Throwable {
+ void testCheckpointFailueOnClosedOperator() throws Exception {
ClosingOperator<Integer> operator = new ClosingOperator<>();
StreamTaskMailboxTestHarnessBuilder<Integer> builder =
new StreamTaskMailboxTestHarnessBuilder<>(
OneInputStreamTask::new, BasicTypeInfo.INT_TYPE_INFO)
.addInput(BasicTypeInfo.INT_TYPE_INFO);
try (StreamTaskMailboxTestHarness<Integer> harness =
builder.setupOutputForSingletonOperatorChain(operator).build()) {
- // keeps the mailbox from suspending
- harness.setAutoProcess(false);
- harness.processElement(new StreamRecord<>(1));
-
- harness.streamTask.operatorChain.finishOperators(
- harness.streamTask.getActionExecutor(), StopMode.DRAIN);
- harness.streamTask.operatorChain.closeAllOperators();
- assertTrue(ClosingOperator.closed.get());
-
- harness.streamTask.triggerCheckpointOnBarrier(
- new CheckpointMetaData(1, 0),
- CheckpointOptions.forCheckpointWithDefaultLocation(),
- new CheckpointMetricsBuilder());
- } catch (Exception ex) {
- ExceptionUtils.assertThrowableWithMessage(
- ex, "OperatorChain and Task should never be closed at this point");
+ assertThatThrownBy(
+ () -> {
+ // keeps the mailbox from suspending
+ harness.setAutoProcess(false);
+ harness.processElement(new StreamRecord<>(1));
+
+ harness.streamTask.operatorChain.finishOperators(
+ harness.streamTask.getActionExecutor(), StopMode.DRAIN);
+ harness.streamTask.operatorChain.closeAllOperators();
+ assertThat(ClosingOperator.closed.get()).isTrue();
Review Comment:
Yes, the small scope could help devs to locate the lines with expected exception. I'd like it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] 1996fanrui commented on a diff in pull request #21999: [FLINK-29816][streaming] Fix the bug that StreamTask doesn't handle exception during restoring
Posted by "1996fanrui (via GitHub)" <gi...@apache.org>.
1996fanrui commented on code in PR #21999:
URL: https://github.com/apache/flink/pull/21999#discussion_r1117222444
##########
flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTest.java:
##########
@@ -362,30 +358,35 @@ private void testSyncSavepointWithEndInput(
"savepointResult");
harness.processAll();
- Assert.assertEquals(expectEndInput, TestBoundedOneInputStreamOperator.isInputEnded());
+ assertThat(TestBoundedOneInputStreamOperator.isInputEnded()).isEqualTo(expectEndInput);
}
@Test
- public void testCleanUpExceptionSuppressing() throws Exception {
+ void testCleanUpExceptionSuppressing() throws Exception {
try (StreamTaskMailboxTestHarness<String> testHarness =
new StreamTaskMailboxTestHarnessBuilder<>(OneInputStreamTask::new, STRING_TYPE_INFO)
.addInput(STRING_TYPE_INFO)
.setupOutputForSingletonOperatorChain(new FailingTwiceOperator())
.build()) {
- try {
- testHarness.processElement(new StreamRecord<>("Doesn't matter", 0));
- throw new RuntimeException("Expected an exception but ran successfully");
- } catch (Exception ex) {
- ExceptionUtils.assertThrowable(ex, ExpectedTestException.class);
- }
+ assertThatThrownBy(
+ () ->
+ testHarness.processElement(
+ new StreamRecord<>("Doesn't matter", 0)))
+ .satisfies(
+ (Consumer<Throwable>)
+ throwable ->
+ ExceptionUtils.assertThrowable(
+ throwable, ExpectedTestException.class));
Review Comment:
```suggestion
.isInstanceOf(ExpectedTestException.class);
```
--
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] reswqa commented on pull request #21999: [FLINK-29816][streaming] Fix the bug that StreamTask doesn't handle exception during restoring
Posted by "reswqa (via GitHub)" <gi...@apache.org>.
reswqa commented on PR #21999:
URL: https://github.com/apache/flink/pull/21999#issuecomment-1441958963
What do you mean by `hotfix-ticket`? For the current case, we do not need another jira ticket or another PR.
In my opinion, this PR will eventually have two commitments:
1.[hotfix] Migrate StreamTaskTest to Junit5 and Assertj
2.[FLINK-29816][streaming] Fix the bug that StreamTask xxxx
--
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] 1996fanrui commented on a diff in pull request #21999: [FLINK-29816][streaming] Fix SteamTask skip exception in userfunction of ProcessWindowFunction during Steam.restore
Posted by "1996fanrui (via GitHub)" <gi...@apache.org>.
1996fanrui commented on code in PR #21999:
URL: https://github.com/apache/flink/pull/21999#discussion_r1114484862
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java:
##########
@@ -689,6 +695,7 @@ void restoreInternal() throws Exception {
LOG.debug("Re-restore attempt rejected.");
return;
}
+ isInitializing = true;
Review Comment:
I guess he missed some code during migration😂
--
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] akalash commented on a diff in pull request #21999: [FLINK-29816][streaming] Fix the bug that StreamTask doesn't handle exception during restoring
Posted by "akalash (via GitHub)" <gi...@apache.org>.
akalash commented on code in PR #21999:
URL: https://github.com/apache/flink/pull/21999#discussion_r1115524445
##########
flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskITCase.java:
##########
@@ -110,6 +116,42 @@ public void testFailInEndOfConstructor() throws Exception {
.build(EXECUTOR_RESOURCE.getExecutor()));
}
+ @Test
+ public void testAsyncExceptionCanBeHandledDuringRestoring() throws Exception {
Review Comment:
I don't think that it is the right place for this test. Your test is just a simple unit test while this class is for IT tests. So I think `StreamTaskTest` is a better place for this test.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] reswqa commented on a diff in pull request #21999: [FLINK-29816][streaming] Fix the bug that StreamTask doesn't handle exception during restoring
Posted by "reswqa (via GitHub)" <gi...@apache.org>.
reswqa commented on code in PR #21999:
URL: https://github.com/apache/flink/pull/21999#discussion_r1116563643
##########
flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTest.java:
##########
@@ -368,17 +368,19 @@ void testCleanUpExceptionSuppressing() throws Exception {
.setupOutputForSingletonOperatorChain(new FailingTwiceOperator())
.build()) {
- assertThatThrownBy(
- () -> {
- testHarness.processElement(new StreamRecord<>("Doesn't matter", 0));
- throw new RuntimeException(
- "Expected an exception but ran successfully");
- })
- .isInstanceOf(ExpectedTestException.class);
+ try {
+ testHarness.processElement(new StreamRecord<>("Doesn't matter", 0));
+ throw new RuntimeException("Expected an exception but ran successfully");
+ } catch (Exception ex) {
+ ExceptionUtils.assertThrowable(ex, ExpectedTestException.class);
+ }
Review Comment:
```suggestion
Assertions.assertThatThrownBy(
() ->
testHarness.processElement(
new StreamRecord<>("Doesn't matter", 0)))
.satisfies(
throwable ->
ExceptionUtils.assertThrowable(
throwable, ExpectedTestException.class));
```
--
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] RocMarshal commented on a diff in pull request #21999: [FLINK-29816][streaming] Fix the bug that SteamTask skip handle exception during restoring.
Posted by "RocMarshal (via GitHub)" <gi...@apache.org>.
RocMarshal commented on code in PR #21999:
URL: https://github.com/apache/flink/pull/21999#discussion_r1114607304
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java:
##########
@@ -261,6 +261,12 @@
*/
private volatile boolean isRunning;
+ /**
+ * Flag to mark the task "in operation", in which case check needs to be process the exceptions
+ * thrown in the initializing duration.
+ */
+ private volatile boolean isInitializing;
Review Comment:
Thanks for the clarify. I got it.
We need to define or describe different states for different abstract level as accurately as possible, right?
I'd prefer your description `isRestoring`, 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] reswqa commented on a diff in pull request #21999: [FLINK-29816][streaming] Fix the bug that StreamTask doesn't handle exception during restoring
Posted by "reswqa (via GitHub)" <gi...@apache.org>.
reswqa commented on code in PR #21999:
URL: https://github.com/apache/flink/pull/21999#discussion_r1115766066
##########
flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTest.java:
##########
@@ -410,6 +410,42 @@ public CloseException() {
}
}
+ @Test
+ public void testAsyncExceptionCanBeHandledDuringRestoring() throws Exception {
Review Comment:
This is just my personal preference, and the current name is also acceptable.
--
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] akalash commented on pull request #21999: [FLINK-29816][streaming] Fix SteamTask skip exception in userfunction of ProcessWindowFunction during Steam.restore
Posted by "akalash (via GitHub)" <gi...@apache.org>.
akalash commented on PR #21999:
URL: https://github.com/apache/flink/pull/21999#issuecomment-1440106317
@1996fanrui, how urgent is this PR ? Can we wait for merging #21923?(I think it helps you avoid a new flag)
--
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] RocMarshal commented on pull request #21999: [FLINK-29816][streaming] Fix the bug that StreamTask doesn't handle exception during restoring
Posted by "RocMarshal (via GitHub)" <gi...@apache.org>.
RocMarshal commented on PR #21999:
URL: https://github.com/apache/flink/pull/21999#issuecomment-1446413820
Hi, @reswqa I drive the BP patches in https://github.com/apache/flink/pull/22035 & https://github.com/apache/flink/pull/22036
PTAL if you're in free time. Thank you~.
CC @akalash @1996fanrui
--
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] RocMarshal commented on a diff in pull request #21999: [FLINK-29816][streaming] Fix the bug that StreamTask doesn't handle exception during restoring
Posted by "RocMarshal (via GitHub)" <gi...@apache.org>.
RocMarshal commented on code in PR #21999:
URL: https://github.com/apache/flink/pull/21999#discussion_r1116991198
##########
flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTest.java:
##########
@@ -945,12 +982,15 @@ public void testAsyncCheckpointingConcurrentCloseBeforeAcknowledge() throws Exce
// make sure that all state handles have been discarded
discardFuture.get();
- try {
- mockEnvironment.getAcknowledgeCheckpointFuture().get(10L, TimeUnit.MILLISECONDS);
- fail("The checkpoint should not get acknowledged.");
- } catch (TimeoutException expected) {
- // future should not be completed
- }
+ assertThatThrownBy(
+ () -> {
+ // future should not be completed
+ mockEnvironment
+ .getAcknowledgeCheckpointFuture()
+ .get(10L, TimeUnit.MILLISECONDS);
+ fail("The checkpoint should not get acknowledged.");
Review Comment:
Yes. I'll check the same cases in the test.
thx~
--
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] 1996fanrui commented on a diff in pull request #21999: [FLINK-29816][streaming] Fix the bug that StreamTask doesn't handle exception during restoring
Posted by "1996fanrui (via GitHub)" <gi...@apache.org>.
1996fanrui commented on code in PR #21999:
URL: https://github.com/apache/flink/pull/21999#discussion_r1117218805
##########
flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTest.java:
##########
@@ -362,30 +358,35 @@ private void testSyncSavepointWithEndInput(
"savepointResult");
harness.processAll();
- Assert.assertEquals(expectEndInput, TestBoundedOneInputStreamOperator.isInputEnded());
+ assertThat(TestBoundedOneInputStreamOperator.isInputEnded()).isEqualTo(expectEndInput);
}
@Test
- public void testCleanUpExceptionSuppressing() throws Exception {
+ void testCleanUpExceptionSuppressing() throws Exception {
try (StreamTaskMailboxTestHarness<String> testHarness =
new StreamTaskMailboxTestHarnessBuilder<>(OneInputStreamTask::new, STRING_TYPE_INFO)
.addInput(STRING_TYPE_INFO)
.setupOutputForSingletonOperatorChain(new FailingTwiceOperator())
.build()) {
- try {
- testHarness.processElement(new StreamRecord<>("Doesn't matter", 0));
- throw new RuntimeException("Expected an exception but ran successfully");
- } catch (Exception ex) {
- ExceptionUtils.assertThrowable(ex, ExpectedTestException.class);
- }
+ assertThatThrownBy(
+ () ->
+ testHarness.processElement(
+ new StreamRecord<>("Doesn't matter", 0)))
+ .satisfies(
+ (Consumer<Throwable>)
+ throwable ->
+ ExceptionUtils.assertThrowable(
+ throwable, ExpectedTestException.class));
- try {
- testHarness.finishProcessing();
- } catch (Exception ex) {
- // todo: checking for suppression if there are more exceptions during cleanup
- ExceptionUtils.assertThrowable(ex, FailingTwiceOperator.CloseException.class);
- }
+ // todo: checking for suppression if there are more exceptions during cleanup
+ assertThatThrownBy(testHarness::finishProcessing)
+ .satisfies(
+ (ThrowingConsumer<Throwable>)
+ throwable ->
+ ExceptionUtils.assertThrowable(
+ throwable,
+ FailingTwiceOperator.CloseException.class));
Review Comment:
```suggestion
.isInstanceOf(FailingTwiceOperator.CloseException.class);
```
I have tried it, it works.
--
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] 1996fanrui commented on pull request #21999: [FLINK-29816][streaming] Fix SteamTask skip exception in userfunction of ProcessWindowFunction during Steam.restore
Posted by "1996fanrui (via GitHub)" <gi...@apache.org>.
1996fanrui commented on PR #21999:
URL: https://github.com/apache/flink/pull/21999#issuecomment-1440165179
Hi @akalash , thanks for your quick feedback.
> how urgent is this PR ?
I think it isn't very urgent.
> Can we wait for merging #21923?(I think it helps you avoid a new flag)
Yeah, #21923 is very useful for maintainability. However, I recommend fixing the bug before refactoring the state machine for two reasons:
1. After the introduction of initializing, the state of TaskState will become more complicated, and FLINK-13871 will do more thought when designing TaskState. If we design TaskState first, not sure if it will be easily compatible with initializing in the future.
2. Fix bugs before refactoring the state machine, it will be easier to backport the bugfix to 1.16 and 1.17.
Anyway, I don't have strong opinion about which PR is merged first. If #21923 is merged first, we need to consider the scalability of TaskState.
--
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] RocMarshal commented on a diff in pull request #21999: [FLINK-29816][streaming] Fix SteamTask skip exception in userfunction of ProcessWindowFunction during Steam.restore
Posted by "RocMarshal (via GitHub)" <gi...@apache.org>.
RocMarshal commented on code in PR #21999:
URL: https://github.com/apache/flink/pull/21999#discussion_r1114503782
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java:
##########
@@ -689,6 +695,7 @@ void restoreInternal() throws Exception {
LOG.debug("Re-restore attempt rejected.");
return;
}
+ isInitializing = true;
Review Comment:
Thanks for your quick reply.
I add the `false-set` for `isInitializing`.
--
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] 1996fanrui commented on a diff in pull request #21999: [FLINK-29816][streaming] Fix the bug that SteamTask skip handle exception during restoring.
Posted by "1996fanrui (via GitHub)" <gi...@apache.org>.
1996fanrui commented on code in PR #21999:
URL: https://github.com/apache/flink/pull/21999#discussion_r1115424478
##########
flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskITCase.java:
##########
@@ -110,6 +118,47 @@ public void testFailInEndOfConstructor() throws Exception {
.build(EXECUTOR_RESOURCE.getExecutor()));
}
+ @Test
+ public void streamTaskAsyncExceptionHandlerHandleExceptionInInitializingDuration()
Review Comment:
There are some issues about the method name:
1. Test method name should start with `test`.
2. It's too long, and the Class Name is about `StreamTask`, so I think the `StreamTask` can be removed for the method name.
3. Rename the `Initializing` to `restoring`
How about `testAsyncExceptionCanBeHandledDuringRestoring`?
##########
flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskITCase.java:
##########
@@ -110,6 +118,47 @@ public void testFailInEndOfConstructor() throws Exception {
.build(EXECUTOR_RESOURCE.getExecutor()));
}
+ @Test
+ public void streamTaskAsyncExceptionHandlerHandleExceptionInInitializingDuration()
+ throws Exception {
+ MockEnvironment mockEnvironment = MockEnvironment.builder().build();
+ RuntimeException expectedException = new RuntimeException("RUNTIME EXCEPTION");
+
+ mockEnvironment.setExpectedExternalFailureCause(AsynchronousException.class);
+ final String expectedErrorMessage = "EXPECTED_ERROR MESSAGE";
+
+ NoOpStreamTask noOpStreamTask =
+ new NoOpStreamTask(mockEnvironment) {
+ volatile boolean handledAsyncException = false;
+
+ @Override
+ protected void init() throws Exception {
+ super.init();
+
+ CompletableFuture.runAsync(
+ () -> {
+ this.handleAsyncException(
+ expectedErrorMessage, expectedException);
+ handledAsyncException = true;
+ });
+ while (!handledAsyncException) {
+ ThreadUtils.sleep(Duration.ofMillis(10));
+ }
+ }
+ };
+ noOpStreamTask.invoke();
Review Comment:
How about call the `noOpStreamTask.restore()` due to this test checks whether AsyncException can be handled during restoring?
Please correct me, if `noOpStreamTask.restore()` doesn't work here.
##########
flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskITCase.java:
##########
@@ -110,6 +118,47 @@ public void testFailInEndOfConstructor() throws Exception {
.build(EXECUTOR_RESOURCE.getExecutor()));
}
+ @Test
+ public void streamTaskAsyncExceptionHandlerHandleExceptionInInitializingDuration()
+ throws Exception {
+ MockEnvironment mockEnvironment = MockEnvironment.builder().build();
+ RuntimeException expectedException = new RuntimeException("RUNTIME EXCEPTION");
+
+ mockEnvironment.setExpectedExternalFailureCause(AsynchronousException.class);
+ final String expectedErrorMessage = "EXPECTED_ERROR MESSAGE";
+
+ NoOpStreamTask noOpStreamTask =
+ new NoOpStreamTask(mockEnvironment) {
+ volatile boolean handledAsyncException = false;
+
+ @Override
+ protected void init() throws Exception {
+ super.init();
+
+ CompletableFuture.runAsync(
Review Comment:
Add a comment : `Throw exception during restoring.`
--
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] akalash commented on a diff in pull request #21999: [FLINK-29816][streaming] Fix the bug that SteamTask skip handle exception during restoring.
Posted by "akalash (via GitHub)" <gi...@apache.org>.
akalash commented on code in PR #21999:
URL: https://github.com/apache/flink/pull/21999#discussion_r1114678942
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java:
##########
@@ -261,6 +261,12 @@
*/
private volatile boolean isRunning;
+ /**
+ * Flag to mark the task "in operation", in which case check needs to be process the exceptions
+ * thrown in the initializing duration.
+ */
+ private volatile boolean isInitializing;
Review Comment:
> We need to define or describe different states for different abstract level as accurately as possible, right?
Yes, I think it is pretty much what I meant. We have a class/entity with a pretty clear lifecycle let's use these rather than trying to adapt to something else.
But of course, I'm happy to hear opinions from everybody.
--
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] RocMarshal commented on a diff in pull request #21999: [FLINK-29816][streaming] Fix the bug that StreamTask doesn't handle exception during restoring
Posted by "RocMarshal (via GitHub)" <gi...@apache.org>.
RocMarshal commented on code in PR #21999:
URL: https://github.com/apache/flink/pull/21999#discussion_r1115487606
##########
flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskITCase.java:
##########
@@ -110,6 +118,47 @@ public void testFailInEndOfConstructor() throws Exception {
.build(EXECUTOR_RESOURCE.getExecutor()));
}
+ @Test
+ public void streamTaskAsyncExceptionHandlerHandleExceptionInInitializingDuration()
+ throws Exception {
+ MockEnvironment mockEnvironment = MockEnvironment.builder().build();
+ RuntimeException expectedException = new RuntimeException("RUNTIME EXCEPTION");
+
+ mockEnvironment.setExpectedExternalFailureCause(AsynchronousException.class);
+ final String expectedErrorMessage = "EXPECTED_ERROR MESSAGE";
+
+ NoOpStreamTask noOpStreamTask =
+ new NoOpStreamTask(mockEnvironment) {
+ volatile boolean handledAsyncException = false;
+
+ @Override
+ protected void init() throws Exception {
+ super.init();
+
+ CompletableFuture.runAsync(
Review Comment:
updated~.
--
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] RocMarshal commented on a diff in pull request #21999: [FLINK-29816][streaming] Fix the bug that StreamTask doesn't handle exception during restoring
Posted by "RocMarshal (via GitHub)" <gi...@apache.org>.
RocMarshal commented on code in PR #21999:
URL: https://github.com/apache/flink/pull/21999#discussion_r1115494416
##########
flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskITCase.java:
##########
@@ -110,6 +118,47 @@ public void testFailInEndOfConstructor() throws Exception {
.build(EXECUTOR_RESOURCE.getExecutor()));
}
+ @Test
+ public void streamTaskAsyncExceptionHandlerHandleExceptionInInitializingDuration()
Review Comment:
> How about `testAsyncExceptionCanBeHandledDuringRestoring`?
Thanks for the items, which is fine to improve the code style.
updated.
--
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] RocMarshal commented on a diff in pull request #21999: [FLINK-29816][streaming] Fix SteamTask skip exception in userfunction of ProcessWindowFunction during Steam.restore
Posted by "RocMarshal (via GitHub)" <gi...@apache.org>.
RocMarshal commented on code in PR #21999:
URL: https://github.com/apache/flink/pull/21999#discussion_r1114503782
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java:
##########
@@ -689,6 +695,7 @@ void restoreInternal() throws Exception {
LOG.debug("Re-restore attempt rejected.");
return;
}
+ isInitializing = true;
Review Comment:
So sorry for this & Thanks for your quick reply.
I indeed missed the line during migration & I added the `false-set` for `isInitializing`.
--
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] RocMarshal commented on pull request #21999: [FLINK-29816][streaming] Fix SteamTask skip exception in userfunction of ProcessWindowFunction during Steam.restore
Posted by "RocMarshal (via GitHub)" <gi...@apache.org>.
RocMarshal commented on PR #21999:
URL: https://github.com/apache/flink/pull/21999#issuecomment-1440348971
> The pr title and commit message isn't general. `ProcessWindowFunction` is just a special case here.
>
> How about this? `Fix the bug that SteamTask skip handle exception during restoring`
I'm fine to it. good proposal
--
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] akalash commented on a diff in pull request #21999: [FLINK-29816][streaming] Fix the bug that StreamTask doesn't handle exception during restoring
Posted by "akalash (via GitHub)" <gi...@apache.org>.
akalash commented on code in PR #21999:
URL: https://github.com/apache/flink/pull/21999#discussion_r1115732271
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java:
##########
@@ -1018,6 +1023,11 @@ public final boolean isRunning() {
return isRunning;
}
+ @VisibleForTesting
+ public final boolean isRestoring() {
Review Comment:
I actually don't see that this is used in tests. I think you can remove this method
--
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] reswqa commented on a diff in pull request #21999: [FLINK-29816][streaming] Fix the bug that StreamTask doesn't handle exception during restoring
Posted by "reswqa (via GitHub)" <gi...@apache.org>.
reswqa commented on code in PR #21999:
URL: https://github.com/apache/flink/pull/21999#discussion_r1116492529
##########
flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTest.java:
##########
@@ -362,30 +357,28 @@ private void testSyncSavepointWithEndInput(
"savepointResult");
harness.processAll();
- Assert.assertEquals(expectEndInput, TestBoundedOneInputStreamOperator.isInputEnded());
+ assertThat(TestBoundedOneInputStreamOperator.isInputEnded()).isEqualTo(expectEndInput);
}
@Test
- public void testCleanUpExceptionSuppressing() throws Exception {
+ void testCleanUpExceptionSuppressing() throws Exception {
try (StreamTaskMailboxTestHarness<String> testHarness =
new StreamTaskMailboxTestHarnessBuilder<>(OneInputStreamTask::new, STRING_TYPE_INFO)
.addInput(STRING_TYPE_INFO)
.setupOutputForSingletonOperatorChain(new FailingTwiceOperator())
.build()) {
- try {
- testHarness.processElement(new StreamRecord<>("Doesn't matter", 0));
- throw new RuntimeException("Expected an exception but ran successfully");
- } catch (Exception ex) {
- ExceptionUtils.assertThrowable(ex, ExpectedTestException.class);
- }
+ assertThatThrownBy(
+ () -> {
+ testHarness.processElement(new StreamRecord<>("Doesn't matter", 0));
+ throw new RuntimeException(
Review Comment:
Why throw exception here?
##########
flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTest.java:
##########
@@ -1085,27 +1081,27 @@ public void testNotifyCheckpointOnClosedOperator() throws Throwable {
harness.streamTask.notifyCheckpointCompleteAsync(1);
harness.streamTask.runMailboxStep();
- assertEquals(1, ClosingOperator.notified.get());
- assertFalse(ClosingOperator.closed.get());
+ assertThat(ClosingOperator.notified.get()).isOne();
Review Comment:
IIRC, AssertJ does have specific assertions for `AtomicXXX`, please avoid using `get`.
It is also necessary to check the similar problems of the whole test class.
##########
flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTest.java:
##########
@@ -1121,13 +1117,14 @@ private void testFailToConfirmCheckpointMessage(Consumer<StreamTask<?, ?>> consu
StreamTaskMailboxTestHarness<Integer> harness =
builder.setupOutputForSingletonOperatorChain(streamMap).build();
- try {
- consumer.accept(harness.streamTask);
- harness.streamTask.runMailboxLoop();
- fail();
- } catch (ExpectedTestException expected) {
- // expected exceptionestProcessWithUnAvailableInput
- }
+ // expected exceptionestProcessWithUnAvailableInput
+ assertThatThrownBy(
+ () -> {
+ consumer.accept(harness.streamTask);
+ harness.streamTask.runMailboxLoop();
+ fail(null);
Review Comment:
Why we need `fail`?
##########
flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTest.java:
##########
@@ -362,30 +357,28 @@ private void testSyncSavepointWithEndInput(
"savepointResult");
harness.processAll();
- Assert.assertEquals(expectEndInput, TestBoundedOneInputStreamOperator.isInputEnded());
+ assertThat(TestBoundedOneInputStreamOperator.isInputEnded()).isEqualTo(expectEndInput);
}
@Test
- public void testCleanUpExceptionSuppressing() throws Exception {
+ void testCleanUpExceptionSuppressing() throws Exception {
try (StreamTaskMailboxTestHarness<String> testHarness =
new StreamTaskMailboxTestHarnessBuilder<>(OneInputStreamTask::new, STRING_TYPE_INFO)
.addInput(STRING_TYPE_INFO)
.setupOutputForSingletonOperatorChain(new FailingTwiceOperator())
.build()) {
- try {
- testHarness.processElement(new StreamRecord<>("Doesn't matter", 0));
- throw new RuntimeException("Expected an exception but ran successfully");
- } catch (Exception ex) {
- ExceptionUtils.assertThrowable(ex, ExpectedTestException.class);
- }
+ assertThatThrownBy(
+ () -> {
+ testHarness.processElement(new StreamRecord<>("Doesn't matter", 0));
+ throw new RuntimeException(
+ "Expected an exception but ran successfully");
+ })
+ .isInstanceOf(ExpectedTestException.class);
- try {
- testHarness.finishProcessing();
- } catch (Exception ex) {
- // todo: checking for suppression if there are more exceptions during cleanup
- ExceptionUtils.assertThrowable(ex, FailingTwiceOperator.CloseException.class);
- }
+ // todo: checking for suppression if there are more exceptions during cleanup
+ assertThatThrownBy(testHarness::finishProcessing)
+ .isInstanceOf(FailingTwiceOperator.CloseException.class);
Review Comment:
Maybe we should check weather `ExceptionUtils.assertThrowable` will find throwable into more deeper level, but `isInstanceOf` indeed only check the throwable in the first level.
##########
flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTest.java:
##########
@@ -1139,35 +1136,38 @@ private void testFailToConfirmCheckpointMessage(Consumer<StreamTask<?, ?>> consu
* finished.
*/
@Test
- public void testCheckpointFailueOnClosedOperator() throws Throwable {
+ void testCheckpointFailueOnClosedOperator() throws Exception {
ClosingOperator<Integer> operator = new ClosingOperator<>();
StreamTaskMailboxTestHarnessBuilder<Integer> builder =
new StreamTaskMailboxTestHarnessBuilder<>(
OneInputStreamTask::new, BasicTypeInfo.INT_TYPE_INFO)
.addInput(BasicTypeInfo.INT_TYPE_INFO);
try (StreamTaskMailboxTestHarness<Integer> harness =
builder.setupOutputForSingletonOperatorChain(operator).build()) {
- // keeps the mailbox from suspending
- harness.setAutoProcess(false);
- harness.processElement(new StreamRecord<>(1));
-
- harness.streamTask.operatorChain.finishOperators(
- harness.streamTask.getActionExecutor(), StopMode.DRAIN);
- harness.streamTask.operatorChain.closeAllOperators();
- assertTrue(ClosingOperator.closed.get());
-
- harness.streamTask.triggerCheckpointOnBarrier(
- new CheckpointMetaData(1, 0),
- CheckpointOptions.forCheckpointWithDefaultLocation(),
- new CheckpointMetricsBuilder());
- } catch (Exception ex) {
- ExceptionUtils.assertThrowableWithMessage(
- ex, "OperatorChain and Task should never be closed at this point");
+ assertThatThrownBy(
+ () -> {
+ // keeps the mailbox from suspending
+ harness.setAutoProcess(false);
+ harness.processElement(new StreamRecord<>(1));
+
+ harness.streamTask.operatorChain.finishOperators(
+ harness.streamTask.getActionExecutor(), StopMode.DRAIN);
+ harness.streamTask.operatorChain.closeAllOperators();
+ assertThat(ClosingOperator.closed.get()).isTrue();
Review Comment:
ditto.
--
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