From 6527094449529ecccd1e8cf30ae0526514eebc33 Mon Sep 17 00:00:00 2001 From: Yash Mayya Date: Sat, 18 Feb 2023 17:17:33 +0530 Subject: [PATCH 1/3] Use an exponential backoff retry mechanism while reconfiguring connector tasks --- .../common/utils/ExponentialBackoff.java | 11 +++-- .../distributed/DistributedHerder.java | 31 +++++++++--- .../distributed/DistributedHerderTest.java | 48 +++++++++++++++++++ 3 files changed, 80 insertions(+), 10 deletions(-) diff --git a/clients/src/main/java/org/apache/kafka/common/utils/ExponentialBackoff.java b/clients/src/main/java/org/apache/kafka/common/utils/ExponentialBackoff.java index 7550184ba39d0..30a6bf8e7fed8 100644 --- a/clients/src/main/java/org/apache/kafka/common/utils/ExponentialBackoff.java +++ b/clients/src/main/java/org/apache/kafka/common/utils/ExponentialBackoff.java @@ -20,12 +20,15 @@ import java.util.concurrent.ThreadLocalRandom; /** - * An utility class for keeping the parameters and providing the value of exponential + * A utility class for keeping the parameters and providing the value of exponential * retry backoff, exponential reconnect backoff, exponential timeout, etc. + *

* The formula is: - * Backoff(attempts) = random(1 - jitter, 1 + jitter) * initialInterval * multiplier ^ attempts - * If initialInterval is greater or equal than maxInterval, a constant backoff of will be provided - * This class is thread-safe + * 

Backoff(attempts) = random(1 - jitter, 1 + jitter) * initialInterval * multiplier ^ attempts
+ * If {@code initialInterval} is greater than or equal to {@code maxInterval}, a constant backoff of + * {@code initialInterval} will be provided. + *

+ * This class is thread-safe. */ public class ExponentialBackoff { private final int multiplier; diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedHerder.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedHerder.java index b68cc443ea742..8b33f66be79c1 100644 --- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedHerder.java +++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedHerder.java @@ -26,6 +26,7 @@ import org.apache.kafka.common.metrics.stats.CumulativeSum; import org.apache.kafka.common.metrics.stats.Max; import org.apache.kafka.common.utils.Exit; +import org.apache.kafka.common.utils.ExponentialBackoff; import org.apache.kafka.common.utils.LogContext; import org.apache.kafka.common.utils.ThreadUtils; import org.apache.kafka.common.utils.Time; @@ -147,7 +148,8 @@ public class DistributedHerder extends AbstractHerder implements Runnable { private static final long FORWARD_REQUEST_SHUTDOWN_TIMEOUT_MS = TimeUnit.SECONDS.toMillis(10); private static final long START_AND_STOP_SHUTDOWN_TIMEOUT_MS = TimeUnit.SECONDS.toMillis(1); - private static final long RECONFIGURE_CONNECTOR_TASKS_BACKOFF_MS = 250; + private static final long RECONFIGURE_CONNECTOR_TASKS_BACKOFF_INITIAL_MS = 250; + private static final long RECONFIGURE_CONNECTOR_TASKS_BACKOFF_MAX_MS = 60000; private static final long CONFIG_TOPIC_WRITE_PRIVILEGES_BACKOFF_MS = 250; private static final int START_STOP_THREAD_POOL_SIZE = 8; private static final short BACKOFF_RETRIES = 5; @@ -1090,7 +1092,11 @@ public void requestTaskReconfiguration(final String connName) { addRequest( () -> { - reconfigureConnectorTasksWithRetry(time.milliseconds(), connName); + ExponentialBackoff exponentialBackoff = new ExponentialBackoff( + RECONFIGURE_CONNECTOR_TASKS_BACKOFF_INITIAL_MS, + 2, RECONFIGURE_CONNECTOR_TASKS_BACKOFF_MAX_MS, + 0); + reconfigureConnectorTasksWithRetry(time.milliseconds(), connName, exponentialBackoff, 0); return null; }, (error, result) -> { @@ -1827,10 +1833,14 @@ private void startConnector(String connectorName, Callback callback) { if (newState == TargetState.STARTED) { addRequest( () -> { + ExponentialBackoff exponentialBackoff = new ExponentialBackoff( + RECONFIGURE_CONNECTOR_TASKS_BACKOFF_INITIAL_MS, + 2, RECONFIGURE_CONNECTOR_TASKS_BACKOFF_MAX_MS, + 0); // Request configuration since this could be a brand new connector. However, also only update those // task configs if they are actually different from the existing ones to avoid unnecessary updates when this is // just restoring an existing connector. - reconfigureConnectorTasksWithRetry(time.milliseconds(), connectorName); + reconfigureConnectorTasksWithRetry(time.milliseconds(), connectorName, exponentialBackoff, 0); callback.onCompletion(null, null); return null; }, @@ -1870,7 +1880,16 @@ private Callable getConnectorStoppingCallable(final String connectorName) }; } - private void reconfigureConnectorTasksWithRetry(long initialRequestTime, final String connName) { + /** + * Request task configs from the connector and write them to the config storage in case the configs are detected to + * have changed. This method retries infinitely in case of any errors. + * + * @param initialRequestTime the time in milliseconds when the original request was made (i.e. before any retries) + * @param connName the name of the connector + * @param exponentialBackoff {@link ExponentialBackoff} used to calculate the retry backoff duration + * @param attempts the number of retry attempts that have been made + */ + private void reconfigureConnectorTasksWithRetry(long initialRequestTime, final String connName, ExponentialBackoff exponentialBackoff, int attempts) { reconfigureConnector(connName, (error, result) -> { // If we encountered an error, we don't have much choice but to just retry. If we don't, we could get // stuck with a connector that thinks it has generated tasks, but wasn't actually successful and therefore @@ -1882,9 +1901,9 @@ private void reconfigureConnectorTasksWithRetry(long initialRequestTime, final S } else { log.error("Failed to reconfigure connector's tasks ({}), retrying after backoff:", connName, error); } - addRequest(RECONFIGURE_CONNECTOR_TASKS_BACKOFF_MS, + addRequest(exponentialBackoff.backoff(attempts), () -> { - reconfigureConnectorTasksWithRetry(initialRequestTime, connName); + reconfigureConnectorTasksWithRetry(initialRequestTime, connName, exponentialBackoff, attempts + 1); return null; }, (err, res) -> { if (err != null) { diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/distributed/DistributedHerderTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/distributed/DistributedHerderTest.java index 99dffe17d9d3e..a02d5b7a80382 100644 --- a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/distributed/DistributedHerderTest.java +++ b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/distributed/DistributedHerderTest.java @@ -3694,6 +3694,54 @@ public void shouldThrowWhenStartAndStopExecutorThrowsRejectedExecutionExceptionA PowerMock.verifyAll(); } + @Test + public void testTaskReconfigurationRetries() { + EasyMock.expect(member.memberId()).andStubReturn("leader"); + EasyMock.expect(member.currentProtocolVersion()).andStubReturn(CONNECT_PROTOCOL_V0); + expectRebalance(1, Collections.emptyList(), Collections.emptyList(), true); + expectConfigRefreshAndSnapshot(SNAPSHOT); + + // end of initial tick + member.poll(EasyMock.anyInt()); + PowerMock.expectLastCall(); + + member.wakeup(); + PowerMock.expectLastCall(); + + // second tick + member.ensureActive(); + PowerMock.expectLastCall(); + + EasyMock.expect(worker.isRunning(CONN1)).andReturn(true).anyTimes(); + EasyMock.expect(worker.getPlugins()).andReturn(plugins).anyTimes(); + + SinkConnectorConfig sinkConnectorConfig = new SinkConnectorConfig(plugins, CONN1_CONFIG); + EasyMock.expect(worker.connectorTaskConfigs(CONN1, sinkConnectorConfig)) + .andThrow(new ConnectException("Failed to generate task configs")).anyTimes(); + + // task reconfiguration request with initial retry backoff + member.poll(EasyMock.eq(250L)); + PowerMock.expectLastCall(); + + member.ensureActive(); + PowerMock.expectLastCall(); + + // task reconfiguration request with double the initial retry backoff + member.poll(EasyMock.eq(500L)); + PowerMock.expectLastCall(); + + PowerMock.replayAll(); + + // initial tick + herder.tick(); + herder.requestTaskReconfiguration(CONN1); + // process the task reconfiguration request in this tick + herder.tick(); + // advance the time by 250ms so that the task reconfiguration request with initial retry backoff is processed + time.sleep(250); + herder.tick(); + } + private void expectRebalance(final long offset, final List assignedConnectors, final List assignedTasks) { From 724b714780bb19e91943ae5223d1f6a9f0754d6e Mon Sep 17 00:00:00 2001 From: Yash Mayya Date: Tue, 21 Feb 2023 17:56:52 +0530 Subject: [PATCH 2/3] Minor refactor to centralize instantiation of ExponentialBackoff --- .../distributed/DistributedHerder.java | 36 ++++++++++++------- 1 file changed, 23 insertions(+), 13 deletions(-) diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedHerder.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedHerder.java index 8b33f66be79c1..9c07f36944f08 100644 --- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedHerder.java +++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedHerder.java @@ -1092,11 +1092,7 @@ public void requestTaskReconfiguration(final String connName) { addRequest( () -> { - ExponentialBackoff exponentialBackoff = new ExponentialBackoff( - RECONFIGURE_CONNECTOR_TASKS_BACKOFF_INITIAL_MS, - 2, RECONFIGURE_CONNECTOR_TASKS_BACKOFF_MAX_MS, - 0); - reconfigureConnectorTasksWithRetry(time.milliseconds(), connName, exponentialBackoff, 0); + reconfigureConnectorTasksWithRetry(time.milliseconds(), connName); return null; }, (error, result) -> { @@ -1833,14 +1829,10 @@ private void startConnector(String connectorName, Callback callback) { if (newState == TargetState.STARTED) { addRequest( () -> { - ExponentialBackoff exponentialBackoff = new ExponentialBackoff( - RECONFIGURE_CONNECTOR_TASKS_BACKOFF_INITIAL_MS, - 2, RECONFIGURE_CONNECTOR_TASKS_BACKOFF_MAX_MS, - 0); // Request configuration since this could be a brand new connector. However, also only update those // task configs if they are actually different from the existing ones to avoid unnecessary updates when this is // just restoring an existing connector. - reconfigureConnectorTasksWithRetry(time.milliseconds(), connectorName, exponentialBackoff, 0); + reconfigureConnectorTasksWithRetry(time.milliseconds(), connectorName); callback.onCompletion(null, null); return null; }, @@ -1882,14 +1874,32 @@ private Callable getConnectorStoppingCallable(final String connectorName) /** * Request task configs from the connector and write them to the config storage in case the configs are detected to - * have changed. This method retries infinitely in case of any errors. + * have changed. This method retries infinitely with exponential backoff in case of any errors. The initial backoff + * used is {@link #RECONFIGURE_CONNECTOR_TASKS_BACKOFF_INITIAL_MS} with a multiplier of 2 and a maximum backoff of + * {@link #RECONFIGURE_CONNECTOR_TASKS_BACKOFF_MAX_MS} + * + * @param initialRequestTime the time in milliseconds when the original request was made (i.e. before any retries) + * @param connName the name of the connector + */ + private void reconfigureConnectorTasksWithRetry(long initialRequestTime, final String connName) { + ExponentialBackoff exponentialBackoff = new ExponentialBackoff( + RECONFIGURE_CONNECTOR_TASKS_BACKOFF_INITIAL_MS, + 2, RECONFIGURE_CONNECTOR_TASKS_BACKOFF_MAX_MS, + 0); + + reconfigureConnectorTasksWithExponentialBackoffRetries(initialRequestTime, connName, exponentialBackoff, 0); + } + + /** + * Request task configs from the connector and write them to the config storage in case the configs are detected to + * have changed. This method retries infinitely with exponential backoff in case of any errors. * * @param initialRequestTime the time in milliseconds when the original request was made (i.e. before any retries) * @param connName the name of the connector * @param exponentialBackoff {@link ExponentialBackoff} used to calculate the retry backoff duration * @param attempts the number of retry attempts that have been made */ - private void reconfigureConnectorTasksWithRetry(long initialRequestTime, final String connName, ExponentialBackoff exponentialBackoff, int attempts) { + private void reconfigureConnectorTasksWithExponentialBackoffRetries(long initialRequestTime, final String connName, ExponentialBackoff exponentialBackoff, int attempts) { reconfigureConnector(connName, (error, result) -> { // If we encountered an error, we don't have much choice but to just retry. If we don't, we could get // stuck with a connector that thinks it has generated tasks, but wasn't actually successful and therefore @@ -1903,7 +1913,7 @@ private void reconfigureConnectorTasksWithRetry(long initialRequestTime, final S } addRequest(exponentialBackoff.backoff(attempts), () -> { - reconfigureConnectorTasksWithRetry(initialRequestTime, connName, exponentialBackoff, attempts + 1); + reconfigureConnectorTasksWithExponentialBackoffRetries(initialRequestTime, connName, exponentialBackoff, attempts + 1); return null; }, (err, res) -> { if (err != null) { From a75630b11a82edfacfaf306659ee944c077c8b21 Mon Sep 17 00:00:00 2001 From: Yash Mayya Date: Tue, 28 Feb 2023 16:09:53 +0530 Subject: [PATCH 3/3] Update testTaskReconfigurationRetries to verify mock invocations and check extra herder tick after task configs generated successfully; Add testTaskReconfigurationRetriesWithLeaderRequestForwardingException; Revert Javadoc changes to ExponentialBackoff --- .../common/utils/ExponentialBackoff.java | 11 +- .../distributed/DistributedHerder.java | 14 +- .../distributed/DistributedHerderTest.java | 123 ++++++++++++++++-- 3 files changed, 123 insertions(+), 25 deletions(-) diff --git a/clients/src/main/java/org/apache/kafka/common/utils/ExponentialBackoff.java b/clients/src/main/java/org/apache/kafka/common/utils/ExponentialBackoff.java index 30a6bf8e7fed8..7550184ba39d0 100644 --- a/clients/src/main/java/org/apache/kafka/common/utils/ExponentialBackoff.java +++ b/clients/src/main/java/org/apache/kafka/common/utils/ExponentialBackoff.java @@ -20,15 +20,12 @@ import java.util.concurrent.ThreadLocalRandom; /** - * A utility class for keeping the parameters and providing the value of exponential + * An utility class for keeping the parameters and providing the value of exponential * retry backoff, exponential reconnect backoff, exponential timeout, etc. - *

* The formula is: - * 

Backoff(attempts) = random(1 - jitter, 1 + jitter) * initialInterval * multiplier ^ attempts
- * If {@code initialInterval} is greater than or equal to {@code maxInterval}, a constant backoff of - * {@code initialInterval} will be provided. - *

- * This class is thread-safe. + * Backoff(attempts) = random(1 - jitter, 1 + jitter) * initialInterval * multiplier ^ attempts + * If initialInterval is greater or equal than maxInterval, a constant backoff of will be provided + * This class is thread-safe */ public class ExponentialBackoff { private final int multiplier; diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedHerder.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedHerder.java index 9c07f36944f08..8a8d8dd95df7f 100644 --- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedHerder.java +++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedHerder.java @@ -253,7 +253,7 @@ public DistributedHerder(DistributedConfig config, List restNamespace, AutoCloseable... uponShutdown) { this(config, worker, worker.workerId(), kafkaClusterId, statusBackingStore, configBackingStore, null, restUrl, restClient, worker.metrics(), - time, connectorClientConfigOverridePolicy, restNamespace, uponShutdown); + time, connectorClientConfigOverridePolicy, restNamespace, null, uponShutdown); configBackingStore.setUpdateListener(new ConfigUpdateListener()); } @@ -271,6 +271,7 @@ public DistributedHerder(DistributedConfig config, Time time, ConnectorClientConfigOverridePolicy connectorClientConfigOverridePolicy, List restNamespace, + ExecutorService forwardRequestExecutor, AutoCloseable... uponShutdown) { super(worker, workerId, kafkaClusterId, statusBackingStore, configBackingStore, connectorClientConfigOverridePolicy); @@ -304,9 +305,12 @@ public DistributedHerder(DistributedConfig config, ThreadUtils.createThreadFactory( this.getClass().getSimpleName() + "-" + clientId + "-%d", false)); - this.forwardRequestExecutor = Executors.newFixedThreadPool(1, - ThreadUtils.createThreadFactory( - "ForwardRequestExecutor-" + clientId + "-%d", false)); + this.forwardRequestExecutor = forwardRequestExecutor != null + ? forwardRequestExecutor + : Executors.newFixedThreadPool( + 1, + ThreadUtils.createThreadFactory("ForwardRequestExecutor-" + clientId + "-%d", false) + ); this.startAndStopExecutor = Executors.newFixedThreadPool(START_STOP_THREAD_POOL_SIZE, ThreadUtils.createThreadFactory( "StartAndStopExecutor-" + clientId + "-%d", false)); @@ -1909,7 +1913,7 @@ private void reconfigureConnectorTasksWithExponentialBackoffRetries(long initial if (isPossibleExpiredKeyException(initialRequestTime, error)) { log.debug("Failed to reconfigure connector's tasks ({}), possibly due to expired session key. Retrying after backoff", connName); } else { - log.error("Failed to reconfigure connector's tasks ({}), retrying after backoff:", connName, error); + log.error("Failed to reconfigure connector's tasks ({}), retrying after backoff.", connName, error); } addRequest(exponentialBackoff.backoff(attempts), () -> { diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/distributed/DistributedHerderTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/distributed/DistributedHerderTest.java index a02d5b7a80382..722be88fb7edc 100644 --- a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/distributed/DistributedHerderTest.java +++ b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/distributed/DistributedHerderTest.java @@ -80,6 +80,7 @@ import org.powermock.reflect.Whitebox; import javax.crypto.SecretKey; +import javax.ws.rs.core.HttpHeaders; import java.util.ArrayList; import java.util.Arrays; import java.util.Collection; @@ -90,15 +91,16 @@ import java.util.Map; import java.util.Optional; import java.util.Set; +import java.util.concurrent.AbstractExecutorService; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Future; +import java.util.concurrent.RejectedExecutionException; import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; -import java.util.concurrent.RejectedExecutionException; import java.util.function.Function; import java.util.stream.Collectors; import java.util.stream.IntStream; @@ -216,8 +218,7 @@ public class DistributedHerderTest { @Mock private Callback> putConnectorCallback; @Mock private Plugins plugins; @Mock private RestClient restClient; - private CountDownLatch shutdownCalled = new CountDownLatch(1); - + private final CountDownLatch shutdownCalled = new CountDownLatch(1); private ConfigBackingStore.UpdateListener configUpdateListener; private WorkerRebalanceListener rebalanceListener; private ExecutorService herderExecutor; @@ -244,7 +245,7 @@ public void setUp() throws Exception { new String[]{"connectorType", "updateDeletedConnectorStatus", "updateDeletedTaskStatus", "validateConnectorConfig", "buildRestartPlan", "recordRestarting"}, new DistributedConfig(HERDER_CONFIG), worker, WORKER_ID, KAFKA_CLUSTER_ID, statusBackingStore, configBackingStore, member, MEMBER_URL, restClient, metrics, time, noneConnectorClientConfigOverridePolicy, - Collections.emptyList(), new AutoCloseable[]{uponShutdown}); + Collections.emptyList(), null, new AutoCloseable[]{uponShutdown}); configUpdateListener = herder.new ConfigUpdateListener(); rebalanceListener = herder.new RebalanceListener(time); @@ -3695,7 +3696,7 @@ public void shouldThrowWhenStartAndStopExecutorThrowsRejectedExecutionExceptionA } @Test - public void testTaskReconfigurationRetries() { + public void testTaskReconfigurationRetriesWithConnectorTaskConfigsException() { EasyMock.expect(member.memberId()).andStubReturn("leader"); EasyMock.expect(member.currentProtocolVersion()).andStubReturn(CONNECT_PROTOCOL_V0); expectRebalance(1, Collections.emptyList(), Collections.emptyList(), true); @@ -3706,30 +3707,84 @@ public void testTaskReconfigurationRetries() { PowerMock.expectLastCall(); member.wakeup(); - PowerMock.expectLastCall(); + PowerMock.expectLastCall().anyTimes(); - // second tick member.ensureActive(); - PowerMock.expectLastCall(); + PowerMock.expectLastCall().anyTimes(); EasyMock.expect(worker.isRunning(CONN1)).andReturn(true).anyTimes(); EasyMock.expect(worker.getPlugins()).andReturn(plugins).anyTimes(); SinkConnectorConfig sinkConnectorConfig = new SinkConnectorConfig(plugins, CONN1_CONFIG); EasyMock.expect(worker.connectorTaskConfigs(CONN1, sinkConnectorConfig)) - .andThrow(new ConnectException("Failed to generate task configs")).anyTimes(); + .andThrow(new ConnectException("Failed to generate task configs")).times(2); - // task reconfiguration request with initial retry backoff - member.poll(EasyMock.eq(250L)); + EasyMock.expect(worker.connectorTaskConfigs(CONN1, sinkConnectorConfig)).andReturn(TASK_CONFIGS); + + expectAndVerifyTaskReconfigurationRetries(); + } + + @Test + public void testTaskReconfigurationRetriesWithLeaderRequestForwardingException() { + herder = PowerMock.createPartialMock(DistributedHerder.class, + new String[]{"connectorType", "updateDeletedConnectorStatus", "updateDeletedTaskStatus", "validateConnectorConfig", "buildRestartPlan", "recordRestarting"}, + new DistributedConfig(HERDER_CONFIG), worker, WORKER_ID, KAFKA_CLUSTER_ID, + statusBackingStore, configBackingStore, member, MEMBER_URL, restClient, metrics, time, noneConnectorClientConfigOverridePolicy, + Collections.emptyList(), new MockSynchronousExecutor(), new AutoCloseable[]{}); + + rebalanceListener = herder.new RebalanceListener(time); + + EasyMock.expect(member.memberId()).andStubReturn("member"); + EasyMock.expect(member.currentProtocolVersion()).andStubReturn(CONNECT_PROTOCOL_V0); + expectRebalance(1, Collections.emptyList(), Collections.emptyList(), false); + expectConfigRefreshAndSnapshot(SNAPSHOT); + + // end of initial tick + member.poll(EasyMock.anyInt()); PowerMock.expectLastCall(); + member.wakeup(); + PowerMock.expectLastCall().anyTimes(); + member.ensureActive(); + PowerMock.expectLastCall().anyTimes(); + + EasyMock.expect(worker.isRunning(CONN1)).andReturn(true).anyTimes(); + EasyMock.expect(worker.getPlugins()).andReturn(plugins).anyTimes(); + + SinkConnectorConfig sinkConnectorConfig = new SinkConnectorConfig(plugins, CONN1_CONFIG); + + List> changedTaskConfigs = new ArrayList<>(TASK_CONFIGS); + changedTaskConfigs.add(TASK_CONFIG); + EasyMock.expect(worker.connectorTaskConfigs(CONN1, sinkConnectorConfig)).andReturn(changedTaskConfigs).anyTimes(); + + EasyMock.expect(restClient.httpRequest( + EasyMock.anyString(), EasyMock.eq("POST"), EasyMock.anyObject(HttpHeaders.class), + EasyMock.anyObject(), EasyMock.anyObject(), EasyMock.anyObject(SecretKey.class), EasyMock.anyString()) + ).andThrow(new ConnectException("Request to leader to reconfigure connector tasks failed")).times(2); + + EasyMock.expect(restClient.httpRequest( + EasyMock.anyString(), EasyMock.eq("POST"), EasyMock.anyObject(HttpHeaders.class), + EasyMock.anyObject(), EasyMock.anyObject(), EasyMock.anyObject(SecretKey.class), EasyMock.anyString()) + ).andReturn(null); + + expectAndVerifyTaskReconfigurationRetries(); + } + + private void expectAndVerifyTaskReconfigurationRetries() { + // task reconfiguration herder request with initial retry backoff + member.poll(EasyMock.eq(250L)); PowerMock.expectLastCall(); - // task reconfiguration request with double the initial retry backoff + // task reconfiguration herder request with double the initial retry backoff member.poll(EasyMock.eq(500L)); PowerMock.expectLastCall(); + // the third task reconfiguration request is expected to pass; so expect no more retries (a Long.MAX_VALUE poll + // timeout indicates that there is no herder request currently in the queue) + member.poll(EasyMock.eq(Long.MAX_VALUE)); + PowerMock.expectLastCall(); + PowerMock.replayAll(); // initial tick @@ -3740,6 +3795,11 @@ public void testTaskReconfigurationRetries() { // advance the time by 250ms so that the task reconfiguration request with initial retry backoff is processed time.sleep(250); herder.tick(); + // advance the time by 500ms so that the task reconfiguration request with double the initial retry backoff is processed + time.sleep(500); + herder.tick(); + + PowerMock.verifyAll(); } private void expectRebalance(final long offset, @@ -4061,6 +4121,43 @@ private abstract class BogusSourceConnector extends SourceConnector { private abstract class BogusSourceTask extends SourceTask { } + /** + * A mock {@link ExecutorService} that runs tasks synchronously on the same thread as the caller. This mock + * implementation can't be "shut down" and it is the responsibility of the caller to ensure that {@link Runnable}s + * submitted via {@link #execute(Runnable)} don't hang indefinitely. + */ + private static class MockSynchronousExecutor extends AbstractExecutorService { + @Override + public void execute(Runnable command) { + command.run(); + } + + @Override + public void shutdown() { + + } + + @Override + public List shutdownNow() { + return null; + } + + @Override + public boolean isShutdown() { + return false; + } + + @Override + public boolean isTerminated() { + return false; + } + + @Override + public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException { + return false; + } + } + private DistributedHerder exactlyOnceHerder() { Map config = new HashMap<>(HERDER_CONFIG); config.put(EXACTLY_ONCE_SOURCE_SUPPORT_CONFIG, "enabled"); @@ -4068,7 +4165,7 @@ private DistributedHerder exactlyOnceHerder() { new String[]{"connectorType", "updateDeletedConnectorStatus", "updateDeletedTaskStatus", "validateConnectorConfig"}, new DistributedConfig(config), worker, WORKER_ID, KAFKA_CLUSTER_ID, statusBackingStore, configBackingStore, member, MEMBER_URL, restClient, metrics, time, noneConnectorClientConfigOverridePolicy, - Collections.emptyList(), new AutoCloseable[0]); + Collections.emptyList(), null, new AutoCloseable[0]); } }