You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ar...@apache.org on 2021/08/20 09:00:40 UTC
[flink] branch master updated: [FLINK-23528][kinesis] Do not
interrupt main thread during cancellation.
This is an automated email from the ASF dual-hosted git repository.
arvid pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 273dce5 [FLINK-23528][kinesis] Do not interrupt main thread during cancellation.
273dce5 is described below
commit 273dce5b030e12dd3d7bebb2f51036a198d07112
Author: Arvid Heise <ar...@ververica.com>
AuthorDate: Wed Aug 18 16:38:59 2021 +0200
[FLINK-23528][kinesis] Do not interrupt main thread during cancellation.
This commit instead uses a future for a soft wakeup during cancellation.
---
.../kinesis/internals/KinesisDataFetcher.java | 19 ++++-----
.../kinesis/internals/KinesisDataFetcherTest.java | 48 ++++++++++++++++++++++
.../testutils/TestableKinesisDataFetcher.java | 12 ++++--
3 files changed, 64 insertions(+), 15 deletions(-)
diff --git a/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/internals/KinesisDataFetcher.java b/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/internals/KinesisDataFetcher.java
index 9aee3e4..acdb2aa 100644
--- a/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/internals/KinesisDataFetcher.java
+++ b/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/internals/KinesisDataFetcher.java
@@ -73,11 +73,13 @@ import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.Set;
+import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
@@ -215,8 +217,7 @@ public class KinesisDataFetcher<T> {
/** The factory used to create record publishers that consumer from Kinesis shards. */
private final RecordPublisherFactory recordPublisherFactory;
- /** Thread that executed runFetcher(). */
- private volatile Thread mainThread;
+ private final CompletableFuture<Void> cancelFuture = new CompletableFuture<>();
/**
* The current number of shards that are actively read by this fetcher.
@@ -511,8 +512,6 @@ public class KinesisDataFetcher<T> {
return;
}
- this.mainThread = Thread.currentThread();
-
// ------------------------------------------------------------------------
// Procedures before starting the infinite while loop:
// ------------------------------------------------------------------------
@@ -748,9 +747,10 @@ public class KinesisDataFetcher<T> {
// interval if the running flag was set to false during the middle of the while loop
if (running && discoveryIntervalMillis != 0) {
try {
- Thread.sleep(discoveryIntervalMillis);
- } catch (InterruptedException iex) {
- // the sleep may be interrupted by shutdownFetcher()
+ cancelFuture.get(discoveryIntervalMillis, TimeUnit.MILLISECONDS);
+ LOG.debug("Cancelled discovery");
+ } catch (TimeoutException iex) {
+ // timeout is expected when fetcher is not cancelled
}
}
}
@@ -821,10 +821,7 @@ public class KinesisDataFetcher<T> {
} finally {
shardConsumersExecutor.shutdownNow();
- if (mainThread != null) {
- mainThread
- .interrupt(); // the main thread may be sleeping for the discovery interval
- }
+ cancelFuture.complete(null);
if (watermarkTracker != null) {
watermarkTracker.close();
diff --git a/flink-connectors/flink-connector-kinesis/src/test/java/org/apache/flink/streaming/connectors/kinesis/internals/KinesisDataFetcherTest.java b/flink-connectors/flink-connector-kinesis/src/test/java/org/apache/flink/streaming/connectors/kinesis/internals/KinesisDataFetcherTest.java
index aea01b0..c2d6dde 100644
--- a/flink-connectors/flink-connector-kinesis/src/test/java/org/apache/flink/streaming/connectors/kinesis/internals/KinesisDataFetcherTest.java
+++ b/flink-connectors/flink-connector-kinesis/src/test/java/org/apache/flink/streaming/connectors/kinesis/internals/KinesisDataFetcherTest.java
@@ -72,6 +72,7 @@ import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collectors;
import static java.util.Collections.singletonList;
+import static org.apache.flink.streaming.connectors.kinesis.config.ConsumerConfigConstants.SHARD_DISCOVERY_INTERVAL_MILLIS;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNull;
@@ -103,6 +104,53 @@ public class KinesisDataFetcherTest extends TestLogger {
assertFalse(fetcher.isRunning());
}
+ @Test
+ public void testCancelDuringDiscovery() throws Exception {
+ final String stream = "fakeStream";
+ final int numShards = 3;
+
+ Properties standardProperties = TestUtils.getStandardProperties();
+ standardProperties.setProperty(SHARD_DISCOVERY_INTERVAL_MILLIS, "10000000");
+ final LinkedList<KinesisStreamShardState> testShardStates = new LinkedList<>();
+ final TestSourceContext<String> sourceContext = new TestSourceContext<>();
+ TestableKinesisDataFetcher<String> fetcher =
+ new TestableKinesisDataFetcher<String>(
+ singletonList(stream),
+ sourceContext,
+ standardProperties,
+ new KinesisDeserializationSchemaWrapper<>(new SimpleStringSchema()),
+ 1,
+ 0,
+ new AtomicReference<>(),
+ testShardStates,
+ new HashMap<>(),
+ FakeKinesisBehavioursFactory.nonReshardedStreamsBehaviour(
+ Collections.singletonMap(stream, numShards)));
+
+ // FlinkKinesisConsumer is responsible for setting up the fetcher before it can be run;
+ // run the consumer until it reaches the point where the fetcher starts to run
+ final DummyFlinkKinesisConsumer<String> consumer =
+ new DummyFlinkKinesisConsumer<>(TestUtils.getStandardProperties(), fetcher, 1, 0);
+
+ CheckedThread consumerThread =
+ new CheckedThread() {
+ @Override
+ public void go() throws Exception {
+ consumer.run(new TestSourceContext<>());
+ }
+ };
+ consumerThread.start();
+
+ // wait for the second discovery to be triggered, that has a high probability to be inside
+ // discovery sleep (10k s)
+ fetcher.waitUntilDiscovery(2);
+
+ Thread.sleep(1000);
+
+ consumer.cancel();
+ consumerThread.sync();
+ }
+
@Test(expected = RuntimeException.class)
public void testIfNoShardsAreFoundShouldThrowException() throws Exception {
KinesisDataFetcher<String> fetcher =
diff --git a/flink-connectors/flink-connector-kinesis/src/test/java/org/apache/flink/streaming/connectors/kinesis/testutils/TestableKinesisDataFetcher.java b/flink-connectors/flink-connector-kinesis/src/test/java/org/apache/flink/streaming/connectors/kinesis/testutils/TestableKinesisDataFetcher.java
index d517cc3..b4c69ee 100644
--- a/flink-connectors/flink-connector-kinesis/src/test/java/org/apache/flink/streaming/connectors/kinesis/testutils/TestableKinesisDataFetcher.java
+++ b/flink-connectors/flink-connector-kinesis/src/test/java/org/apache/flink/streaming/connectors/kinesis/testutils/TestableKinesisDataFetcher.java
@@ -34,6 +34,7 @@ import java.util.LinkedList;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
@@ -46,7 +47,7 @@ import static org.mockito.Mockito.when;
public class TestableKinesisDataFetcher<T> extends KinesisDataFetcher<T> {
private final OneShotLatch runWaiter;
- private final OneShotLatch initialDiscoveryWaiter;
+ private final Semaphore discoveryWaiter = new Semaphore(0);
private final OneShotLatch shutdownWaiter;
private volatile boolean running;
@@ -107,7 +108,6 @@ public class TestableKinesisDataFetcher<T> extends KinesisDataFetcher<T> {
properties -> fakeKinesisV2);
this.runWaiter = new OneShotLatch();
- this.initialDiscoveryWaiter = new OneShotLatch();
this.shutdownWaiter = new OneShotLatch();
this.running = true;
@@ -163,11 +163,15 @@ public class TestableKinesisDataFetcher<T> extends KinesisDataFetcher<T> {
@Override
public List<StreamShardHandle> discoverNewShardsToSubscribe() throws InterruptedException {
List<StreamShardHandle> newShards = super.discoverNewShardsToSubscribe();
- initialDiscoveryWaiter.trigger();
+ discoveryWaiter.release();
return newShards;
}
public void waitUntilInitialDiscovery() throws InterruptedException {
- initialDiscoveryWaiter.await();
+ discoveryWaiter.acquire();
+ }
+
+ public void waitUntilDiscovery(int number) throws InterruptedException {
+ discoveryWaiter.acquire(number);
}
}