You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by se...@apache.org on 2020/10/13 09:29:47 UTC
[flink] branch master updated (232b61f -> 401f56f)
This is an automated email from the ASF dual-hosted git repository.
sewen pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git.
from 232b61f [FLINK-17857][e2e] Fix Kubernetes e2e tests on Mac OS
new 68c5c2f [FLINK-19448][connector base] Explicitly check for un-expected condition that would leave an inconsistent state
new d0e41a0 [tmpfix] Set CI log level to TRACE for 'org.apache.flink.connector.base.source.reader' to debug FLINK-19448
new 401f56f [FLINK-19427][FLINK-19489][tests] Fix test conditions for 'SplitFetcherTest.testNotifiesWhenGoingIdleConcurrent()'
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../base/source/reader/SourceReaderBase.java | 7 ++-
.../source/reader/fetcher/SplitFetcherTest.java | 54 +++++++++++-----------
tools/ci/log4j.properties | 5 ++
3 files changed, 36 insertions(+), 30 deletions(-)
[flink] 01/03: [FLINK-19448][connector base] Explicitly check for
un-expected condition that would leave an inconsistent state
Posted by se...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
sewen pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 68c5c2ff779d82a1ff81ffaf60d8a1b283797db1
Author: Stephan Ewen <se...@apache.org>
AuthorDate: Mon Oct 12 17:38:30 2020 +0200
[FLINK-19448][connector base] Explicitly check for un-expected condition that would leave an inconsistent state
This condition should never happen, but if it ever happened, it would leave the source in an idle state waiting for
more input data, rather than shutting down.
---
.../flink/connector/base/source/reader/SourceReaderBase.java | 7 +++++--
1 file changed, 5 insertions(+), 2 deletions(-)
diff --git a/flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/reader/SourceReaderBase.java b/flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/reader/SourceReaderBase.java
index 1182cae..e1da654 100644
--- a/flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/reader/SourceReaderBase.java
+++ b/flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/reader/SourceReaderBase.java
@@ -262,10 +262,13 @@ public abstract class SourceReaderBase<E, T, SplitT extends SourceSplit, SplitSt
private InputStatus finishedOrAvailableLater() {
final boolean allFetchersHaveShutdown = splitFetcherManager.maybeShutdownFinishedFetchers();
- if (noMoreSplitsAssignment && allFetchersHaveShutdown && elementsQueue.isEmpty()) {
+ if (!(noMoreSplitsAssignment && allFetchersHaveShutdown)) {
+ return InputStatus.NOTHING_AVAILABLE;
+ }
+ if (elementsQueue.isEmpty()) {
return InputStatus.END_OF_INPUT;
} else {
- return InputStatus.NOTHING_AVAILABLE;
+ throw new IllegalStateException("Called 'finishedOrAvailableLater()' with shut-down fetchers but non-empty queue");
}
}
[flink] 03/03: [FLINK-19427][FLINK-19489][tests] Fix test
conditions for 'SplitFetcherTest.testNotifiesWhenGoingIdleConcurrent()'
Posted by se...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
sewen pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 401f56fe9d6b0271260edf9787cdcbfe4d03874d
Author: Stephan Ewen <se...@apache.org>
AuthorDate: Mon Oct 12 13:43:57 2020 +0200
[FLINK-19427][FLINK-19489][tests] Fix test conditions for 'SplitFetcherTest.testNotifiesWhenGoingIdleConcurrent()'
The changed logic also fixes flaky thread shutdown logic as a side effect, because it no longer relies
on thread interrupting.
This closes #13593
---
.../source/reader/fetcher/SplitFetcherTest.java | 54 +++++++++++-----------
1 file changed, 26 insertions(+), 28 deletions(-)
diff --git a/flink-connectors/flink-connector-base/src/test/java/org/apache/flink/connector/base/source/reader/fetcher/SplitFetcherTest.java b/flink-connectors/flink-connector-base/src/test/java/org/apache/flink/connector/base/source/reader/fetcher/SplitFetcherTest.java
index 5027e3f..5082ebf 100644
--- a/flink-connectors/flink-connector-base/src/test/java/org/apache/flink/connector/base/source/reader/fetcher/SplitFetcherTest.java
+++ b/flink-connectors/flink-connector-base/src/test/java/org/apache/flink/connector/base/source/reader/fetcher/SplitFetcherTest.java
@@ -120,16 +120,16 @@ public class SplitFetcherTest {
final SplitFetcher<Object, TestingSourceSplit> fetcher = createFetcherWithSplit(
"test-split", queue, new TestingSplitReader<>(finishedSplitFetch("test-split")));
- final QueueDrainerThread queueDrainer = new QueueDrainerThread(queue);
+ final QueueDrainerThread queueDrainer = new QueueDrainerThread(queue, fetcher, 1);
queueDrainer.start();
- try {
- fetcher.runOnce();
+ fetcher.runOnce();
- assertTrue(queue.getAvailabilityFuture().isDone());
- } finally {
- queueDrainer.shutdown();
- }
+ queueDrainer.sync();
+
+ // either we got the notification that the fetcher went idle after the queue was drained (thread finished)
+ // or the fetcher was already idle when the thread drained the queue (then we need no additional notification)
+ assertTrue(queue.getAvailabilityFuture().isDone() || queueDrainer.wasIdleWhenFinished());
}
@Test
@@ -139,18 +139,15 @@ public class SplitFetcherTest {
final SplitFetcher<Object, TestingSourceSplit> fetcher = createFetcherWithSplit(
"test-split", queue, new TestingSplitReader<>(finishedSplitFetch("test-split")));
- final QueueDrainerThread queueDrainer = new QueueDrainerThread(queue);
+ final QueueDrainerThread queueDrainer = new QueueDrainerThread(queue, fetcher, 1);
queueDrainer.start();
final CompletableFuture<?> future = queue.getAvailabilityFuture();
- try {
- fetcher.runOnce();
+ fetcher.runOnce();
+ assertTrue(future.isDone());
- assertTrue(future.isDone());
- } finally {
- queueDrainer.shutdown();
- }
+ queueDrainer.sync();
}
@Test
@@ -274,31 +271,32 @@ public class SplitFetcherTest {
private static final class QueueDrainerThread extends CheckedThread {
private final FutureCompletingBlockingQueue<?> queue;
- private volatile boolean running = true;
+ private final SplitFetcher<?, ?> fetcher;
+ private final int numFetchesToTake;
- QueueDrainerThread(FutureCompletingBlockingQueue<?> queue) {
+ private volatile boolean wasIdleWhenFinished;
+
+ QueueDrainerThread(FutureCompletingBlockingQueue<?> queue, SplitFetcher<?, ?> fetcher, int numFetchesToTake) {
super("Queue Drainer");
setPriority(Thread.MAX_PRIORITY);
this.queue = queue;
+ this.fetcher = fetcher;
+ this.numFetchesToTake = numFetchesToTake;
}
@Override
public void go() throws Exception {
- while (running) {
- try {
- queue.take();
- }
- catch (InterruptedException ignored) {
- Thread.currentThread().interrupt();
- // fall through the loop
- }
+ int remaining = numFetchesToTake;
+ while (remaining > 0) {
+ remaining--;
+ queue.take();
}
+
+ wasIdleWhenFinished = fetcher.isIdle();
}
- public void shutdown() throws Exception {
- running = false;
- interrupt();
- sync();
+ public boolean wasIdleWhenFinished() {
+ return wasIdleWhenFinished;
}
}
}
[flink] 02/03: [tmpfix] Set CI log level to TRACE for
'org.apache.flink.connector.base.source.reader' to debug FLINK-19448
Posted by se...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
sewen pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit d0e41a04b8bec19aedf5ebe63950bc24958dcb9d
Author: Stephan Ewen <se...@apache.org>
AuthorDate: Mon Oct 12 17:44:07 2020 +0200
[tmpfix] Set CI log level to TRACE for 'org.apache.flink.connector.base.source.reader' to debug FLINK-19448
---
tools/ci/log4j.properties | 5 +++++
1 file changed, 5 insertions(+)
diff --git a/tools/ci/log4j.properties b/tools/ci/log4j.properties
index 4550a98..bc5dd0d 100644
--- a/tools/ci/log4j.properties
+++ b/tools/ci/log4j.properties
@@ -73,3 +73,8 @@ logger.consumer.level = OFF
logger.queryablestate.name = org.apache.flink.queryablestate
logger.queryablestate.level = TRACE
logger.queryablestate.appenderRef.out.ref = ConsoleAppender
+
+# Enable TRACE logging for the sources to debug FLINK-19448
+logger.sources.name = org.apache.flink.connector.base.source.reader
+logger.sources.level = TRACE
+logger.sources.appenderRef.out.ref = ConsoleAppender