From 24ba9bd343483dfb037bb7552c1a5dd5f3beacb1 Mon Sep 17 00:00:00 2001 From: Chris Egerton Date: Mon, 14 Oct 2019 20:07:01 -0700 Subject: [PATCH 1/7] KAFKA-9051: Prematurely complete source offset read requests for stopped tasks --- .../apache/kafka/connect/runtime/Worker.java | 3 +- .../connect/runtime/WorkerSourceTask.java | 7 +- .../storage/KafkaOffsetBackingStore.java | 54 ++++++++--- .../storage/MemoryOffsetBackingStore.java | 91 +++++++++++++++---- .../connect/storage/OffsetBackingStore.java | 8 +- .../connect/storage/OffsetReadFuture.java | 32 +++++++ .../storage/OffsetStorageReaderImpl.java | 31 ++++++- .../runtime/ErrorHandlingTaskTest.java | 4 +- .../connect/runtime/WorkerSourceTaskTest.java | 25 ++++- .../storage/FileOffsetBackingStoreTest.java | 14 +-- .../storage/KafkaOffsetBackingStoreTest.java | 51 +++-------- 11 files changed, 227 insertions(+), 93 deletions(-) create mode 100644 connect/runtime/src/main/java/org/apache/kafka/connect/storage/OffsetReadFuture.java diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Worker.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Worker.java index 27955477fc573..af6593ad3019d 100644 --- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Worker.java +++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Worker.java @@ -52,7 +52,6 @@ import org.apache.kafka.connect.storage.Converter; import org.apache.kafka.connect.storage.HeaderConverter; import org.apache.kafka.connect.storage.OffsetBackingStore; -import org.apache.kafka.connect.storage.OffsetStorageReader; import org.apache.kafka.connect.storage.OffsetStorageReaderImpl; import org.apache.kafka.connect.storage.OffsetStorageWriter; import org.apache.kafka.connect.util.ConnectorTaskId; @@ -511,7 +510,7 @@ private WorkerTask buildWorkerTask(ClusterConfigState configState, retryWithToleranceOperator.reporters(sourceTaskReporters(id, connConfig, errorHandlingMetrics)); TransformationChain transformationChain = new TransformationChain<>(connConfig.transformations(), retryWithToleranceOperator); log.info("Initializing: {}", transformationChain); - OffsetStorageReader offsetReader = new OffsetStorageReaderImpl(offsetBackingStore, id.connector(), + OffsetStorageReaderImpl offsetReader = new OffsetStorageReaderImpl(offsetBackingStore, id.connector(), internalKeyConverter, internalValueConverter); OffsetStorageWriter offsetWriter = new OffsetStorageWriter(offsetBackingStore, id.connector(), internalKeyConverter, internalValueConverter); diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSourceTask.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSourceTask.java index 5e24d4145aef2..f135bfa09e624 100644 --- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSourceTask.java +++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSourceTask.java @@ -41,7 +41,7 @@ import org.apache.kafka.connect.source.SourceTask; import org.apache.kafka.connect.storage.Converter; import org.apache.kafka.connect.storage.HeaderConverter; -import org.apache.kafka.connect.storage.OffsetStorageReader; +import org.apache.kafka.connect.storage.OffsetStorageReaderImpl; import org.apache.kafka.connect.storage.OffsetStorageWriter; import org.apache.kafka.connect.util.ConnectUtils; import org.apache.kafka.connect.util.ConnectorTaskId; @@ -75,7 +75,7 @@ class WorkerSourceTask extends WorkerTask { private final HeaderConverter headerConverter; private final TransformationChain transformationChain; private KafkaProducer producer; - private final OffsetStorageReader offsetReader; + private final OffsetStorageReaderImpl offsetReader; private final OffsetStorageWriter offsetWriter; private final Time time; private final SourceTaskMetricsGroup sourceTaskMetricsGroup; @@ -105,7 +105,7 @@ public WorkerSourceTask(ConnectorTaskId id, HeaderConverter headerConverter, TransformationChain transformationChain, KafkaProducer producer, - OffsetStorageReader offsetReader, + OffsetStorageReaderImpl offsetReader, OffsetStorageWriter offsetWriter, WorkerConfig workerConfig, ClusterConfigState configState, @@ -175,6 +175,7 @@ protected void releaseResources() { @Override public void stop() { super.stop(); + offsetReader.prematurelyCompleteRequests(); stopRequestedLatch.countDown(); synchronized (this) { if (finishedStart) diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/storage/KafkaOffsetBackingStore.java b/connect/runtime/src/main/java/org/apache/kafka/connect/storage/KafkaOffsetBackingStore.java index 195c498edb76f..b457e80911112 100644 --- a/connect/runtime/src/main/java/org/apache/kafka/connect/storage/KafkaOffsetBackingStore.java +++ b/connect/runtime/src/main/java/org/apache/kafka/connect/storage/KafkaOffsetBackingStore.java @@ -42,6 +42,7 @@ import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicBoolean; /** *

@@ -118,17 +119,8 @@ public void stop() { } @Override - public Future> get(final Collection keys, - final Callback> callback) { - ConvertingFutureCallback> future = new ConvertingFutureCallback>(callback) { - @Override - public Map convert(Void result) { - Map values = new HashMap<>(); - for (ByteBuffer key : keys) - values.put(key, data.get(key)); - return values; - } - }; + public OffsetReadFuture get(final Collection keys) { + KafkaOffsetReadFuture future = new KafkaOffsetReadFuture(keys); // This operation may be relatively (but not too) expensive since it always requires checking end offsets, even // if we've already read up to the end. However, it also should not be common (offsets should only be read when // resetting a task). Always requiring that we read to the end is simpler than trying to differentiate when it @@ -231,5 +223,45 @@ public synchronized Void get(long timeout, TimeUnit unit) throws InterruptedExce } } + private class KafkaOffsetReadFuture extends ConvertingFutureCallback> implements OffsetReadFuture { + private final Collection keys; + private final AtomicBoolean completed; + + public KafkaOffsetReadFuture(Collection keys) { + super(null); + this.keys = keys; + this.completed = new AtomicBoolean(false); + } + + @Override + public Map convert(Void result) { + Map values = new HashMap<>(); + for (ByteBuffer key : keys) + values.put(key, data.get(key)); + return values; + } + @Override + public void prematurelyComplete() { + log.debug("Prematurely completing offset read future"); + onCompletion(null, null); + } + + @Override + public Map get() throws InterruptedException, ExecutionException { + log.trace("Blocking on offset read request"); + return super.get(); + } + + @Override + public void onCompletion(Throwable error, Void result) { + if (!completed.getAndSet(true)) { + log.trace("Offset request completed (either normally or prematurely)"); + super.onCompletion(error, result); + } else { + log.trace("Ignoring onCompletion invocation as offset read has already completed"); + } + } + + } } diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/storage/MemoryOffsetBackingStore.java b/connect/runtime/src/main/java/org/apache/kafka/connect/storage/MemoryOffsetBackingStore.java index ab8130b6ef776..8024eb857f210 100644 --- a/connect/runtime/src/main/java/org/apache/kafka/connect/storage/MemoryOffsetBackingStore.java +++ b/connect/runtime/src/main/java/org/apache/kafka/connect/storage/MemoryOffsetBackingStore.java @@ -27,10 +27,12 @@ import java.util.HashMap; import java.util.Map; import java.util.concurrent.Callable; +import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; /** * Implementation of OffsetBackingStore that doesn't actually persist any data. To ensure this @@ -75,22 +77,8 @@ public void stop() { } @Override - public Future> get( - final Collection keys, - final Callback> callback) { - return executor.submit(new Callable>() { - @Override - public Map call() throws Exception { - Map result = new HashMap<>(); - for (ByteBuffer key : keys) { - result.put(key, data.get(key)); - } - if (callback != null) - callback.onCompletion(null, result); - return result; - } - }); - + public OffsetReadFuture get(final Collection keys) { + return new MemoryOffsetReadFuture(keys); } @Override @@ -114,4 +102,75 @@ public Void call() throws Exception { protected void save() { } + + private class MemoryOffsetReadFuture implements OffsetReadFuture { + private final Map result; + private final Collection keys; + private final CountDownLatch completed; + private final Future underlyingOffsetFuture; + + public MemoryOffsetReadFuture(Collection keys) { + this.result = new HashMap<>(); + this.keys = keys; + this.completed = new CountDownLatch(1); + this.underlyingOffsetFuture = executor.submit(new Runnable() { + @Override + public void run() { + synchronized (this) { + if (isDone()) { + return; + } + collectResults(); + completed.countDown(); + } + } + }); + } + + @Override + public void prematurelyComplete() { + synchronized (this) { + if (isDone()) { + return; + } + collectResults(); + completed.countDown(); + } + } + + @Override + public boolean cancel(boolean mayInterruptIfRunning) { + return underlyingOffsetFuture.cancel(mayInterruptIfRunning); + } + + @Override + public boolean isCancelled() { + return underlyingOffsetFuture.isCancelled(); + } + + @Override + public boolean isDone() { + return completed.getCount() == 0; + } + + @Override + public Map get() throws InterruptedException { + completed.await(); + return result; + } + + @Override + public Map get(long timeout, TimeUnit unit) + throws TimeoutException, InterruptedException { + if (!completed.await(timeout, unit)) + throw new TimeoutException("Timed out waiting for future"); + return result; + } + + private void collectResults() { + for (ByteBuffer key : keys) { + result.put(key, data.get(key)); + } + } + } } diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/storage/OffsetBackingStore.java b/connect/runtime/src/main/java/org/apache/kafka/connect/storage/OffsetBackingStore.java index 9998164ddf5bf..a9002320b4e66 100644 --- a/connect/runtime/src/main/java/org/apache/kafka/connect/storage/OffsetBackingStore.java +++ b/connect/runtime/src/main/java/org/apache/kafka/connect/storage/OffsetBackingStore.java @@ -53,12 +53,9 @@ public interface OffsetBackingStore { /** * Get the values for the specified keys * @param keys list of keys to look up - * @param callback callback to invoke on completion * @return future for the resulting map from key to value */ - Future> get( - Collection keys, - Callback> callback); + OffsetReadFuture get(Collection keys); /** * Set the specified keys and values. @@ -66,8 +63,7 @@ Future> get( * @param callback callback to invoke on completion * @return void future for the operation */ - Future set(Map values, - Callback callback); + Future set(Map values, Callback callback); /** * Configure class with the given key-value pairs diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/storage/OffsetReadFuture.java b/connect/runtime/src/main/java/org/apache/kafka/connect/storage/OffsetReadFuture.java new file mode 100644 index 0000000000000..de58d204531c7 --- /dev/null +++ b/connect/runtime/src/main/java/org/apache/kafka/connect/storage/OffsetReadFuture.java @@ -0,0 +1,32 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.kafka.connect.storage; + + +import java.nio.ByteBuffer; +import java.util.Map; +import java.util.concurrent.Future; + +public interface OffsetReadFuture extends Future> { + + /** + * Without {@link Future#cancel(boolean) cancelling} the operation, attempt to immediately + * unblock any threads which are waiting for this future to complete, providing them with + * potentially incomplete or out-of-date offset information if necessary. + */ + void prematurelyComplete(); +} \ No newline at end of file diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/storage/OffsetStorageReaderImpl.java b/connect/runtime/src/main/java/org/apache/kafka/connect/storage/OffsetStorageReaderImpl.java index 9f926dc5040aa..d01956bf991c9 100644 --- a/connect/runtime/src/main/java/org/apache/kafka/connect/storage/OffsetStorageReaderImpl.java +++ b/connect/runtime/src/main/java/org/apache/kafka/connect/storage/OffsetStorageReaderImpl.java @@ -26,7 +26,9 @@ import java.util.Collection; import java.util.Collections; import java.util.HashMap; +import java.util.HashSet; import java.util.Map; +import java.util.Set; /** * Implementation of OffsetStorageReader. Unlike OffsetStorageWriter which is implemented @@ -40,6 +42,7 @@ public class OffsetStorageReaderImpl implements OffsetStorageReader { private final String namespace; private final Converter keyConverter; private final Converter valueConverter; + private final Set offsetReadFutures; public OffsetStorageReaderImpl(OffsetBackingStore backingStore, String namespace, Converter keyConverter, Converter valueConverter) { @@ -47,6 +50,7 @@ public OffsetStorageReaderImpl(OffsetBackingStore backingStore, String namespace this.namespace = namespace; this.keyConverter = keyConverter; this.valueConverter = valueConverter; + this.offsetReadFutures = new HashSet<>(); } @Override @@ -76,7 +80,18 @@ public Map, Map> offsets(Collection serialized value from backing store Map raw; try { - raw = backingStore.get(serializedToOriginal.keySet(), null).get(); + OffsetReadFuture offsetReadFuture; + synchronized (offsetReadFutures) { + offsetReadFuture = backingStore.get(serializedToOriginal.keySet()); + offsetReadFutures.add(offsetReadFuture); + } + try { + raw = offsetReadFuture.get(); + } finally { + synchronized (offsetReadFutures) { + offsetReadFutures.remove(offsetReadFuture); + } + } } catch (Exception e) { log.error("Failed to fetch offsets from namespace {}: ", namespace, e); throw new ConnectException("Failed to fetch offsets.", e); @@ -108,4 +123,18 @@ public Map, Map> offsets(Collection producer; @Mock - OffsetStorageReader offsetReader; + OffsetStorageReaderImpl offsetReader; @Mock OffsetStorageWriter offsetWriter; diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSourceTaskTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSourceTaskTest.java index e8039310b20da..d1ecdb4e23e4f 100644 --- a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSourceTaskTest.java +++ b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSourceTaskTest.java @@ -43,7 +43,7 @@ import org.apache.kafka.connect.source.SourceTaskContext; import org.apache.kafka.connect.storage.Converter; import org.apache.kafka.connect.storage.HeaderConverter; -import org.apache.kafka.connect.storage.OffsetStorageReader; +import org.apache.kafka.connect.storage.OffsetStorageReaderImpl; import org.apache.kafka.connect.storage.OffsetStorageWriter; import org.apache.kafka.connect.storage.StringConverter; import org.apache.kafka.connect.util.Callback; @@ -114,7 +114,7 @@ public class WorkerSourceTaskTest extends ThreadedTest { @Mock private HeaderConverter headerConverter; @Mock private TransformationChain transformationChain; @Mock private KafkaProducer producer; - @Mock private OffsetStorageReader offsetReader; + @Mock private OffsetStorageReaderImpl offsetReader; @Mock private OffsetStorageWriter offsetWriter; @Mock private ClusterConfigState clusterConfigState; private WorkerSourceTask workerTask; @@ -193,6 +193,9 @@ public Void answer() throws Throwable { statusListener.onShutdown(taskId); EasyMock.expectLastCall(); + offsetReader.prematurelyCompleteRequests(); + EasyMock.expectLastCall(); + PowerMock.replayAll(); workerTask.initialize(TASK_CONFIG); @@ -238,6 +241,9 @@ public void testPause() throws Exception { transformationChain.close(); EasyMock.expectLastCall(); + offsetReader.prematurelyCompleteRequests(); + EasyMock.expectLastCall(); + PowerMock.replayAll(); workerTask.initialize(TASK_CONFIG); @@ -287,6 +293,9 @@ public void testPollsInBackground() throws Exception { transformationChain.close(); EasyMock.expectLastCall(); + offsetReader.prematurelyCompleteRequests(); + EasyMock.expectLastCall(); + PowerMock.replayAll(); workerTask.initialize(TASK_CONFIG); @@ -336,6 +345,9 @@ public List answer() throws Throwable { transformationChain.close(); EasyMock.expectLastCall(); + offsetReader.prematurelyCompleteRequests(); + EasyMock.expectLastCall(); + PowerMock.replayAll(); workerTask.initialize(TASK_CONFIG); @@ -425,6 +437,9 @@ public void testCommit() throws Exception { transformationChain.close(); EasyMock.expectLastCall(); + offsetReader.prematurelyCompleteRequests(); + EasyMock.expectLastCall(); + PowerMock.replayAll(); workerTask.initialize(TASK_CONFIG); @@ -470,6 +485,9 @@ public void testCommitFailure() throws Exception { transformationChain.close(); EasyMock.expectLastCall(); + offsetReader.prematurelyCompleteRequests(); + EasyMock.expectLastCall(); + PowerMock.replayAll(); workerTask.initialize(TASK_CONFIG); @@ -675,6 +693,9 @@ public Object answer() throws Throwable { transformationChain.close(); EasyMock.expectLastCall(); + offsetReader.prematurelyCompleteRequests(); + EasyMock.expectLastCall(); + PowerMock.replayAll(); workerTask.initialize(TASK_CONFIG); diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/storage/FileOffsetBackingStoreTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/storage/FileOffsetBackingStoreTest.java index df955f8813d53..1fdb91a2dcb2d 100644 --- a/connect/runtime/src/test/java/org/apache/kafka/connect/storage/FileOffsetBackingStoreTest.java +++ b/connect/runtime/src/test/java/org/apache/kafka/connect/storage/FileOffsetBackingStoreTest.java @@ -71,12 +71,11 @@ public void teardown() { @Test public void testGetSet() throws Exception { Callback setCallback = expectSuccessfulSetCallback(); - Callback> getCallback = expectSuccessfulGetCallback(); PowerMock.replayAll(); store.set(firstSet, setCallback).get(); - Map values = store.get(Arrays.asList(buffer("key"), buffer("bad")), getCallback).get(); + Map values = store.get(Arrays.asList(buffer("key"), buffer("bad"))).get(); assertEquals(buffer("value"), values.get(buffer("key"))); assertEquals(null, values.get(buffer("bad"))); @@ -86,7 +85,6 @@ public void testGetSet() throws Exception { @Test public void testSaveRestore() throws Exception { Callback setCallback = expectSuccessfulSetCallback(); - Callback> getCallback = expectSuccessfulGetCallback(); PowerMock.replayAll(); store.set(firstSet, setCallback).get(); @@ -96,7 +94,7 @@ public void testSaveRestore() throws Exception { FileOffsetBackingStore restore = new FileOffsetBackingStore(); restore.configure(config); restore.start(); - Map values = restore.get(Arrays.asList(buffer("key")), getCallback).get(); + Map values = restore.get(Arrays.asList(buffer("key"))).get(); assertEquals(buffer("value"), values.get(buffer("key"))); PowerMock.verifyAll(); @@ -113,12 +111,4 @@ private Callback expectSuccessfulSetCallback() { PowerMock.expectLastCall(); return setCallback; } - - @SuppressWarnings("unchecked") - private Callback> expectSuccessfulGetCallback() { - Callback> getCallback = PowerMock.createMock(Callback.class); - getCallback.onCompletion(EasyMock.isNull(Throwable.class), EasyMock.anyObject(Map.class)); - PowerMock.expectLastCall(); - return getCallback; - } } diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/storage/KafkaOffsetBackingStoreTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/storage/KafkaOffsetBackingStoreTest.java index 1df63f35e8a76..24b14bf9969a3 100644 --- a/connect/runtime/src/test/java/org/apache/kafka/connect/storage/KafkaOffsetBackingStoreTest.java +++ b/connect/runtime/src/test/java/org/apache/kafka/connect/storage/KafkaOffsetBackingStoreTest.java @@ -218,16 +218,10 @@ public Object answer() throws Throwable { // Getting from empty store should return nulls final AtomicBoolean getInvokedAndPassed = new AtomicBoolean(false); - store.get(Arrays.asList(TP0_KEY, TP1_KEY), new Callback>() { - @Override - public void onCompletion(Throwable error, Map result) { - // Since we didn't read them yet, these will be null - assertEquals(null, result.get(TP0_KEY)); - assertEquals(null, result.get(TP1_KEY)); - getInvokedAndPassed.set(true); - } - }).get(10000, TimeUnit.MILLISECONDS); - assertTrue(getInvokedAndPassed.get()); + Map offsets = store.get(Arrays.asList(TP0_KEY, TP1_KEY)).get(10000, TimeUnit.MILLISECONDS); + // Since we didn't read them yet, these will be null + assertNull(offsets.get(TP0_KEY)); + assertNull(offsets.get(TP1_KEY)); // Set some offsets Map toSet = new HashMap<>(); @@ -251,27 +245,14 @@ public void onCompletion(Throwable error, Void result) { // Getting data should read to end of our published data and return it final AtomicBoolean secondGetInvokedAndPassed = new AtomicBoolean(false); - store.get(Arrays.asList(TP0_KEY, TP1_KEY), new Callback>() { - @Override - public void onCompletion(Throwable error, Map result) { - assertEquals(TP0_VALUE, result.get(TP0_KEY)); - assertEquals(TP1_VALUE, result.get(TP1_KEY)); - secondGetInvokedAndPassed.set(true); - } - }).get(10000, TimeUnit.MILLISECONDS); - assertTrue(secondGetInvokedAndPassed.get()); + offsets = store.get(Arrays.asList(TP0_KEY, TP1_KEY)).get(10000, TimeUnit.MILLISECONDS); + assertEquals(TP0_VALUE, offsets.get(TP0_KEY)); + assertEquals(TP1_VALUE, offsets.get(TP1_KEY)); // Getting data should read to end of our published data and return it - final AtomicBoolean thirdGetInvokedAndPassed = new AtomicBoolean(false); - store.get(Arrays.asList(TP0_KEY, TP1_KEY), new Callback>() { - @Override - public void onCompletion(Throwable error, Map result) { - assertEquals(TP0_VALUE_NEW, result.get(TP0_KEY)); - assertEquals(TP1_VALUE_NEW, result.get(TP1_KEY)); - thirdGetInvokedAndPassed.set(true); - } - }).get(10000, TimeUnit.MILLISECONDS); - assertTrue(thirdGetInvokedAndPassed.get()); + offsets = store.get(Arrays.asList(TP0_KEY, TP1_KEY)).get(10000, TimeUnit.MILLISECONDS); + assertEquals(TP0_VALUE_NEW, offsets.get(TP0_KEY)); + assertEquals(TP1_VALUE_NEW, offsets.get(TP1_KEY)); store.stop(); @@ -330,15 +311,9 @@ public void onCompletion(Throwable error, Void result) { // Getting data should read to end of our published data and return it final AtomicBoolean secondGetInvokedAndPassed = new AtomicBoolean(false); - store.get(Arrays.asList(null, TP1_KEY), new Callback>() { - @Override - public void onCompletion(Throwable error, Map result) { - assertEquals(TP0_VALUE, result.get(null)); - assertNull(result.get(TP1_KEY)); - secondGetInvokedAndPassed.set(true); - } - }).get(10000, TimeUnit.MILLISECONDS); - assertTrue(secondGetInvokedAndPassed.get()); + Map offsets = store.get(Arrays.asList(null, TP1_KEY)).get(10000, TimeUnit.MILLISECONDS); + assertEquals(TP0_VALUE, offsets.get(null)); + assertNull(offsets.get(TP1_KEY)); store.stop(); From f89b4547c0ded76f5387aec8f67024ffe28db401 Mon Sep 17 00:00:00 2001 From: Chris Egerton Date: Wed, 16 Oct 2019 13:31:45 -0700 Subject: [PATCH 2/7] KAFKA-9051: Address review comments --- .../apache/kafka/connect/runtime/Worker.java | 3 +- .../connect/runtime/WorkerSourceTask.java | 8 ++--- .../storage/CloseableOffsetStorageReader.java | 32 +++++++++++++++++++ .../storage/KafkaOffsetBackingStore.java | 15 ++++++--- .../storage/MemoryOffsetBackingStore.java | 3 +- .../connect/storage/OffsetReadFuture.java | 7 ++-- .../storage/OffsetStorageReaderImpl.java | 19 ++++++----- .../connect/runtime/WorkerSourceTaskTest.java | 21 ++++++------ .../storage/KafkaOffsetBackingStoreTest.java | 3 -- 9 files changed, 75 insertions(+), 36 deletions(-) create mode 100644 connect/runtime/src/main/java/org/apache/kafka/connect/storage/CloseableOffsetStorageReader.java diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Worker.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Worker.java index af6593ad3019d..cb236c511c471 100644 --- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Worker.java +++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Worker.java @@ -49,6 +49,7 @@ import org.apache.kafka.connect.sink.SinkTask; import org.apache.kafka.connect.source.SourceRecord; import org.apache.kafka.connect.source.SourceTask; +import org.apache.kafka.connect.storage.CloseableOffsetStorageReader; import org.apache.kafka.connect.storage.Converter; import org.apache.kafka.connect.storage.HeaderConverter; import org.apache.kafka.connect.storage.OffsetBackingStore; @@ -510,7 +511,7 @@ private WorkerTask buildWorkerTask(ClusterConfigState configState, retryWithToleranceOperator.reporters(sourceTaskReporters(id, connConfig, errorHandlingMetrics)); TransformationChain transformationChain = new TransformationChain<>(connConfig.transformations(), retryWithToleranceOperator); log.info("Initializing: {}", transformationChain); - OffsetStorageReaderImpl offsetReader = new OffsetStorageReaderImpl(offsetBackingStore, id.connector(), + CloseableOffsetStorageReader offsetReader = new OffsetStorageReaderImpl(offsetBackingStore, id.connector(), internalKeyConverter, internalValueConverter); OffsetStorageWriter offsetWriter = new OffsetStorageWriter(offsetBackingStore, id.connector(), internalKeyConverter, internalValueConverter); diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSourceTask.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSourceTask.java index f135bfa09e624..9ddcbe1f1bd63 100644 --- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSourceTask.java +++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSourceTask.java @@ -39,9 +39,9 @@ import org.apache.kafka.connect.runtime.errors.Stage; import org.apache.kafka.connect.source.SourceRecord; import org.apache.kafka.connect.source.SourceTask; +import org.apache.kafka.connect.storage.CloseableOffsetStorageReader; import org.apache.kafka.connect.storage.Converter; import org.apache.kafka.connect.storage.HeaderConverter; -import org.apache.kafka.connect.storage.OffsetStorageReaderImpl; import org.apache.kafka.connect.storage.OffsetStorageWriter; import org.apache.kafka.connect.util.ConnectUtils; import org.apache.kafka.connect.util.ConnectorTaskId; @@ -75,7 +75,7 @@ class WorkerSourceTask extends WorkerTask { private final HeaderConverter headerConverter; private final TransformationChain transformationChain; private KafkaProducer producer; - private final OffsetStorageReaderImpl offsetReader; + private final CloseableOffsetStorageReader offsetReader; private final OffsetStorageWriter offsetWriter; private final Time time; private final SourceTaskMetricsGroup sourceTaskMetricsGroup; @@ -105,7 +105,7 @@ public WorkerSourceTask(ConnectorTaskId id, HeaderConverter headerConverter, TransformationChain transformationChain, KafkaProducer producer, - OffsetStorageReaderImpl offsetReader, + CloseableOffsetStorageReader offsetReader, OffsetStorageWriter offsetWriter, WorkerConfig workerConfig, ClusterConfigState configState, @@ -175,7 +175,7 @@ protected void releaseResources() { @Override public void stop() { super.stop(); - offsetReader.prematurelyCompleteRequests(); + offsetReader.close(); stopRequestedLatch.countDown(); synchronized (this) { if (finishedStart) diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/storage/CloseableOffsetStorageReader.java b/connect/runtime/src/main/java/org/apache/kafka/connect/storage/CloseableOffsetStorageReader.java new file mode 100644 index 0000000000000..65f925dc0ac24 --- /dev/null +++ b/connect/runtime/src/main/java/org/apache/kafka/connect/storage/CloseableOffsetStorageReader.java @@ -0,0 +1,32 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.kafka.connect.storage; + +import java.io.Closeable; +import java.util.Collection; +import java.util.Map; + +public interface CloseableOffsetStorageReader extends Closeable, OffsetStorageReader { + + /** + * Invoke {@link OffsetReadFuture#forceComplete()} on all outstanding offset read requests, and + * return an empty map and {@code null} in all future calls to {@link #offsets(Collection)} and + * {@link #offset(Map)}, respectively. This is useful for unblocking task threads which need to shut + * down but are blocked on offset reads. + */ + void close(); +} diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/storage/KafkaOffsetBackingStore.java b/connect/runtime/src/main/java/org/apache/kafka/connect/storage/KafkaOffsetBackingStore.java index b457e80911112..28fb90ee46c1f 100644 --- a/connect/runtime/src/main/java/org/apache/kafka/connect/storage/KafkaOffsetBackingStore.java +++ b/connect/runtime/src/main/java/org/apache/kafka/connect/storage/KafkaOffsetBackingStore.java @@ -226,24 +226,29 @@ public synchronized Void get(long timeout, TimeUnit unit) throws InterruptedExce private class KafkaOffsetReadFuture extends ConvertingFutureCallback> implements OffsetReadFuture { private final Collection keys; private final AtomicBoolean completed; + private transient boolean forceComplete; public KafkaOffsetReadFuture(Collection keys) { super(null); this.keys = keys; this.completed = new AtomicBoolean(false); + this.forceComplete = false; } @Override public Map convert(Void result) { Map values = new HashMap<>(); - for (ByteBuffer key : keys) - values.put(key, data.get(key)); + if (!forceComplete) { + for (ByteBuffer key : keys) + values.put(key, data.get(key)); + } return values; } @Override - public void prematurelyComplete() { - log.debug("Prematurely completing offset read future"); + public void forceComplete() { + log.debug("Forcing completion of offset read future"); + forceComplete = true; onCompletion(null, null); } @@ -256,7 +261,7 @@ public Map get() throws InterruptedException, ExecutionE @Override public void onCompletion(Throwable error, Void result) { if (!completed.getAndSet(true)) { - log.trace("Offset request completed (either normally or prematurely)"); + log.trace("Offset request completed (either normally or forcibly)"); super.onCompletion(error, result); } else { log.trace("Ignoring onCompletion invocation as offset read has already completed"); diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/storage/MemoryOffsetBackingStore.java b/connect/runtime/src/main/java/org/apache/kafka/connect/storage/MemoryOffsetBackingStore.java index 8024eb857f210..3d5e1b246be62 100644 --- a/connect/runtime/src/main/java/org/apache/kafka/connect/storage/MemoryOffsetBackingStore.java +++ b/connect/runtime/src/main/java/org/apache/kafka/connect/storage/MemoryOffsetBackingStore.java @@ -128,12 +128,11 @@ public void run() { } @Override - public void prematurelyComplete() { + public void forceComplete() { synchronized (this) { if (isDone()) { return; } - collectResults(); completed.countDown(); } } diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/storage/OffsetReadFuture.java b/connect/runtime/src/main/java/org/apache/kafka/connect/storage/OffsetReadFuture.java index de58d204531c7..142acca0d189a 100644 --- a/connect/runtime/src/main/java/org/apache/kafka/connect/storage/OffsetReadFuture.java +++ b/connect/runtime/src/main/java/org/apache/kafka/connect/storage/OffsetReadFuture.java @@ -24,9 +24,8 @@ public interface OffsetReadFuture extends Future> { /** - * Without {@link Future#cancel(boolean) cancelling} the operation, attempt to immediately - * unblock any threads which are waiting for this future to complete, providing them with - * potentially incomplete or out-of-date offset information if necessary. + * Without {@link Future#cancel(boolean) cancelling} the operation, immediately return an empty + * map to any threads which are waiting for this future to complete. */ - void prematurelyComplete(); + void forceComplete(); } \ No newline at end of file diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/storage/OffsetStorageReaderImpl.java b/connect/runtime/src/main/java/org/apache/kafka/connect/storage/OffsetStorageReaderImpl.java index d01956bf991c9..e69d5130afd47 100644 --- a/connect/runtime/src/main/java/org/apache/kafka/connect/storage/OffsetStorageReaderImpl.java +++ b/connect/runtime/src/main/java/org/apache/kafka/connect/storage/OffsetStorageReaderImpl.java @@ -29,19 +29,21 @@ import java.util.HashSet; import java.util.Map; import java.util.Set; +import java.util.concurrent.atomic.AtomicBoolean; /** * Implementation of OffsetStorageReader. Unlike OffsetStorageWriter which is implemented * directly, the interface is only separate from this implementation because it needs to be * included in the public API package. */ -public class OffsetStorageReaderImpl implements OffsetStorageReader { +public class OffsetStorageReaderImpl implements CloseableOffsetStorageReader { private static final Logger log = LoggerFactory.getLogger(OffsetStorageReaderImpl.class); private final OffsetBackingStore backingStore; private final String namespace; private final Converter keyConverter; private final Converter valueConverter; + private final AtomicBoolean closed; private final Set offsetReadFutures; public OffsetStorageReaderImpl(OffsetBackingStore backingStore, String namespace, @@ -50,6 +52,7 @@ public OffsetStorageReaderImpl(OffsetBackingStore backingStore, String namespace this.namespace = namespace; this.keyConverter = keyConverter; this.valueConverter = valueConverter; + this.closed = new AtomicBoolean(false); this.offsetReadFutures = new HashSet<>(); } @@ -82,9 +85,13 @@ public Map, Map> offsets(Collection Map, Map> offsets(Collection transformationChain; @Mock private KafkaProducer producer; - @Mock private OffsetStorageReaderImpl offsetReader; + @Mock private CloseableOffsetStorageReader offsetReader; @Mock private OffsetStorageWriter offsetWriter; @Mock private ClusterConfigState clusterConfigState; private WorkerSourceTask workerTask; @@ -193,7 +193,7 @@ public Void answer() throws Throwable { statusListener.onShutdown(taskId); EasyMock.expectLastCall(); - offsetReader.prematurelyCompleteRequests(); + offsetReader.close(); EasyMock.expectLastCall(); PowerMock.replayAll(); @@ -241,7 +241,7 @@ public void testPause() throws Exception { transformationChain.close(); EasyMock.expectLastCall(); - offsetReader.prematurelyCompleteRequests(); + offsetReader.close(); EasyMock.expectLastCall(); PowerMock.replayAll(); @@ -293,7 +293,7 @@ public void testPollsInBackground() throws Exception { transformationChain.close(); EasyMock.expectLastCall(); - offsetReader.prematurelyCompleteRequests(); + offsetReader.close(); EasyMock.expectLastCall(); PowerMock.replayAll(); @@ -345,7 +345,7 @@ public List answer() throws Throwable { transformationChain.close(); EasyMock.expectLastCall(); - offsetReader.prematurelyCompleteRequests(); + offsetReader.close(); EasyMock.expectLastCall(); PowerMock.replayAll(); @@ -392,6 +392,9 @@ public void testPollReturnsNoRecords() throws Exception { transformationChain.close(); EasyMock.expectLastCall(); + offsetReader.close(); + EasyMock.expectLastCall(); + PowerMock.replayAll(); workerTask.initialize(TASK_CONFIG); @@ -437,7 +440,7 @@ public void testCommit() throws Exception { transformationChain.close(); EasyMock.expectLastCall(); - offsetReader.prematurelyCompleteRequests(); + offsetReader.close(); EasyMock.expectLastCall(); PowerMock.replayAll(); @@ -485,7 +488,7 @@ public void testCommitFailure() throws Exception { transformationChain.close(); EasyMock.expectLastCall(); - offsetReader.prematurelyCompleteRequests(); + offsetReader.close(); EasyMock.expectLastCall(); PowerMock.replayAll(); @@ -693,7 +696,7 @@ public Object answer() throws Throwable { transformationChain.close(); EasyMock.expectLastCall(); - offsetReader.prematurelyCompleteRequests(); + offsetReader.close(); EasyMock.expectLastCall(); PowerMock.replayAll(); diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/storage/KafkaOffsetBackingStoreTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/storage/KafkaOffsetBackingStoreTest.java index 24b14bf9969a3..c04363b974605 100644 --- a/connect/runtime/src/test/java/org/apache/kafka/connect/storage/KafkaOffsetBackingStoreTest.java +++ b/connect/runtime/src/test/java/org/apache/kafka/connect/storage/KafkaOffsetBackingStoreTest.java @@ -217,7 +217,6 @@ public Object answer() throws Throwable { store.start(); // Getting from empty store should return nulls - final AtomicBoolean getInvokedAndPassed = new AtomicBoolean(false); Map offsets = store.get(Arrays.asList(TP0_KEY, TP1_KEY)).get(10000, TimeUnit.MILLISECONDS); // Since we didn't read them yet, these will be null assertNull(offsets.get(TP0_KEY)); @@ -244,7 +243,6 @@ public void onCompletion(Throwable error, Void result) { assertTrue(invoked.get()); // Getting data should read to end of our published data and return it - final AtomicBoolean secondGetInvokedAndPassed = new AtomicBoolean(false); offsets = store.get(Arrays.asList(TP0_KEY, TP1_KEY)).get(10000, TimeUnit.MILLISECONDS); assertEquals(TP0_VALUE, offsets.get(TP0_KEY)); assertEquals(TP1_VALUE, offsets.get(TP1_KEY)); @@ -310,7 +308,6 @@ public void onCompletion(Throwable error, Void result) { assertTrue(invoked.get()); // Getting data should read to end of our published data and return it - final AtomicBoolean secondGetInvokedAndPassed = new AtomicBoolean(false); Map offsets = store.get(Arrays.asList(null, TP1_KEY)).get(10000, TimeUnit.MILLISECONDS); assertEquals(TP0_VALUE, offsets.get(null)); assertNull(offsets.get(TP1_KEY)); From 9bda84b989b1a1ad5858a28f6ea706fd036312ae Mon Sep 17 00:00:00 2001 From: Chris Egerton Date: Wed, 23 Oct 2019 11:11:33 -0700 Subject: [PATCH 3/7] KAFKA-9051: Alter cancellation logic to throw exception instead of returning null/empty map --- .../connect/runtime/WorkerSourceTask.java | 7 +- .../storage/CloseableOffsetStorageReader.java | 9 +- .../storage/KafkaOffsetBackingStore.java | 60 +++---------- .../storage/MemoryOffsetBackingStore.java | 86 +++---------------- .../connect/storage/OffsetBackingStore.java | 2 +- .../connect/storage/OffsetReadFuture.java | 31 ------- .../storage/OffsetStorageReaderImpl.java | 20 +++-- .../util/ConvertingFutureCallback.java | 43 +++++++--- .../connect/runtime/WorkerSourceTaskTest.java | 24 ------ 9 files changed, 81 insertions(+), 201 deletions(-) delete mode 100644 connect/runtime/src/main/java/org/apache/kafka/connect/storage/OffsetReadFuture.java diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSourceTask.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSourceTask.java index 9ddcbe1f1bd63..cea99ee79744a 100644 --- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSourceTask.java +++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSourceTask.java @@ -172,10 +172,15 @@ protected void releaseResources() { sourceTaskMetricsGroup.close(); } + @Override + public void cancel() { + super.cancel(); + offsetReader.close(); + } + @Override public void stop() { super.stop(); - offsetReader.close(); stopRequestedLatch.countDown(); synchronized (this) { if (finishedStart) diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/storage/CloseableOffsetStorageReader.java b/connect/runtime/src/main/java/org/apache/kafka/connect/storage/CloseableOffsetStorageReader.java index 65f925dc0ac24..b90273936bfb0 100644 --- a/connect/runtime/src/main/java/org/apache/kafka/connect/storage/CloseableOffsetStorageReader.java +++ b/connect/runtime/src/main/java/org/apache/kafka/connect/storage/CloseableOffsetStorageReader.java @@ -19,14 +19,15 @@ import java.io.Closeable; import java.util.Collection; import java.util.Map; +import java.util.concurrent.Future; public interface CloseableOffsetStorageReader extends Closeable, OffsetStorageReader { /** - * Invoke {@link OffsetReadFuture#forceComplete()} on all outstanding offset read requests, and - * return an empty map and {@code null} in all future calls to {@link #offsets(Collection)} and - * {@link #offset(Map)}, respectively. This is useful for unblocking task threads which need to shut - * down but are blocked on offset reads. + * {@link Future#cancel(boolean) Cancel} all outstanding offset read requests, and throw an + * exception in all current and future calls to {@link #offsets(Collection)} and + * {@link #offset(Map)}. This is useful for unblocking task threads which need to shut down but + * are blocked on offset reads. */ void close(); } diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/storage/KafkaOffsetBackingStore.java b/connect/runtime/src/main/java/org/apache/kafka/connect/storage/KafkaOffsetBackingStore.java index 28fb90ee46c1f..22bcb7a5ae2e6 100644 --- a/connect/runtime/src/main/java/org/apache/kafka/connect/storage/KafkaOffsetBackingStore.java +++ b/connect/runtime/src/main/java/org/apache/kafka/connect/storage/KafkaOffsetBackingStore.java @@ -42,7 +42,6 @@ import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; -import java.util.concurrent.atomic.AtomicBoolean; /** *

@@ -119,8 +118,16 @@ public void stop() { } @Override - public OffsetReadFuture get(final Collection keys) { - KafkaOffsetReadFuture future = new KafkaOffsetReadFuture(keys); + public Future> get(final Collection keys) { + ConvertingFutureCallback> future = new ConvertingFutureCallback>() { + @Override + public Map convert(Void result) { + Map values = new HashMap<>(); + for (ByteBuffer key : keys) + values.put(key, data.get(key)); + return values; + } + }; // This operation may be relatively (but not too) expensive since it always requires checking end offsets, even // if we've already read up to the end. However, it also should not be common (offsets should only be read when // resetting a task). Always requiring that we read to the end is simpler than trying to differentiate when it @@ -222,51 +229,4 @@ public synchronized Void get(long timeout, TimeUnit unit) throws InterruptedExce return null; } } - - private class KafkaOffsetReadFuture extends ConvertingFutureCallback> implements OffsetReadFuture { - private final Collection keys; - private final AtomicBoolean completed; - private transient boolean forceComplete; - - public KafkaOffsetReadFuture(Collection keys) { - super(null); - this.keys = keys; - this.completed = new AtomicBoolean(false); - this.forceComplete = false; - } - - @Override - public Map convert(Void result) { - Map values = new HashMap<>(); - if (!forceComplete) { - for (ByteBuffer key : keys) - values.put(key, data.get(key)); - } - return values; - } - - @Override - public void forceComplete() { - log.debug("Forcing completion of offset read future"); - forceComplete = true; - onCompletion(null, null); - } - - @Override - public Map get() throws InterruptedException, ExecutionException { - log.trace("Blocking on offset read request"); - return super.get(); - } - - @Override - public void onCompletion(Throwable error, Void result) { - if (!completed.getAndSet(true)) { - log.trace("Offset request completed (either normally or forcibly)"); - super.onCompletion(error, result); - } else { - log.trace("Ignoring onCompletion invocation as offset read has already completed"); - } - } - - } } diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/storage/MemoryOffsetBackingStore.java b/connect/runtime/src/main/java/org/apache/kafka/connect/storage/MemoryOffsetBackingStore.java index 3d5e1b246be62..72439e7d687b3 100644 --- a/connect/runtime/src/main/java/org/apache/kafka/connect/storage/MemoryOffsetBackingStore.java +++ b/connect/runtime/src/main/java/org/apache/kafka/connect/storage/MemoryOffsetBackingStore.java @@ -27,12 +27,10 @@ import java.util.HashMap; import java.util.Map; import java.util.concurrent.Callable; -import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; -import java.util.concurrent.TimeoutException; /** * Implementation of OffsetBackingStore that doesn't actually persist any data. To ensure this @@ -77,8 +75,18 @@ public void stop() { } @Override - public OffsetReadFuture get(final Collection keys) { - return new MemoryOffsetReadFuture(keys); + public Future> get(final Collection keys) { + return executor.submit(new Callable>() { + @Override + public Map call() throws Exception { + Map result = new HashMap<>(); + for (ByteBuffer key : keys) { + result.put(key, data.get(key)); + } + return result; + } + }); + } @Override @@ -102,74 +110,4 @@ public Void call() throws Exception { protected void save() { } - - private class MemoryOffsetReadFuture implements OffsetReadFuture { - private final Map result; - private final Collection keys; - private final CountDownLatch completed; - private final Future underlyingOffsetFuture; - - public MemoryOffsetReadFuture(Collection keys) { - this.result = new HashMap<>(); - this.keys = keys; - this.completed = new CountDownLatch(1); - this.underlyingOffsetFuture = executor.submit(new Runnable() { - @Override - public void run() { - synchronized (this) { - if (isDone()) { - return; - } - collectResults(); - completed.countDown(); - } - } - }); - } - - @Override - public void forceComplete() { - synchronized (this) { - if (isDone()) { - return; - } - completed.countDown(); - } - } - - @Override - public boolean cancel(boolean mayInterruptIfRunning) { - return underlyingOffsetFuture.cancel(mayInterruptIfRunning); - } - - @Override - public boolean isCancelled() { - return underlyingOffsetFuture.isCancelled(); - } - - @Override - public boolean isDone() { - return completed.getCount() == 0; - } - - @Override - public Map get() throws InterruptedException { - completed.await(); - return result; - } - - @Override - public Map get(long timeout, TimeUnit unit) - throws TimeoutException, InterruptedException { - if (!completed.await(timeout, unit)) - throw new TimeoutException("Timed out waiting for future"); - return result; - } - - private void collectResults() { - for (ByteBuffer key : keys) { - result.put(key, data.get(key)); - } - } - } } diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/storage/OffsetBackingStore.java b/connect/runtime/src/main/java/org/apache/kafka/connect/storage/OffsetBackingStore.java index a9002320b4e66..1e4375b7d8eff 100644 --- a/connect/runtime/src/main/java/org/apache/kafka/connect/storage/OffsetBackingStore.java +++ b/connect/runtime/src/main/java/org/apache/kafka/connect/storage/OffsetBackingStore.java @@ -55,7 +55,7 @@ public interface OffsetBackingStore { * @param keys list of keys to look up * @return future for the resulting map from key to value */ - OffsetReadFuture get(Collection keys); + Future> get(Collection keys); /** * Set the specified keys and values. diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/storage/OffsetReadFuture.java b/connect/runtime/src/main/java/org/apache/kafka/connect/storage/OffsetReadFuture.java deleted file mode 100644 index 142acca0d189a..0000000000000 --- a/connect/runtime/src/main/java/org/apache/kafka/connect/storage/OffsetReadFuture.java +++ /dev/null @@ -1,31 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.apache.kafka.connect.storage; - - -import java.nio.ByteBuffer; -import java.util.Map; -import java.util.concurrent.Future; - -public interface OffsetReadFuture extends Future> { - - /** - * Without {@link Future#cancel(boolean) cancelling} the operation, immediately return an empty - * map to any threads which are waiting for this future to complete. - */ - void forceComplete(); -} \ No newline at end of file diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/storage/OffsetStorageReaderImpl.java b/connect/runtime/src/main/java/org/apache/kafka/connect/storage/OffsetStorageReaderImpl.java index e69d5130afd47..6271ece3782ee 100644 --- a/connect/runtime/src/main/java/org/apache/kafka/connect/storage/OffsetStorageReaderImpl.java +++ b/connect/runtime/src/main/java/org/apache/kafka/connect/storage/OffsetStorageReaderImpl.java @@ -29,6 +29,8 @@ import java.util.HashSet; import java.util.Map; import java.util.Set; +import java.util.concurrent.CancellationException; +import java.util.concurrent.Future; import java.util.concurrent.atomic.AtomicBoolean; /** @@ -44,7 +46,7 @@ public class OffsetStorageReaderImpl implements CloseableOffsetStorageReader { private final Converter keyConverter; private final Converter valueConverter; private final AtomicBoolean closed; - private final Set offsetReadFutures; + private final Set>> offsetReadFutures; public OffsetStorageReaderImpl(OffsetBackingStore backingStore, String namespace, Converter keyConverter, Converter valueConverter) { @@ -83,10 +85,13 @@ public Map, Map> offsets(Collection serialized value from backing store Map raw; try { - OffsetReadFuture offsetReadFuture; + Future> offsetReadFuture; synchronized (offsetReadFutures) { if (closed.get()) { - return Collections.emptyMap(); + throw new ConnectException( + "Offset reader is closed. This is likely because the task has already been " + + "scheduled to stop but has taken longer than the graceful shutdown " + + "period to do so."); } offsetReadFuture = backingStore.get(serializedToOriginal.keySet()); offsetReadFutures.add(offsetReadFuture); @@ -94,6 +99,11 @@ public Map, Map> offsets(Collection Map, Map> offsets(Collection> offsetReadFuture : offsetReadFutures) { + offsetReadFuture.cancel(true); } offsetReadFutures.clear(); } diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/util/ConvertingFutureCallback.java b/connect/runtime/src/main/java/org/apache/kafka/connect/util/ConvertingFutureCallback.java index d5abed9385cc0..7eb255effc80b 100644 --- a/connect/runtime/src/main/java/org/apache/kafka/connect/util/ConvertingFutureCallback.java +++ b/connect/runtime/src/main/java/org/apache/kafka/connect/util/ConvertingFutureCallback.java @@ -16,6 +16,7 @@ */ package org.apache.kafka.connect.util; +import java.util.concurrent.CancellationException; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutionException; import java.util.concurrent.Future; @@ -24,10 +25,15 @@ public abstract class ConvertingFutureCallback implements Callback, Future { - private Callback underlying; - private CountDownLatch finishedLatch; - private T result = null; - private Throwable exception = null; + private final Callback underlying; + private final CountDownLatch finishedLatch; + private volatile T result = null; + private volatile Throwable exception = null; + private volatile boolean cancelled = false; + + public ConvertingFutureCallback() { + this(null); + } public ConvertingFutureCallback(Callback underlying) { this.underlying = underlying; @@ -38,21 +44,33 @@ public ConvertingFutureCallback(Callback underlying) { @Override public void onCompletion(Throwable error, U result) { - this.exception = error; - this.result = convert(result); - if (underlying != null) - underlying.onCompletion(error, this.result); - finishedLatch.countDown(); + synchronized (this) { + if (isDone()) { + return; + } + this.exception = error; + this.result = convert(result); + if (underlying != null) + underlying.onCompletion(error, this.result); + finishedLatch.countDown(); + } } @Override public boolean cancel(boolean b) { - return false; + synchronized (this) { + if (isDone()) { + return false; + } + this.cancelled = true; + finishedLatch.countDown(); + return true; + } } @Override public boolean isCancelled() { - return false; + return cancelled; } @Override @@ -75,6 +93,9 @@ public T get(long l, TimeUnit timeUnit) } private T result() throws ExecutionException { + if (cancelled) { + throw new CancellationException(); + } if (exception != null) { throw new ExecutionException(exception); } diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSourceTaskTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSourceTaskTest.java index 6ff816a1e6805..9c1c4735c7a5e 100644 --- a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSourceTaskTest.java +++ b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSourceTaskTest.java @@ -193,9 +193,6 @@ public Void answer() throws Throwable { statusListener.onShutdown(taskId); EasyMock.expectLastCall(); - offsetReader.close(); - EasyMock.expectLastCall(); - PowerMock.replayAll(); workerTask.initialize(TASK_CONFIG); @@ -241,9 +238,6 @@ public void testPause() throws Exception { transformationChain.close(); EasyMock.expectLastCall(); - offsetReader.close(); - EasyMock.expectLastCall(); - PowerMock.replayAll(); workerTask.initialize(TASK_CONFIG); @@ -293,9 +287,6 @@ public void testPollsInBackground() throws Exception { transformationChain.close(); EasyMock.expectLastCall(); - offsetReader.close(); - EasyMock.expectLastCall(); - PowerMock.replayAll(); workerTask.initialize(TASK_CONFIG); @@ -345,9 +336,6 @@ public List answer() throws Throwable { transformationChain.close(); EasyMock.expectLastCall(); - offsetReader.close(); - EasyMock.expectLastCall(); - PowerMock.replayAll(); workerTask.initialize(TASK_CONFIG); @@ -392,9 +380,6 @@ public void testPollReturnsNoRecords() throws Exception { transformationChain.close(); EasyMock.expectLastCall(); - offsetReader.close(); - EasyMock.expectLastCall(); - PowerMock.replayAll(); workerTask.initialize(TASK_CONFIG); @@ -440,9 +425,6 @@ public void testCommit() throws Exception { transformationChain.close(); EasyMock.expectLastCall(); - offsetReader.close(); - EasyMock.expectLastCall(); - PowerMock.replayAll(); workerTask.initialize(TASK_CONFIG); @@ -488,9 +470,6 @@ public void testCommitFailure() throws Exception { transformationChain.close(); EasyMock.expectLastCall(); - offsetReader.close(); - EasyMock.expectLastCall(); - PowerMock.replayAll(); workerTask.initialize(TASK_CONFIG); @@ -696,9 +675,6 @@ public Object answer() throws Throwable { transformationChain.close(); EasyMock.expectLastCall(); - offsetReader.close(); - EasyMock.expectLastCall(); - PowerMock.replayAll(); workerTask.initialize(TASK_CONFIG); From 7f2f21be25843b23277cf6d7fe3bb54ba6ab892c Mon Sep 17 00:00:00 2001 From: Chris Egerton Date: Wed, 23 Oct 2019 11:53:02 -0700 Subject: [PATCH 4/7] KAFKA-9051: Add unit test --- .../connect/runtime/WorkerSourceTaskTest.java | 14 ++++++++++++++ 1 file changed, 14 insertions(+) diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSourceTaskTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSourceTaskTest.java index 9c1c4735c7a5e..ea3716d207111 100644 --- a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSourceTaskTest.java +++ b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSourceTaskTest.java @@ -693,6 +693,20 @@ public Object answer() throws Throwable { PowerMock.verifyAll(); } + @Test + public void testCancel() { + createWorkerTask(); + + offsetReader.close(); + PowerMock.expectLastCall(); + + PowerMock.replayAll(); + + workerTask.cancel(); + + PowerMock.verifyAll(); + } + @Test public void testMetricsGroup() { SourceTaskMetricsGroup group = new SourceTaskMetricsGroup(taskId, metrics); From 746b46b3053279a37e8b99d87cd1cee1ea205456 Mon Sep 17 00:00:00 2001 From: Chris Egerton Date: Wed, 23 Oct 2019 12:47:37 -0700 Subject: [PATCH 5/7] KAFKA-9051: Add unit tests for ConvertingFutureCallback class --- .../util/ConvertingFutureCallback.java | 13 +- .../util/ConvertingFutureCallbackTest.java | 208 ++++++++++++++++++ 2 files changed, 219 insertions(+), 2 deletions(-) create mode 100644 connect/runtime/src/test/java/org/apache/kafka/connect/util/ConvertingFutureCallbackTest.java diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/util/ConvertingFutureCallback.java b/connect/runtime/src/main/java/org/apache/kafka/connect/util/ConvertingFutureCallback.java index 7eb255effc80b..193dcb81e2bcb 100644 --- a/connect/runtime/src/main/java/org/apache/kafka/connect/util/ConvertingFutureCallback.java +++ b/connect/runtime/src/main/java/org/apache/kafka/connect/util/ConvertingFutureCallback.java @@ -48,8 +48,13 @@ public void onCompletion(Throwable error, U result) { if (isDone()) { return; } - this.exception = error; - this.result = convert(result); + + if (error != null) { + this.exception = error; + } else { + this.result = convert(result); + } + if (underlying != null) underlying.onCompletion(error, this.result); finishedLatch.countDown(); @@ -58,6 +63,10 @@ public void onCompletion(Throwable error, U result) { @Override public boolean cancel(boolean b) { + if (!b) { + return false; + } + synchronized (this) { if (isDone()) { return false; diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/util/ConvertingFutureCallbackTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/util/ConvertingFutureCallbackTest.java new file mode 100644 index 0000000000000..f212ef964bc18 --- /dev/null +++ b/connect/runtime/src/test/java/org/apache/kafka/connect/util/ConvertingFutureCallbackTest.java @@ -0,0 +1,208 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.kafka.connect.util; + +import org.junit.Test; + +import java.util.concurrent.CancellationException; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; + +public class ConvertingFutureCallbackTest { + + @Test + public void shouldConvertBeforeGetOnSuccessfulCompletion() throws Exception { + final Object expectedConversion = new Object(); + TestConvertingFutureCallback testCallback = new TestConvertingFutureCallback(); + testCallback.onCompletion(null, expectedConversion); + assertEquals(1, testCallback.numberOfConversions()); + assertEquals(expectedConversion, testCallback.get()); + } + + @Test + public void shouldConvertOnlyOnceBeforeGetOnSuccessfulCompletion() throws Exception { + final Object expectedConversion = new Object(); + TestConvertingFutureCallback testCallback = new TestConvertingFutureCallback(); + testCallback.onCompletion(null, expectedConversion); + testCallback.onCompletion(null, 69); + testCallback.cancel(true); + testCallback.onCompletion(new RuntimeException(), null); + assertEquals(1, testCallback.numberOfConversions()); + assertEquals(expectedConversion, testCallback.get()); + } + + @Test + public void shouldNotConvertBeforeGetOnFailedCompletion() throws Exception { + final Throwable expectedError = new Throwable(); + TestConvertingFutureCallback testCallback = new TestConvertingFutureCallback(); + testCallback.onCompletion(expectedError, null); + assertEquals(0, testCallback.numberOfConversions()); + try { + testCallback.get(); + } catch (ExecutionException e) { + assertEquals(expectedError, e.getCause()); + } + } + + @Test + public void shouldRecordOnlyFirstErrorBeforeGetOnFailedCompletion() throws Exception { + final Throwable expectedError = new Throwable(); + TestConvertingFutureCallback testCallback = new TestConvertingFutureCallback(); + testCallback.onCompletion(expectedError, null); + testCallback.onCompletion(new RuntimeException(), null); + testCallback.cancel(true); + testCallback.onCompletion(null, "420"); + assertEquals(0, testCallback.numberOfConversions()); + try { + testCallback.get(); + } catch (ExecutionException e) { + assertEquals(expectedError, e.getCause()); + } + } + + @Test(expected = CancellationException.class) + public void shouldCancelBeforeGetIfMayCancelWhileRunning() throws Exception { + TestConvertingFutureCallback testCallback = new TestConvertingFutureCallback(); + assertTrue(testCallback.cancel(true)); + testCallback.get(); + } + + @Test + public void shouldNotCancelBeforeGetIfMayNotCancelWhileRunning() throws Exception { + TestConvertingFutureCallback testCallback = new TestConvertingFutureCallback(); + assertFalse(testCallback.cancel(false)); + final Object expectedConversion = new Object(); + testCallback.onCompletion(null, expectedConversion); + assertEquals(1, testCallback.numberOfConversions()); + assertEquals(expectedConversion, testCallback.get()); + } + + @Test + public void shouldBlockUntilSuccessfulCompletion() throws Exception { + AtomicReference testThreadException = new AtomicReference<>(); + TestConvertingFutureCallback testCallback = new TestConvertingFutureCallback(); + final Object expectedConversion = new Object(); + runSeparateTestThread(() -> { + try { + testCallback.waitForGet(); + testCallback.onCompletion(null, expectedConversion); + } catch (Exception e) { + testThreadException.compareAndSet(null, e); + } + }); + assertFalse(testCallback.isDone()); + assertEquals(expectedConversion, testCallback.get()); + assertEquals(1, testCallback.numberOfConversions()); + assertTrue(testCallback.isDone()); + if (testThreadException.get() != null) { + throw testThreadException.get(); + } + } + + @Test + public void shouldBlockUntilFailedCompletion() throws Exception { + AtomicReference testThreadException = new AtomicReference<>(); + TestConvertingFutureCallback testCallback = new TestConvertingFutureCallback(); + final Throwable expectedError = new Throwable(); + runSeparateTestThread(() -> { + try { + testCallback.waitForGet(); + testCallback.onCompletion(expectedError, null); + } catch (Exception e) { + testThreadException.compareAndSet(null, e); + } + }); + assertFalse(testCallback.isDone()); + try { + testCallback.get(); + } catch (ExecutionException e) { + assertEquals(expectedError, e.getCause()); + } + assertEquals(0, testCallback.numberOfConversions()); + assertTrue(testCallback.isDone()); + if (testThreadException.get() != null) { + throw testThreadException.get(); + } + } + + @Test(expected = CancellationException.class) + public void shouldBlockUntilCancellation() throws Exception { + AtomicReference testThreadException = new AtomicReference<>(); + TestConvertingFutureCallback testCallback = new TestConvertingFutureCallback(); + runSeparateTestThread(() -> { + try { + testCallback.waitForGet(); + testCallback.cancel(true); + } catch (Exception e) { + testThreadException.compareAndSet(null, e); + } + }); + assertFalse(testCallback.isDone()); + testCallback.get(); + if (testThreadException.get() != null) { + throw testThreadException.get(); + } + } + + protected static void runSeparateTestThread(Runnable task) { + Thread t = new Thread(task); + t.setDaemon(true); + t.start(); + } + + protected static class TestConvertingFutureCallback extends ConvertingFutureCallback { + private AtomicInteger numberOfConversions = new AtomicInteger(); + private CountDownLatch getInvoked = new CountDownLatch(1); + + public int numberOfConversions() { + return numberOfConversions.get(); + } + + public void waitForGet() throws InterruptedException { + getInvoked.await(); + } + + @Override + public Object convert(Object result) { + numberOfConversions.incrementAndGet(); + return result; + } + + @Override + public Object get() throws InterruptedException, ExecutionException { + getInvoked.countDown(); + return super.get(); + } + + @Override + public Object get( + long duration, + TimeUnit unit + ) throws InterruptedException, ExecutionException, TimeoutException { + getInvoked.countDown(); + return super.get(duration, unit); + } + } +} From 3828abfb17c4586bf140ebe40d5826b0a1dc4eee Mon Sep 17 00:00:00 2001 From: Chris Egerton Date: Wed, 23 Oct 2019 14:15:17 -0700 Subject: [PATCH 6/7] KAFKA-9051: Address review comments --- .../util/ConvertingFutureCallback.java | 22 +++--- .../util/ConvertingFutureCallbackTest.java | 68 ++++++++++++++----- 2 files changed, 65 insertions(+), 25 deletions(-) diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/util/ConvertingFutureCallback.java b/connect/runtime/src/main/java/org/apache/kafka/connect/util/ConvertingFutureCallback.java index 193dcb81e2bcb..e15c38ea4c4ae 100644 --- a/connect/runtime/src/main/java/org/apache/kafka/connect/util/ConvertingFutureCallback.java +++ b/connect/runtime/src/main/java/org/apache/kafka/connect/util/ConvertingFutureCallback.java @@ -16,6 +16,8 @@ */ package org.apache.kafka.connect.util; +import org.apache.kafka.connect.errors.ConnectException; + import java.util.concurrent.CancellationException; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutionException; @@ -62,19 +64,23 @@ public void onCompletion(Throwable error, U result) { } @Override - public boolean cancel(boolean b) { - if (!b) { - return false; - } - + public boolean cancel(boolean mayInterruptIfRunning) { synchronized (this) { if (isDone()) { return false; } - this.cancelled = true; - finishedLatch.countDown(); - return true; + if (mayInterruptIfRunning) { + this.cancelled = true; + finishedLatch.countDown(); + return true; + } + } + try { + finishedLatch.await(); + } catch (InterruptedException e) { + throw new ConnectException("Interrupted while waiting for task to complete", e); } + return false; } @Override diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/util/ConvertingFutureCallbackTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/util/ConvertingFutureCallbackTest.java index f212ef964bc18..9535003c863c3 100644 --- a/connect/runtime/src/test/java/org/apache/kafka/connect/util/ConvertingFutureCallbackTest.java +++ b/connect/runtime/src/test/java/org/apache/kafka/connect/util/ConvertingFutureCallbackTest.java @@ -16,11 +16,14 @@ */ package org.apache.kafka.connect.util; +import org.junit.Before; import org.junit.Test; import java.util.concurrent.CancellationException; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicInteger; @@ -29,8 +32,16 @@ import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; public class ConvertingFutureCallbackTest { + + private ExecutorService executor; + + @Before + public void setup() { + executor = Executors.newSingleThreadExecutor(); + } @Test public void shouldConvertBeforeGetOnSuccessfulCompletion() throws Exception { @@ -61,6 +72,7 @@ public void shouldNotConvertBeforeGetOnFailedCompletion() throws Exception { assertEquals(0, testCallback.numberOfConversions()); try { testCallback.get(); + fail("Expected ExecutionException"); } catch (ExecutionException e) { assertEquals(expectedError, e.getCause()); } @@ -77,6 +89,7 @@ public void shouldRecordOnlyFirstErrorBeforeGetOnFailedCompletion() throws Excep assertEquals(0, testCallback.numberOfConversions()); try { testCallback.get(); + fail("Expected ExecutionException"); } catch (ExecutionException e) { assertEquals(expectedError, e.getCause()); } @@ -88,23 +101,13 @@ public void shouldCancelBeforeGetIfMayCancelWhileRunning() throws Exception { assertTrue(testCallback.cancel(true)); testCallback.get(); } - - @Test - public void shouldNotCancelBeforeGetIfMayNotCancelWhileRunning() throws Exception { - TestConvertingFutureCallback testCallback = new TestConvertingFutureCallback(); - assertFalse(testCallback.cancel(false)); - final Object expectedConversion = new Object(); - testCallback.onCompletion(null, expectedConversion); - assertEquals(1, testCallback.numberOfConversions()); - assertEquals(expectedConversion, testCallback.get()); - } @Test public void shouldBlockUntilSuccessfulCompletion() throws Exception { AtomicReference testThreadException = new AtomicReference<>(); TestConvertingFutureCallback testCallback = new TestConvertingFutureCallback(); final Object expectedConversion = new Object(); - runSeparateTestThread(() -> { + executor.submit(() -> { try { testCallback.waitForGet(); testCallback.onCompletion(null, expectedConversion); @@ -126,7 +129,7 @@ public void shouldBlockUntilFailedCompletion() throws Exception { AtomicReference testThreadException = new AtomicReference<>(); TestConvertingFutureCallback testCallback = new TestConvertingFutureCallback(); final Throwable expectedError = new Throwable(); - runSeparateTestThread(() -> { + executor.submit(() -> { try { testCallback.waitForGet(); testCallback.onCompletion(expectedError, null); @@ -137,6 +140,7 @@ public void shouldBlockUntilFailedCompletion() throws Exception { assertFalse(testCallback.isDone()); try { testCallback.get(); + fail("Expected ExecutionException"); } catch (ExecutionException e) { assertEquals(expectedError, e.getCause()); } @@ -151,7 +155,7 @@ public void shouldBlockUntilFailedCompletion() throws Exception { public void shouldBlockUntilCancellation() throws Exception { AtomicReference testThreadException = new AtomicReference<>(); TestConvertingFutureCallback testCallback = new TestConvertingFutureCallback(); - runSeparateTestThread(() -> { + executor.submit(() -> { try { testCallback.waitForGet(); testCallback.cancel(true); @@ -166,15 +170,35 @@ public void shouldBlockUntilCancellation() throws Exception { } } - protected static void runSeparateTestThread(Runnable task) { - Thread t = new Thread(task); - t.setDaemon(true); - t.start(); + @Test + public void shouldNotCancelIfMayNotCancelWhileRunning() throws Exception { + AtomicReference testThreadException = new AtomicReference<>(); + TestConvertingFutureCallback testCallback = new TestConvertingFutureCallback(); + final Object expectedConversion = new Object(); + executor.submit(() -> { + try { + testCallback.waitForCancel(); + testCallback.onCompletion(null, expectedConversion); + } catch (Exception e) { + testThreadException.compareAndSet(null, e); + } + }); + assertFalse(testCallback.isCancelled()); + assertFalse(testCallback.isDone()); + testCallback.cancel(false); + assertFalse(testCallback.isCancelled()); + assertTrue(testCallback.isDone()); + assertEquals(expectedConversion, testCallback.get()); + assertEquals(1, testCallback.numberOfConversions()); + if (testThreadException.get() != null) { + throw testThreadException.get(); + } } protected static class TestConvertingFutureCallback extends ConvertingFutureCallback { private AtomicInteger numberOfConversions = new AtomicInteger(); private CountDownLatch getInvoked = new CountDownLatch(1); + private CountDownLatch cancelInvoked = new CountDownLatch(1); public int numberOfConversions() { return numberOfConversions.get(); @@ -183,6 +207,10 @@ public int numberOfConversions() { public void waitForGet() throws InterruptedException { getInvoked.await(); } + + public void waitForCancel() throws InterruptedException { + cancelInvoked.await(); + } @Override public Object convert(Object result) { @@ -204,5 +232,11 @@ public Object get( getInvoked.countDown(); return super.get(duration, unit); } + + @Override + public boolean cancel(boolean mayInterruptIfRunning) { + cancelInvoked.countDown(); + return super.cancel(mayInterruptIfRunning); + } } } From 28c156e3c79d698a18e193261a812d1cb44f25ff Mon Sep 17 00:00:00 2001 From: Chris Egerton Date: Tue, 19 Nov 2019 14:14:55 -0800 Subject: [PATCH 7/7] KAFKA-9051: Address review comments --- .../connect/storage/OffsetStorageReaderImpl.java | 15 ++++++++++----- 1 file changed, 10 insertions(+), 5 deletions(-) diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/storage/OffsetStorageReaderImpl.java b/connect/runtime/src/main/java/org/apache/kafka/connect/storage/OffsetStorageReaderImpl.java index 6271ece3782ee..a1eea43103a39 100644 --- a/connect/runtime/src/main/java/org/apache/kafka/connect/storage/OffsetStorageReaderImpl.java +++ b/connect/runtime/src/main/java/org/apache/kafka/connect/storage/OffsetStorageReaderImpl.java @@ -142,12 +142,17 @@ public Map, Map> offsets(Collection> offsetReadFuture : offsetReadFutures) { - offsetReadFuture.cancel(true); + if (!closed.getAndSet(true)) { + synchronized (offsetReadFutures) { + for (Future> offsetReadFuture : offsetReadFutures) { + try { + offsetReadFuture.cancel(true); + } catch (Throwable t) { + log.error("Failed to cancel offset read future", t); + } + } + offsetReadFutures.clear(); } - offsetReadFutures.clear(); } } }