diff --git a/checkstyle/import-control.xml b/checkstyle/import-control.xml
index 4fc539d60670..377ce40fd2ca 100644
--- a/checkstyle/import-control.xml
+++ b/checkstyle/import-control.xml
@@ -544,6 +544,7 @@
+
diff --git a/connect/api/src/main/java/org/apache/kafka/connect/sink/SinkConnector.java b/connect/api/src/main/java/org/apache/kafka/connect/sink/SinkConnector.java
index 9627571482bc..d3726fd1dfc6 100644
--- a/connect/api/src/main/java/org/apache/kafka/connect/sink/SinkConnector.java
+++ b/connect/api/src/main/java/org/apache/kafka/connect/sink/SinkConnector.java
@@ -16,8 +16,11 @@
*/
package org.apache.kafka.connect.sink;
+import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.connect.connector.Connector;
+import java.util.Map;
+
/**
* SinkConnectors implement the Connector interface to send Kafka data to another system.
*/
@@ -39,4 +42,33 @@ protected SinkConnectorContext context() {
return (SinkConnectorContext) context;
}
+ /**
+ * Invoked when users request to manually alter/reset the offsets for this connector via the Connect worker's REST
+ * API. Connectors that manage offsets externally can propagate offset changes to their external system in this
+ * method. Connectors may also validate these offsets if, for example, an offset is out of range for what can be
+ * feasibly written to the external system.
+ *
+ * Connectors that neither manage offsets externally nor require custom offset validation need not implement this
+ * method beyond simply returning {@code true}.
+ *
+ * User requests to alter/reset offsets will be handled by the Connect runtime and will be reflected in the offsets
+ * for this connector's consumer group.
+ *
+ * Similar to {@link #validate(Map) validate}, this method may be called by the runtime before the
+ * {@link #start(Map) start} method is invoked.
+ *
+ * @param connectorConfig the configuration of the connector
+ * @param offsets a map from topic partition to offset, containing the offsets that the user has requested to
+ * alter/reset. For any topic partitions whose offsets are being reset instead of altered, their
+ * corresponding value in the map will be {@code null}.
+ * @return whether this method has been overridden by the connector; the default implementation returns
+ * {@code false}, and all other implementations (that do not unconditionally throw exceptions) should return
+ * {@code true}
+ * @throws UnsupportedOperationException if it is impossible to alter/reset the offsets for this connector
+ * @throws org.apache.kafka.connect.errors.ConnectException if the offsets for this connector cannot be
+ * reset for any other reason (for example, they have failed custom validation logic specific to this connector)
+ */
+ public boolean alterOffsets(Map connectorConfig, Map offsets) {
+ return false;
+ }
}
diff --git a/connect/api/src/main/java/org/apache/kafka/connect/source/SourceConnector.java b/connect/api/src/main/java/org/apache/kafka/connect/source/SourceConnector.java
index c06279df4211..eaaf56566c86 100644
--- a/connect/api/src/main/java/org/apache/kafka/connect/source/SourceConnector.java
+++ b/connect/api/src/main/java/org/apache/kafka/connect/source/SourceConnector.java
@@ -71,4 +71,35 @@ public ExactlyOnceSupport exactlyOnceSupport(Map connectorConfig
public ConnectorTransactionBoundaries canDefineTransactionBoundaries(Map connectorConfig) {
return ConnectorTransactionBoundaries.UNSUPPORTED;
}
+
+ /**
+ * Invoked when users request to manually alter/reset the offsets for this connector via the Connect worker's REST
+ * API. Connectors that manage offsets externally can propagate offset changes to their external system in this
+ * method. Connectors may also validate these offsets to ensure that the source partitions and source offsets are
+ * in a format that is recognizable to them.
+ *
+ * Connectors that neither manage offsets externally nor require custom offset validation need not implement this
+ * method beyond simply returning {@code true}.
+ *
+ * User requests to alter/reset offsets will be handled by the Connect runtime and will be reflected in the offsets
+ * returned by any {@link org.apache.kafka.connect.storage.OffsetStorageReader OffsetStorageReader instances}
+ * provided to this connector and its tasks.
+ *
+ * Similar to {@link #validate(Map) validate}, this method may be called by the runtime before the
+ * {@link #start(Map) start} method is invoked.
+ *
+ * @param connectorConfig the configuration of the connector
+ * @param offsets a map from source partition to source offset, containing the offsets that the user has requested
+ * to alter/reset. For any source partitions whose offsets are being reset instead of altered, their
+ * corresponding source offset value in the map will be {@code null}
+ * @return whether this method has been overridden by the connector; the default implementation returns
+ * {@code false}, and all other implementations (that do not unconditionally throw exceptions) should return
+ * {@code true}
+ * @throws UnsupportedOperationException if it is impossible to alter/reset the offsets for this connector
+ * @throws org.apache.kafka.connect.errors.ConnectException if the offsets for this connector cannot be
+ * reset for any other reason (for example, they have failed custom validation logic specific to this connector)
+ */
+ public boolean alterOffsets(Map connectorConfig, Map, Map> offsets) {
+ return false;
+ }
}
diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Herder.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Herder.java
index 099a012be3fa..7d00dba3d35c 100644
--- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Herder.java
+++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Herder.java
@@ -25,6 +25,7 @@
import org.apache.kafka.connect.runtime.rest.entities.ConnectorInfo;
import org.apache.kafka.connect.runtime.rest.entities.ConnectorOffsets;
import org.apache.kafka.connect.runtime.rest.entities.ConnectorStateInfo;
+import org.apache.kafka.connect.runtime.rest.entities.Message;
import org.apache.kafka.connect.runtime.rest.entities.TaskInfo;
import org.apache.kafka.connect.storage.StatusBackingStore;
import org.apache.kafka.connect.util.Callback;
@@ -302,6 +303,14 @@ default void validateConnectorConfig(Map connectorConfig, Callba
*/
void connectorOffsets(String connName, Callback cb);
+ /**
+ * Alter a connector's offsets.
+ * @param connName the name of the connector whose offsets are to be altered
+ * @param offsets a mapping from partitions to offsets that need to be written
+ * @param cb callback to invoke upon completion
+ */
+ void alterConnectorOffsets(String connName, Map, Map> offsets, Callback cb);
+
enum ConfigReloadAction {
NONE,
RESTART
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 bc08c48322e5..f52446f234ed 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
@@ -18,21 +18,29 @@
import org.apache.kafka.clients.admin.Admin;
import org.apache.kafka.clients.admin.AdminClientConfig;
+import org.apache.kafka.clients.admin.AlterConsumerGroupOffsetsOptions;
+import org.apache.kafka.clients.admin.AlterConsumerGroupOffsetsResult;
+import org.apache.kafka.clients.admin.DeleteConsumerGroupOffsetsOptions;
+import org.apache.kafka.clients.admin.DeleteConsumerGroupOffsetsResult;
import org.apache.kafka.clients.admin.FenceProducersOptions;
import org.apache.kafka.clients.admin.ListConsumerGroupOffsetsOptions;
import org.apache.kafka.clients.admin.ListConsumerGroupOffsetsResult;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.IsolationLevel;
import org.apache.kafka.common.KafkaFuture;
import org.apache.kafka.common.MetricNameTemplate;
+import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.config.ConfigDef;
import org.apache.kafka.common.config.ConfigValue;
import org.apache.kafka.common.config.provider.ConfigProvider;
import org.apache.kafka.common.utils.ThreadUtils;
+import org.apache.kafka.common.errors.GroupSubscribedToTopicException;
+import org.apache.kafka.common.errors.UnknownMemberIdException;
import org.apache.kafka.common.utils.Time;
import org.apache.kafka.common.utils.Utils;
import org.apache.kafka.connect.connector.Connector;
@@ -56,9 +64,12 @@
import org.apache.kafka.connect.runtime.isolation.Plugins.ClassLoaderUsage;
import org.apache.kafka.connect.runtime.rest.entities.ConnectorOffset;
import org.apache.kafka.connect.runtime.rest.entities.ConnectorOffsets;
+import org.apache.kafka.connect.runtime.rest.entities.Message;
import org.apache.kafka.connect.runtime.rest.resources.ConnectResource;
+import org.apache.kafka.connect.sink.SinkConnector;
import org.apache.kafka.connect.sink.SinkRecord;
import org.apache.kafka.connect.sink.SinkTask;
+import org.apache.kafka.connect.source.SourceConnector;
import org.apache.kafka.connect.source.SourceRecord;
import org.apache.kafka.connect.source.SourceTask;
import org.apache.kafka.connect.storage.CloseableOffsetStorageReader;
@@ -73,6 +84,7 @@
import org.apache.kafka.connect.util.Callback;
import org.apache.kafka.connect.util.ConnectUtils;
import org.apache.kafka.connect.util.ConnectorTaskId;
+import org.apache.kafka.connect.util.FutureCallback;
import org.apache.kafka.connect.util.LoggingContext;
import org.apache.kafka.connect.util.SinkUtils;
import org.apache.kafka.connect.util.TopicAdmin;
@@ -93,9 +105,11 @@
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
+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.function.Function;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
@@ -138,6 +152,7 @@ public class Worker {
private Optional sourceTaskOffsetCommitter;
private final WorkerConfigTransformer workerConfigTransformer;
private final ConnectorClientConfigOverridePolicy connectorClientConfigOverridePolicy;
+ private final Function, Admin> adminFactory;
public Worker(
String workerId,
@@ -146,7 +161,7 @@ public Worker(
WorkerConfig config,
OffsetBackingStore globalOffsetBackingStore,
ConnectorClientConfigOverridePolicy connectorClientConfigOverridePolicy) {
- this(workerId, time, plugins, config, globalOffsetBackingStore, Executors.newCachedThreadPool(), connectorClientConfigOverridePolicy);
+ this(workerId, time, plugins, config, globalOffsetBackingStore, Executors.newCachedThreadPool(), connectorClientConfigOverridePolicy, Admin::create);
}
Worker(
@@ -156,7 +171,8 @@ public Worker(
WorkerConfig config,
OffsetBackingStore globalOffsetBackingStore,
ExecutorService executorService,
- ConnectorClientConfigOverridePolicy connectorClientConfigOverridePolicy
+ ConnectorClientConfigOverridePolicy connectorClientConfigOverridePolicy,
+ Function, Admin> adminFactory
) {
this.kafkaClusterId = config.kafkaClusterId();
this.metrics = new ConnectMetrics(workerId, config, time, kafkaClusterId);
@@ -175,7 +191,7 @@ public Worker(
this.globalOffsetBackingStore = globalOffsetBackingStore;
this.workerConfigTransformer = initConfigTransformer();
-
+ this.adminFactory = adminFactory;
}
private WorkerConfigTransformer initConfigTransformer() {
@@ -297,8 +313,8 @@ public void startConnector(
// Set up the offset backing store for this connector instance
offsetStore = config.exactlyOnceSourceEnabled()
- ? offsetStoreForExactlyOnceSourceConnector(sourceConfig, connName, connector)
- : offsetStoreForRegularSourceConnector(sourceConfig, connName, connector);
+ ? offsetStoreForExactlyOnceSourceConnector(sourceConfig, connName, connector, null)
+ : offsetStoreForRegularSourceConnector(sourceConfig, connName, connector, null);
offsetStore.configure(config);
offsetReader = new OffsetStorageReaderImpl(offsetStore, connName, internalKeyConverter, internalValueConverter);
}
@@ -668,11 +684,6 @@ private boolean startTask(
* @return a {@link KafkaFuture} that will complete when the producers have all been fenced out, or the attempt has failed
*/
public KafkaFuture fenceZombies(String connName, int numTasks, Map connProps) {
- return fenceZombies(connName, numTasks, connProps, Admin::create);
- }
-
- // Allows us to mock out the Admin client for testing
- KafkaFuture fenceZombies(String connName, int numTasks, Map connProps, Function, Admin> adminFactory) {
log.debug("Fencing out {} task producers for source connector {}", numTasks, connName);
try (LoggingContext loggingContext = LoggingContext.forConnector(connName)) {
String connType = connProps.get(ConnectorConfig.CONNECTOR_CLASS_CONFIG);
@@ -1151,19 +1162,16 @@ public void connectorOffsets(String connName, Map connectorConfi
/**
* Get the current consumer group offsets for a sink connector.
+ *
+ * Visible for testing.
+ *
* @param connName the name of the sink connector whose offsets are to be retrieved
* @param connector the sink connector
* @param connectorConfig the sink connector's configurations
* @param cb callback to invoke upon completion of the request
*/
- private void sinkConnectorOffsets(String connName, Connector connector, Map connectorConfig,
- Callback cb) {
- sinkConnectorOffsets(connName, connector, connectorConfig, cb, Admin::create);
- }
-
- // Visible for testing; allows us to mock out the Admin client for testing
void sinkConnectorOffsets(String connName, Connector connector, Map connectorConfig,
- Callback cb, Function, Admin> adminFactory) {
+ Callback cb) {
Map adminConfig = adminConfigs(
connName,
"connector-worker-adminclient-" + connName,
@@ -1199,6 +1207,7 @@ void sinkConnectorOffsets(String connName, Connector connector, Map cb) {
SourceConnectorConfig sourceConfig = new SourceConnectorConfig(plugins, connectorConfig, config.topicCreationEnable());
ConnectorOffsetBackingStore offsetStore = config.exactlyOnceSourceEnabled()
- ? offsetStoreForExactlyOnceSourceConnector(sourceConfig, connName, connector)
- : offsetStoreForRegularSourceConnector(sourceConfig, connName, connector);
+ ? offsetStoreForExactlyOnceSourceConnector(sourceConfig, connName, connector, null)
+ : offsetStoreForRegularSourceConnector(sourceConfig, connName, connector, null);
CloseableOffsetStorageReader offsetReader = new OffsetStorageReaderImpl(offsetStore, connName, internalKeyConverter, internalValueConverter);
sourceConnectorOffsets(connName, offsetStore, offsetReader, cb);
}
@@ -1235,6 +1244,249 @@ void sourceConnectorOffsets(String connName, ConnectorOffsetBackingStore offsetS
});
}
+ /**
+ * Alter a connector's offsets.
+ *
+ * @param connName the name of the connector whose offsets are to be altered
+ * @param connectorConfig the connector's configurations
+ * @param offsets a mapping from partitions (either source partitions for source connectors, or Kafka topic
+ * partitions for sink connectors) to offsets that need to be written; may not be null or empty
+ * @param cb callback to invoke upon completion
+ */
+ public void alterConnectorOffsets(String connName, Map connectorConfig,
+ Map, Map> offsets, Callback cb) {
+
+ if (offsets == null || offsets.isEmpty()) {
+ throw new ConnectException("The offsets to be altered may not be null or empty");
+ }
+
+ String connectorClassOrAlias = connectorConfig.get(ConnectorConfig.CONNECTOR_CLASS_CONFIG);
+ ClassLoader connectorLoader = plugins.connectorLoader(connectorClassOrAlias);
+ Connector connector;
+
+ try (LoaderSwap loaderSwap = plugins.withClassLoader(connectorLoader)) {
+ connector = plugins.newConnector(connectorClassOrAlias);
+ if (ConnectUtils.isSinkConnector(connector)) {
+ log.debug("Altering consumer group offsets for sink connector: {}", connName);
+ alterSinkConnectorOffsets(connName, connector, connectorConfig, offsets, connectorLoader, cb);
+ } else {
+ log.debug("Altering offsets for source connector: {}", connName);
+ alterSourceConnectorOffsets(connName, connector, connectorConfig, offsets, connectorLoader, cb);
+ }
+ }
+ }
+
+ /**
+ * Alter a sink connector's consumer group offsets.
+ *
+ * Visible for testing.
+ *
+ * @param connName the name of the sink connector whose offsets are to be altered
+ * @param connector an instance of the sink connector
+ * @param connectorConfig the sink connector's configuration
+ * @param offsets a mapping from topic partitions to offsets that need to be written; may not be null or empty
+ * @param connectorLoader the connector plugin's classloader to be used as the thread context classloader
+ * @param cb callback to invoke upon completion
+ */
+ void alterSinkConnectorOffsets(String connName, Connector connector, Map connectorConfig,
+ Map, Map> offsets, ClassLoader connectorLoader, Callback cb) {
+ executor.submit(plugins.withClassLoader(connectorLoader, () -> {
+ try {
+ Map parsedOffsets = SinkUtils.parseSinkConnectorOffsets(offsets);
+ boolean alterOffsetsResult;
+ try {
+ alterOffsetsResult = ((SinkConnector) connector).alterOffsets(connectorConfig, parsedOffsets);
+ } catch (UnsupportedOperationException e) {
+ throw new ConnectException("Failed to alter offsets for connector " + connName + " because it doesn't support external " +
+ "modification of offsets", e);
+ }
+
+ SinkConnectorConfig sinkConnectorConfig = new SinkConnectorConfig(plugins, connectorConfig);
+ Class sinkConnectorClass = connector.getClass();
+ Map adminConfig = adminConfigs(
+ connName,
+ "connector-worker-adminclient-" + connName,
+ config,
+ sinkConnectorConfig,
+ sinkConnectorClass,
+ connectorClientConfigOverridePolicy,
+ kafkaClusterId,
+ ConnectorType.SINK);
+
+ String groupId = (String) baseConsumerConfigs(
+ connName, "connector-consumer-", config, sinkConnectorConfig,
+ sinkConnectorClass, connectorClientConfigOverridePolicy, kafkaClusterId, ConnectorType.SINK).get(ConsumerConfig.GROUP_ID_CONFIG);
+
+ Admin admin = adminFactory.apply(adminConfig);
+
+ try {
+ List> adminFutures = new ArrayList<>();
+
+ Map offsetsToAlter = parsedOffsets.entrySet()
+ .stream()
+ .filter(entry -> entry.getValue() != null)
+ .collect(Collectors.toMap(Map.Entry::getKey, e -> new OffsetAndMetadata(e.getValue())));
+
+ if (!offsetsToAlter.isEmpty()) {
+ log.debug("Committing the following consumer group offsets using an admin client for sink connector {}: {}.",
+ connName, offsetsToAlter);
+ AlterConsumerGroupOffsetsOptions alterConsumerGroupOffsetsOptions = new AlterConsumerGroupOffsetsOptions().timeoutMs(
+ (int) ConnectResource.DEFAULT_REST_REQUEST_TIMEOUT_MS);
+ AlterConsumerGroupOffsetsResult alterConsumerGroupOffsetsResult = admin.alterConsumerGroupOffsets(groupId, offsetsToAlter,
+ alterConsumerGroupOffsetsOptions);
+
+ adminFutures.add(alterConsumerGroupOffsetsResult.all());
+ }
+
+ Set partitionsToReset = parsedOffsets.entrySet()
+ .stream()
+ .filter(entry -> entry.getValue() == null)
+ .map(Map.Entry::getKey)
+ .collect(Collectors.toSet());
+
+ if (!partitionsToReset.isEmpty()) {
+ log.debug("Deleting the consumer group offsets for the following topic partitions using an admin client for sink connector {}: {}.",
+ connName, partitionsToReset);
+ DeleteConsumerGroupOffsetsOptions deleteConsumerGroupOffsetsOptions = new DeleteConsumerGroupOffsetsOptions().timeoutMs(
+ (int) ConnectResource.DEFAULT_REST_REQUEST_TIMEOUT_MS);
+ DeleteConsumerGroupOffsetsResult deleteConsumerGroupOffsetsResult = admin.deleteConsumerGroupOffsets(groupId, partitionsToReset,
+ deleteConsumerGroupOffsetsOptions);
+
+ adminFutures.add(deleteConsumerGroupOffsetsResult.all());
+ }
+
+ @SuppressWarnings("rawtypes")
+ KafkaFuture compositeAdminFuture = KafkaFuture.allOf(adminFutures.toArray(new KafkaFuture[0]));
+
+ compositeAdminFuture.whenComplete((ignored, error) -> {
+ if (error != null) {
+ // When a consumer group is non-empty, only group members can commit offsets. An attempt to alter offsets via the admin client
+ // will result in an UnknownMemberIdException if the consumer group is non-empty (i.e. if the sink tasks haven't stopped
+ // completely or if the connector is resumed while the alter offsets request is being processed). Similarly, an attempt to
+ // delete consumer group offsets for a non-empty consumer group will result in a GroupSubscribedToTopicException
+ if (error instanceof UnknownMemberIdException || error instanceof GroupSubscribedToTopicException) {
+ cb.onCompletion(new ConnectException("Failed to alter consumer group offsets for connector " + connName + " either because its tasks " +
+ "haven't stopped completely yet or the connector was resumed before the request to alter its offsets could be successfully " +
+ "completed. If the connector is in a stopped state, this operation can be safely retried. If it doesn't eventually succeed, the " +
+ "Connect cluster may need to be restarted to get rid of the zombie sink tasks."),
+ null);
+ } else {
+ cb.onCompletion(new ConnectException("Failed to alter consumer group offsets for connector " + connName, error), null);
+ }
+ } else {
+ completeAlterOffsetsCallback(alterOffsetsResult, cb);
+ }
+ }).whenComplete((ignored, ignoredError) -> {
+ // errors originating from the original future are handled in the prior whenComplete invocation which isn't expected to throw
+ // an exception itself, and we can thus ignore the error here
+ Utils.closeQuietly(admin, "Offset alter admin for sink connector " + connName);
+ });
+ } catch (Throwable t) {
+ Utils.closeQuietly(admin, "Offset alter admin for sink connector " + connName);
+ throw t;
+ }
+ } catch (Throwable t) {
+ cb.onCompletion(ConnectUtils.maybeWrap(t, "Failed to alter offsets for sink connector " + connName), null);
+ }
+ }));
+ }
+
+ /**
+ * Alter a source connector's offsets.
+ *
+ * @param connName the name of the source connector whose offsets are to be altered
+ * @param connector an instance of the source connector
+ * @param connectorConfig the source connector's configuration
+ * @param offsets a mapping from partitions to offsets that need to be written; may not be null or empty
+ * @param connectorLoader the connector plugin's classloader to be used as the thread context classloader
+ * @param cb callback to invoke upon completion
+ */
+ private void alterSourceConnectorOffsets(String connName, Connector connector, Map connectorConfig,
+ Map, Map> offsets, ClassLoader connectorLoader, Callback cb) {
+ SourceConnectorConfig sourceConfig = new SourceConnectorConfig(plugins, connectorConfig, config.topicCreationEnable());
+ Map producerProps = config.exactlyOnceSourceEnabled()
+ ? exactlyOnceSourceTaskProducerConfigs(new ConnectorTaskId(connName, 0), config, sourceConfig,
+ connector.getClass(), connectorClientConfigOverridePolicy, kafkaClusterId)
+ : baseProducerConfigs(connName, "connector-offset-producer-" + connName, config, sourceConfig,
+ connector.getClass(), connectorClientConfigOverridePolicy, kafkaClusterId);
+ KafkaProducer producer = new KafkaProducer<>(producerProps);
+
+ ConnectorOffsetBackingStore offsetStore = config.exactlyOnceSourceEnabled()
+ ? offsetStoreForExactlyOnceSourceConnector(sourceConfig, connName, connector, producer)
+ : offsetStoreForRegularSourceConnector(sourceConfig, connName, connector, producer);
+ offsetStore.configure(config);
+
+ OffsetStorageWriter offsetWriter = new OffsetStorageWriter(offsetStore, connName, internalKeyConverter, internalValueConverter);
+ alterSourceConnectorOffsets(connName, connector, connectorConfig, offsets, offsetStore, producer, offsetWriter, connectorLoader, cb);
+ }
+
+ // Visible for testing
+ void alterSourceConnectorOffsets(String connName, Connector connector, Map connectorConfig,
+ Map, Map> offsets, ConnectorOffsetBackingStore offsetStore,
+ KafkaProducer producer, OffsetStorageWriter offsetWriter,
+ ClassLoader connectorLoader, Callback cb) {
+ executor.submit(plugins.withClassLoader(connectorLoader, () -> {
+ try {
+ boolean alterOffsetsResult;
+ try {
+ alterOffsetsResult = ((SourceConnector) connector).alterOffsets(connectorConfig, offsets);
+ } catch (UnsupportedOperationException e) {
+ throw new ConnectException("Failed to alter offsets for connector " + connName + " because it doesn't support external " +
+ "modification of offsets", e);
+ }
+ // This reads to the end of the offsets topic and can be a potentially time-consuming operation
+ offsetStore.start();
+
+ // The alterSourceConnectorOffsets method should only be called after all the connector's tasks have been stopped, and it's
+ // safe to write offsets via an offset writer
+ offsets.forEach(offsetWriter::offset);
+
+ // We can call begin flush without a timeout because this newly created single-purpose offset writer can't do concurrent
+ // offset writes. We can also ignore the return value since it returns false if and only if there is no data to be flushed,
+ // and we've just put some data in the previous statement
+ offsetWriter.beginFlush();
+
+ if (config.exactlyOnceSourceEnabled()) {
+ producer.initTransactions();
+ producer.beginTransaction();
+ }
+ log.debug("Committing the following offsets for source connector {}: {}", connName, offsets);
+ FutureCallback offsetWriterCallback = new FutureCallback<>();
+ offsetWriter.doFlush(offsetWriterCallback);
+ if (config.exactlyOnceSourceEnabled()) {
+ producer.commitTransaction();
+ }
+
+ try {
+ offsetWriterCallback.get(ConnectResource.DEFAULT_REST_REQUEST_TIMEOUT_MS, TimeUnit.MILLISECONDS);
+ } catch (ExecutionException e) {
+ throw new ConnectException("Failed to alter offsets for source connector " + connName, e.getCause());
+ } catch (TimeoutException e) {
+ throw new ConnectException("Timed out while attempting to alter offsets for source connector " + connName, e);
+ } catch (InterruptedException e) {
+ throw new ConnectException("Unexpectedly interrupted while attempting to alter offsets for source connector " + connName, e);
+ }
+
+ completeAlterOffsetsCallback(alterOffsetsResult, cb);
+ } catch (Throwable t) {
+ log.error("Failed to alter offsets for source connector {}", connName, t);
+ cb.onCompletion(ConnectUtils.maybeWrap(t, "Failed to alter offsets for source connector " + connName), null);
+ } finally {
+ Utils.closeQuietly(offsetStore::stop, "Offset store for offset alter request for connector " + connName);
+ }
+ }));
+ }
+
+ private void completeAlterOffsetsCallback(boolean alterOffsetsResult, Callback cb) {
+ if (alterOffsetsResult) {
+ cb.onCompletion(null, new Message("The offsets for this connector have been altered successfully"));
+ } else {
+ cb.onCompletion(null, new Message("The Connect framework-managed offsets for this connector have been " +
+ "altered successfully. However, if this connector manages offsets externally, they will need to be " +
+ "manually altered in the system that the connector uses."));
+ }
+ }
+
ConnectorStatusMetricsGroup connectorStatusMetricsGroup() {
return connectorStatusMetricsGroup;
}
@@ -1500,11 +1752,25 @@ public WorkerTask doBuild(Task task,
}
}
- // Visible for testing
+ /**
+ * Builds and returns an offset backing store for a regular source connector (i.e. when exactly-once support for source connectors is disabled).
+ * The offset backing store will either be just the worker's global offset backing store (if the connector doesn't define a connector-specific
+ * offset topic via its configs), just a connector-specific offset backing store (if the connector defines a connector-specific offsets
+ * topic which appears to be the same as the worker's global offset topic) or a combination of both the worker's global offset backing store
+ * and a connector-specific offset backing store.
+ *
+ * Visible for testing.
+ * @param sourceConfig the source connector's config
+ * @param connName the source connector's name
+ * @param connector the source connector
+ * @param producer the Kafka producer for the offset backing store; may be {@code null} if a read-only offset backing store is required
+ * @return An offset backing store for a regular source connector
+ */
ConnectorOffsetBackingStore offsetStoreForRegularSourceConnector(
SourceConnectorConfig sourceConfig,
String connName,
- Connector connector
+ Connector connector,
+ Producer producer
) {
String connectorSpecificOffsetsTopic = sourceConfig.offsetsTopic();
@@ -1527,8 +1793,10 @@ ConnectorOffsetBackingStore offsetStoreForRegularSourceConnector(
sourceConfig, connector.getClass(), connectorClientConfigOverridePolicy, kafkaClusterId, ConnectorType.SOURCE);
TopicAdmin admin = new TopicAdmin(adminOverrides);
- KafkaOffsetBackingStore connectorStore =
- KafkaOffsetBackingStore.forConnector(connectorSpecificOffsetsTopic, consumer, admin, internalKeyConverter);
+
+ KafkaOffsetBackingStore connectorStore = producer == null
+ ? KafkaOffsetBackingStore.readOnlyStore(connectorSpecificOffsetsTopic, consumer, admin, internalKeyConverter)
+ : KafkaOffsetBackingStore.readWriteStore(connectorSpecificOffsetsTopic, producer, consumer, admin, internalKeyConverter);
// If the connector's offsets topic is the same as the worker-global offsets topic, there's no need to construct
// an offset store that has a primary and a secondary store which both read from that same topic.
@@ -1556,6 +1824,7 @@ ConnectorOffsetBackingStore offsetStoreForRegularSourceConnector(
);
}
} else {
+ Utils.closeQuietly(producer, "Unused producer for offset store");
return ConnectorOffsetBackingStore.withOnlyWorkerStore(
() -> LoggingContext.forConnector(connName),
globalOffsetBackingStore,
@@ -1564,11 +1833,23 @@ ConnectorOffsetBackingStore offsetStoreForRegularSourceConnector(
}
}
- // Visible for testing
+ /**
+ * Builds and returns an offset backing store for an exactly-once source connector. The offset backing store will either be just
+ * a connector-specific offset backing store (if the connector's offsets topic is the same as the worker's global offset topic)
+ * or a combination of both the worker's global offset backing store and a connector-specific offset backing store.
+ *
+ * Visible for testing.
+ * @param sourceConfig the source connector's config
+ * @param connName the source connector's name
+ * @param connector the source connector
+ * @param producer the Kafka producer for the offset backing store; may be {@code null} if a read-only offset backing store is required
+ * @return An offset backing store for an exactly-once source connector
+ */
ConnectorOffsetBackingStore offsetStoreForExactlyOnceSourceConnector(
SourceConnectorConfig sourceConfig,
String connName,
- Connector connector
+ Connector connector,
+ Producer producer
) {
String connectorSpecificOffsetsTopic = Optional.ofNullable(sourceConfig.offsetsTopic()).orElse(config.offsetsTopic());
@@ -1584,8 +1865,10 @@ ConnectorOffsetBackingStore offsetStoreForExactlyOnceSourceConnector(
sourceConfig, connector.getClass(), connectorClientConfigOverridePolicy, kafkaClusterId, ConnectorType.SOURCE);
TopicAdmin admin = new TopicAdmin(adminOverrides);
- KafkaOffsetBackingStore connectorStore =
- KafkaOffsetBackingStore.forConnector(connectorSpecificOffsetsTopic, consumer, admin, internalKeyConverter);
+
+ KafkaOffsetBackingStore connectorStore = producer == null
+ ? KafkaOffsetBackingStore.readOnlyStore(connectorSpecificOffsetsTopic, consumer, admin, internalKeyConverter)
+ : KafkaOffsetBackingStore.readWriteStore(connectorSpecificOffsetsTopic, producer, consumer, admin, internalKeyConverter);
// If the connector's offsets topic is the same as the worker-global offsets topic, there's no need to construct
// an offset store that has a primary and a secondary store which both read from that same topic.
@@ -1634,7 +1917,7 @@ ConnectorOffsetBackingStore offsetStoreForRegularSourceTask(
KafkaConsumer consumer = new KafkaConsumer<>(consumerProps);
KafkaOffsetBackingStore connectorStore =
- KafkaOffsetBackingStore.forTask(sourceConfig.offsetsTopic(), producer, consumer, topicAdmin, internalKeyConverter);
+ KafkaOffsetBackingStore.readWriteStore(sourceConfig.offsetsTopic(), producer, consumer, topicAdmin, internalKeyConverter);
// If the connector's offsets topic is the same as the worker-global offsets topic, there's no need to construct
// an offset store that has a primary and a secondary store which both read from that same topic.
@@ -1689,7 +1972,7 @@ ConnectorOffsetBackingStore offsetStoreForExactlyOnceSourceTask(
String connectorOffsetsTopic = Optional.ofNullable(sourceConfig.offsetsTopic()).orElse(config.offsetsTopic());
KafkaOffsetBackingStore connectorStore =
- KafkaOffsetBackingStore.forTask(connectorOffsetsTopic, producer, consumer, topicAdmin, internalKeyConverter);
+ KafkaOffsetBackingStore.readWriteStore(connectorOffsetsTopic, producer, consumer, topicAdmin, internalKeyConverter);
// If the connector's offsets topic is the same as the worker-global offsets topic, there's no need to construct
// an offset store that has a primary and a secondary store which both read from that same topic.
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 51258f655ecd..6d9158d83f70 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
@@ -52,6 +52,7 @@
import org.apache.kafka.connect.runtime.TaskStatus;
import org.apache.kafka.connect.runtime.Worker;
import org.apache.kafka.connect.runtime.rest.entities.ConnectorOffsets;
+import org.apache.kafka.connect.runtime.rest.entities.Message;
import org.apache.kafka.connect.storage.PrivilegedWriteException;
import org.apache.kafka.connect.runtime.rest.InternalRequestSignature;
import org.apache.kafka.connect.runtime.rest.RestClient;
@@ -1234,74 +1235,78 @@ void fenceZombieSourceTasks(final ConnectorTaskId id, Callback callback) {
// Visible for testing
void fenceZombieSourceTasks(final String connName, final Callback callback) {
- addRequest(
- () -> {
- log.trace("Performing zombie fencing request for connector {}", connName);
- if (!isLeader())
- callback.onCompletion(new NotLeaderException("Only the leader may perform zombie fencing.", leaderUrl()), null);
- else if (!configState.contains(connName))
- callback.onCompletion(new NotFoundException("Connector " + connName + " not found"), null);
- else if (!isSourceConnector(connName))
- callback.onCompletion(new BadRequestException("Connector " + connName + " is not a source connector"), null);
- else {
- if (!refreshConfigSnapshot(workerSyncTimeoutMs)) {
- throw new ConnectException("Failed to read to end of config topic before performing zombie fencing");
- }
+ addRequest(() -> {
+ doFenceZombieSourceTasks(connName, callback);
+ return null;
+ }, forwardErrorCallback(callback));
+ }
- int taskCount = configState.taskCount(connName);
- Integer taskCountRecord = configState.taskCountRecord(connName);
-
- ZombieFencing zombieFencing = null;
- boolean newFencing = false;
- synchronized (DistributedHerder.this) {
- // Check first to see if we have to do a fencing. The control flow is a little awkward here (why not stick this in
- // an else block lower down?) but we can't synchronize around the body below since that may contain a synchronous
- // write to the config topic.
- if (configState.pendingFencing(connName) && taskCountRecord != null
- && (taskCountRecord != 1 || taskCount != 1)) {
- int taskGen = configState.taskConfigGeneration(connName);
- zombieFencing = activeZombieFencings.get(connName);
- if (zombieFencing == null) {
- zombieFencing = new ZombieFencing(connName, taskCountRecord, taskCount, taskGen);
- activeZombieFencings.put(connName, zombieFencing);
- newFencing = true;
- }
- }
- }
- if (zombieFencing != null) {
- if (newFencing) {
- zombieFencing.start();
- }
- zombieFencing.addCallback(callback);
- return null;
- }
+ private void doFenceZombieSourceTasks(String connName, Callback callback) {
+ log.trace("Performing zombie fencing request for connector {}", connName);
- if (!configState.pendingFencing(connName)) {
- // If the latest task count record for the connector is present after the latest set of task configs, there's no need to
- // do any zombie fencing or write a new task count record to the config topic
- log.debug("Skipping zombie fencing round for connector {} as all old task generations have already been fenced out", connName);
- } else {
- if (taskCountRecord == null) {
- // If there is no task count record present for the connector, no transactional producers should have been brought up for it,
- // so there's nothing to fence--but we do need to write a task count record now so that we know to fence those tasks if/when
- // the connector is reconfigured
- log.debug("Skipping zombie fencing round but writing task count record for connector {} "
- + "as it is being brought up for the first time with exactly-once source support", connName);
- } else {
- // If the last generation of tasks only had one task, and the next generation only has one, then the new task will automatically
- // fence out the older task if it's still running; no need to fence here, but again, we still need to write a task count record
- log.debug("Skipping zombie fencing round but writing task count record for connector {} "
- + "as both the most recent and the current generation of task configs only contain one task", connName);
- }
- writeToConfigTopicAsLeader(() -> configBackingStore.putTaskCountRecord(connName, taskCount));
- }
- callback.onCompletion(null, null);
- return null;
+ if (checkRebalanceNeeded(callback))
+ return;
+
+ if (!isLeader())
+ callback.onCompletion(new NotLeaderException("Only the leader may perform zombie fencing.", leaderUrl()), null);
+ else if (!configState.contains(connName))
+ callback.onCompletion(new NotFoundException("Connector " + connName + " not found"), null);
+ else if (!isSourceConnector(connName))
+ callback.onCompletion(new BadRequestException("Connector " + connName + " is not a source connector"), null);
+ else {
+ if (!refreshConfigSnapshot(workerSyncTimeoutMs)) {
+ throw new ConnectException("Failed to read to end of config topic before performing zombie fencing");
+ }
+
+ int taskCount = configState.taskCount(connName);
+ Integer taskCountRecord = configState.taskCountRecord(connName);
+
+ ZombieFencing zombieFencing = null;
+ boolean newFencing = false;
+ synchronized (DistributedHerder.this) {
+ // Check first to see if we have to do a fencing. The control flow is a little awkward here (why not stick this in
+ // an else block lower down?) but we can't synchronize around the body below since that may contain a synchronous
+ // write to the config topic.
+ if (configState.pendingFencing(connName) && taskCountRecord != null
+ && (taskCountRecord != 1 || taskCount != 1)) {
+ int taskGen = configState.taskConfigGeneration(connName);
+ zombieFencing = activeZombieFencings.get(connName);
+ if (zombieFencing == null) {
+ zombieFencing = new ZombieFencing(connName, taskCountRecord, taskCount, taskGen);
+ activeZombieFencings.put(connName, zombieFencing);
+ newFencing = true;
}
- return null;
- },
- forwardErrorCallback(callback)
- );
+ }
+ }
+ if (zombieFencing != null) {
+ if (newFencing) {
+ zombieFencing.start();
+ }
+ zombieFencing.addCallback(callback);
+ return;
+ }
+
+ if (!configState.pendingFencing(connName)) {
+ // If the latest task count record for the connector is present after the latest set of task configs, there's no need to
+ // do any zombie fencing or write a new task count record to the config topic
+ log.debug("Skipping zombie fencing round for connector {} as all old task generations have already been fenced out", connName);
+ } else {
+ if (taskCountRecord == null) {
+ // If there is no task count record present for the connector, no transactional producers should have been brought up for it,
+ // so there's nothing to fence--but we do need to write a task count record now so that we know to fence those tasks if/when
+ // the connector is reconfigured
+ log.debug("Skipping zombie fencing round but writing task count record for connector {} "
+ + "as it is being brought up for the first time with exactly-once source support", connName);
+ } else {
+ // If the last generation of tasks only had one task, and the next generation only has one, then the new task will automatically
+ // fence out the older task if it's still running; no need to fence here, but again, we still need to write a task count record
+ log.debug("Skipping zombie fencing round but writing task count record for connector {} "
+ + "as both the most recent and the current generation of task configs only contain one task", connName);
+ }
+ writeToConfigTopicAsLeader(() -> configBackingStore.putTaskCountRecord(connName, taskCount));
+ }
+ callback.onCompletion(null, null);
+ }
}
@Override
@@ -1515,6 +1520,80 @@ public void connectorOffsets(String connName, Callback cb) {
);
}
+ @Override
+ public void alterConnectorOffsets(String connName, Map, Map> offsets, Callback callback) {
+ log.trace("Submitting alter offsets request for connector '{}'", connName);
+
+ addRequest(() -> {
+ if (!alterConnectorOffsetsChecks(connName, callback)) {
+ return null;
+ }
+ // At this point, we should be the leader (the call to alterConnectorOffsetsChecks makes sure of that) and can safely run
+ // a zombie fencing request
+ if (isSourceConnector(connName) && config.exactlyOnceSourceEnabled()) {
+ log.debug("Performing a round of zombie fencing before altering offsets for source connector {} with exactly-once support enabled.", connName);
+ doFenceZombieSourceTasks(connName, (error, ignored) -> {
+ if (error != null) {
+ log.error("Failed to perform zombie fencing for source connector prior to altering offsets", error);
+ callback.onCompletion(new ConnectException("Failed to perform zombie fencing for source connector prior to altering offsets",
+ error), null);
+ } else {
+ log.debug("Successfully completed zombie fencing for source connector {}; proceeding to alter offsets.", connName);
+ // We need to ensure that we perform the necessary checks again before proceeding to actually altering the connector offsets since
+ // zombie fencing is done asynchronously and the conditions could have changed since the previous check
+ addRequest(() -> {
+ if (alterConnectorOffsetsChecks(connName, callback)) {
+ worker.alterConnectorOffsets(connName, configState.connectorConfig(connName), offsets, callback);
+ }
+ return null;
+ }, forwardErrorCallback(callback));
+ }
+ });
+ } else {
+ worker.alterConnectorOffsets(connName, configState.connectorConfig(connName), offsets, callback);
+ }
+ return null;
+ }, forwardErrorCallback(callback));
+ }
+
+ /**
+ * This method performs a few checks for alter connector offsets request and completes the callback exceptionally
+ * if any check fails.
+ * @param connName the name of the connector whose offsets are to be altered
+ * @param callback callback to invoke upon completion
+ * @return true if all the checks passed, false otherwise
+ */
+ private boolean alterConnectorOffsetsChecks(String connName, Callback callback) {
+ if (checkRebalanceNeeded(callback)) {
+ return false;
+ }
+
+ if (!isLeader()) {
+ callback.onCompletion(new NotLeaderException("Only the leader can process alter offsets requests", leaderUrl()), null);
+ return false;
+ }
+
+ if (!refreshConfigSnapshot(workerSyncTimeoutMs)) {
+ throw new ConnectException("Failed to read to end of config topic before altering connector offsets");
+ }
+
+ if (!configState.contains(connName)) {
+ callback.onCompletion(new NotFoundException("Connector " + connName + " not found", null), null);
+ return false;
+ }
+
+ // If the target state for the connector is stopped, its task count is 0, and there is no rebalance pending (checked above),
+ // we can be sure that the tasks have at least been attempted to be stopped (or cancelled if they took too long to stop).
+ // Zombie tasks are handled by a round of zombie fencing for exactly once source connectors. Zombie sink tasks are handled
+ // naturally because requests to alter consumer group offsets will fail if there are still active members in the group.
+ if (configState.targetState(connName) != TargetState.STOPPED || configState.taskCount(connName) != 0) {
+ callback.onCompletion(new BadRequestException("Connectors must be in the STOPPED state before their offsets can be altered. This " +
+ "can be done for the specified connector by issuing a PUT request to the /connectors/" + connName + "/stop endpoint"), null);
+ return false;
+ }
+ return true;
+ }
+
// Should only be called from work thread, so synchronization should not be needed
private boolean isLeader() {
return assignment != null && member.memberId().equals(assignment.leader());
diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/HerderRequestHandler.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/HerderRequestHandler.java
index c26af44226bb..4dfd093bc2e2 100644
--- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/HerderRequestHandler.java
+++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/HerderRequestHandler.java
@@ -78,14 +78,14 @@ public T completeRequest(FutureCallback cb) throws Throwable {
* request to the indicated target.
*/
public T completeOrForwardRequest(FutureCallback cb,
- String path,
- String method,
- HttpHeaders headers,
- Map queryParameters,
- Object body,
- TypeReference resultType,
- Translator translator,
- Boolean forward) throws Throwable {
+ String path,
+ String method,
+ HttpHeaders headers,
+ Map queryParameters,
+ Object body,
+ TypeReference resultType,
+ Translator translator,
+ Boolean forward) throws Throwable {
try {
return completeRequest(cb);
} catch (RequestTargetException e) {
@@ -111,6 +111,8 @@ public T completeOrForwardRequest(FutureCallback cb,
log.debug("Forwarding request {} {} {}", forwardUrl, method, body);
return translator.translate(restClient.httpRequest(forwardUrl, method, headers, body, resultType));
} else {
+ log.error("Request '{} {}' failed because it couldn't find the target Connect worker within two hops (between workers).",
+ method, path);
// we should find the right target for the query within two hops, so if
// we don't, it probably means that a rebalance has taken place.
throw new ConnectRestException(Response.Status.CONFLICT.getStatusCode(),
@@ -127,13 +129,8 @@ public T completeOrForwardRequest(FutureCallback cb, String path, Stri
return completeOrForwardRequest(cb, path, method, headers, null, body, resultType, translator, forward);
}
- public T completeOrForwardRequest(FutureCallback cb, String path, String method, HttpHeaders headers, Object body,
- TypeReference resultType, Boolean forward) throws Throwable {
- return completeOrForwardRequest(cb, path, method, headers, body, resultType, new IdentityTranslator<>(), forward);
- }
-
public T completeOrForwardRequest(FutureCallback cb, String path, String method, HttpHeaders headers,
- Object body, Boolean forward) throws Throwable {
+ Object body, Boolean forward) throws Throwable {
return completeOrForwardRequest(cb, path, method, headers, body, null, new IdentityTranslator<>(), forward);
}
diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/entities/ConnectorOffsets.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/entities/ConnectorOffsets.java
index dbb3f0b305f8..0b5bcd6588a7 100644
--- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/entities/ConnectorOffsets.java
+++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/entities/ConnectorOffsets.java
@@ -19,7 +19,9 @@
import com.fasterxml.jackson.annotation.JsonCreator;
import com.fasterxml.jackson.annotation.JsonProperty;
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
import java.util.Objects;
/**
@@ -57,6 +59,14 @@ public List offsets() {
return offsets;
}
+ public Map, Map> toMap() {
+ Map, Map> partitionOffsetMap = new HashMap<>();
+ for (ConnectorOffset offset : offsets) {
+ partitionOffsetMap.put(offset.partition(), offset.offset());
+ }
+ return partitionOffsetMap;
+ }
+
@Override
public boolean equals(Object obj) {
if (this == obj) {
diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/entities/Message.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/entities/Message.java
new file mode 100644
index 000000000000..e4dc8fd0b6f6
--- /dev/null
+++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/entities/Message.java
@@ -0,0 +1,61 @@
+/*
+ * 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.runtime.rest.entities;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+import java.util.Objects;
+
+/**
+ * Standard format for regular successful REST API responses that look like:
+ *
+ * {
+ * "message": "Message goes here."
+ * }
+ *
+ */
+public class Message {
+ private final String message;
+
+ @JsonCreator
+ public Message(@JsonProperty("message") String message) {
+ this.message = message;
+ }
+
+ @JsonProperty
+ public String message() {
+ return message;
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (this == obj) {
+ return true;
+ }
+ if (!(obj instanceof Message)) {
+ return false;
+ }
+ Message that = (Message) obj;
+ return Objects.equals(this.message, that.message);
+ }
+
+ @Override
+ public int hashCode() {
+ return message.hashCode();
+ }
+}
diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/errors/BadRequestException.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/errors/BadRequestException.java
index bc9c7f273bf1..33bbb04b3f75 100644
--- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/errors/BadRequestException.java
+++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/errors/BadRequestException.java
@@ -24,4 +24,7 @@ public BadRequestException(String message) {
super(Response.Status.BAD_REQUEST, message);
}
+ public BadRequestException(String message, Throwable throwable) {
+ super(Response.Status.BAD_REQUEST, message, throwable);
+ }
}
diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/resources/ConnectorsResource.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/resources/ConnectorsResource.java
index fd53f5bb1dec..b930ed1fbe79 100644
--- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/resources/ConnectorsResource.java
+++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/resources/ConnectorsResource.java
@@ -31,6 +31,7 @@
import org.apache.kafka.connect.runtime.rest.entities.ConnectorOffsets;
import org.apache.kafka.connect.runtime.rest.entities.ConnectorStateInfo;
import org.apache.kafka.connect.runtime.rest.entities.CreateConnectorRequest;
+import org.apache.kafka.connect.runtime.rest.entities.Message;
import org.apache.kafka.connect.runtime.rest.entities.TaskInfo;
import org.apache.kafka.connect.runtime.rest.errors.ConnectRestException;
import org.apache.kafka.connect.util.ConnectorTaskId;
@@ -44,6 +45,7 @@
import javax.ws.rs.DELETE;
import javax.ws.rs.DefaultValue;
import javax.ws.rs.GET;
+import javax.ws.rs.PATCH;
import javax.ws.rs.POST;
import javax.ws.rs.PUT;
import javax.ws.rs.Path;
@@ -348,6 +350,23 @@ public ConnectorOffsets getOffsets(final @PathParam("connector") String connecto
return requestHandler.completeRequest(cb);
}
+ @PATCH
+ @Path("/{connector}/offsets")
+ @Operation(summary = "Alter the offsets for the specified connector")
+ public Response alterConnectorOffsets(final @Parameter(hidden = true) @QueryParam("forward") Boolean forward,
+ final @Context HttpHeaders headers, final @PathParam("connector") String connector,
+ final ConnectorOffsets offsets) throws Throwable {
+ if (offsets.offsets() == null || offsets.offsets().isEmpty()) {
+ throw new BadRequestException("Partitions / offsets need to be provided for an alter offsets request");
+ }
+
+ FutureCallback cb = new FutureCallback<>();
+ herder.alterConnectorOffsets(connector, offsets.toMap(), cb);
+ Message msg = requestHandler.completeOrForwardRequest(cb, "/connectors/" + connector + "/offsets", "PATCH", headers, offsets,
+ new TypeReference() { }, new IdentityTranslator<>(), forward);
+ return Response.ok().entity(msg).build();
+ }
+
// Check whether the connector name from the url matches the one (if there is one) provided in the connectorConfig
// object. Throw BadRequestException on mismatch, otherwise put connectorName in config
private void checkAndPutConnectorConfigName(String connectorName, Map connectorConfig) {
diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/standalone/StandaloneHerder.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/standalone/StandaloneHerder.java
index b921305143a6..096a3f53d47f 100644
--- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/standalone/StandaloneHerder.java
+++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/standalone/StandaloneHerder.java
@@ -37,7 +37,9 @@
import org.apache.kafka.connect.runtime.rest.entities.ConnectorInfo;
import org.apache.kafka.connect.runtime.rest.entities.ConnectorOffsets;
import org.apache.kafka.connect.runtime.rest.entities.ConnectorStateInfo;
+import org.apache.kafka.connect.runtime.rest.entities.Message;
import org.apache.kafka.connect.runtime.rest.entities.TaskInfo;
+import org.apache.kafka.connect.runtime.rest.errors.BadRequestException;
import org.apache.kafka.connect.storage.ClusterConfigState;
import org.apache.kafka.connect.storage.ConfigBackingStore;
import org.apache.kafka.connect.storage.MemoryConfigBackingStore;
@@ -70,7 +72,8 @@ public class StandaloneHerder extends AbstractHerder {
private final AtomicLong requestSeqNum = new AtomicLong();
private final ScheduledExecutorService requestExecutorService;
- private ClusterConfigState configState;
+ // Visible for testing
+ ClusterConfigState configState;
public StandaloneHerder(Worker worker, String kafkaClusterId,
ConnectorClientConfigOverridePolicy connectorClientConfigOverridePolicy) {
@@ -370,6 +373,22 @@ public synchronized void connectorOffsets(String connName, Callback, Map> offsets, Callback cb) {
+ if (!configState.contains(connName)) {
+ cb.onCompletion(new NotFoundException("Connector " + connName + " not found", null), null);
+ return;
+ }
+
+ if (configState.targetState(connName) != TargetState.STOPPED || configState.taskCount(connName) != 0) {
+ cb.onCompletion(new BadRequestException("Connectors must be in the STOPPED state before their offsets can be altered. " +
+ "This can be done for the specified connector by issuing a PUT request to the /connectors/" + connName + "/stop endpoint"), null);
+ return;
+ }
+
+ worker.alterConnectorOffsets(connName, configState.connectorConfig(connName), offsets, cb);
+ }
+
private void startConnector(String connName, Callback onStart) {
Map connConfigs = configState.connectorConfig(connName);
TargetState targetState = configState.targetState(connName);
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 6b35598d63df..e071534ed223 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
@@ -82,7 +82,7 @@ public class KafkaOffsetBackingStore implements OffsetBackingStore {
* @param keyConverter the worker's internal key converter that can be used to deserialize offset keys from the {@link KafkaBasedLog}
* @return an offset store backed by the given topic and Kafka clients
*/
- public static KafkaOffsetBackingStore forTask(
+ public static KafkaOffsetBackingStore readWriteStore(
String topic,
Producer producer,
Consumer consumer,
@@ -115,7 +115,7 @@ public void configure(final WorkerConfig config) {
* @param keyConverter the worker's internal key converter that can be used to deserialize offset keys from the {@link KafkaBasedLog}
* @return a read-only offset store backed by the given topic and Kafka clients
*/
- public static KafkaOffsetBackingStore forConnector(
+ public static KafkaOffsetBackingStore readOnlyStore(
String topic,
Consumer consumer,
TopicAdmin topicAdmin,
diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/util/SinkUtils.java b/connect/runtime/src/main/java/org/apache/kafka/connect/util/SinkUtils.java
index fc9562be4d82..57703f07a3e3 100644
--- a/connect/runtime/src/main/java/org/apache/kafka/connect/util/SinkUtils.java
+++ b/connect/runtime/src/main/java/org/apache/kafka/connect/util/SinkUtils.java
@@ -20,6 +20,7 @@
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.connect.runtime.rest.entities.ConnectorOffset;
import org.apache.kafka.connect.runtime.rest.entities.ConnectorOffsets;
+import org.apache.kafka.connect.runtime.rest.errors.BadRequestException;
import java.util.ArrayList;
import java.util.Collections;
@@ -52,4 +53,86 @@ public static ConnectorOffsets consumerGroupOffsetsToConnectorOffsets(Map
+ * {
+ * "kafka_topic": "topic"
+ * "kafka_partition": 3
+ * }
+ *
+ *
+ * and that the provided offsets (values in the {@code partitionOffsets} map) look like:
+ *
+ * {
+ * "kafka_offset": 1000
+ * }
+ *
+ *
+ * and then parse them into a mapping from {@link TopicPartition}s to their corresponding {@link Long}
+ * valued offsets.
+ *
+ * @param partitionOffsets the partitions to offset map that needs to be validated and parsed.
+ * @return the parsed mapping from {@link TopicPartition} to its corresponding {@link Long} valued offset.
+ *
+ * @throws BadRequestException if the provided offsets aren't in the expected format
+ */
+ public static Map parseSinkConnectorOffsets(Map, Map> partitionOffsets) {
+ Map parsedOffsetMap = new HashMap<>();
+
+ for (Map.Entry, Map> partitionOffset : partitionOffsets.entrySet()) {
+ Map partitionMap = partitionOffset.getKey();
+ if (partitionMap == null) {
+ throw new BadRequestException("The partition for a sink connector offset cannot be null or missing");
+ }
+ if (!partitionMap.containsKey(KAFKA_TOPIC_KEY) || !partitionMap.containsKey(KAFKA_PARTITION_KEY)) {
+ throw new BadRequestException(String.format("The partition for a sink connector offset must contain the keys '%s' and '%s'",
+ KAFKA_TOPIC_KEY, KAFKA_PARTITION_KEY));
+ }
+ if (partitionMap.get(KAFKA_TOPIC_KEY) == null) {
+ throw new BadRequestException("Kafka topic names must be valid strings and may not be null");
+ }
+ if (partitionMap.get(KAFKA_PARTITION_KEY) == null) {
+ throw new BadRequestException("Kafka partitions must be valid numbers and may not be null");
+ }
+ String topic = String.valueOf(partitionMap.get(KAFKA_TOPIC_KEY));
+ int partition;
+ try {
+ // We parse it this way because both "10" and 10 should be accepted as valid partition values in the REST API's
+ // JSON request payload. If it throws an exception, we should propagate it since it's indicative of a badly formatted value.
+ partition = Integer.parseInt(String.valueOf(partitionMap.get(KAFKA_PARTITION_KEY)));
+ } catch (Exception e) {
+ throw new BadRequestException("Failed to parse the following Kafka partition value in the provided offsets: '" +
+ partitionMap.get(KAFKA_PARTITION_KEY) + "'. Partition values for sink connectors need " +
+ "to be integers.", e);
+ }
+ TopicPartition tp = new TopicPartition(topic, partition);
+
+ Map offsetMap = partitionOffset.getValue();
+
+ if (offsetMap == null) {
+ // represents an offset reset
+ parsedOffsetMap.put(tp, null);
+ } else {
+ if (!offsetMap.containsKey(KAFKA_OFFSET_KEY)) {
+ throw new BadRequestException(String.format("The offset for a sink connector should either be null or contain " +
+ "the key '%s'", KAFKA_OFFSET_KEY));
+ }
+ long offset;
+ try {
+ // We parse it this way because both "1000" and 1000 should be accepted as valid offset values in the REST API's
+ // JSON request payload. If it throws an exception, we should propagate it since it's indicative of a badly formatted value.
+ offset = Long.parseLong(String.valueOf(offsetMap.get(KAFKA_OFFSET_KEY)));
+ } catch (Exception e) {
+ throw new BadRequestException("Failed to parse the following Kafka offset value in the provided offsets: '" +
+ offsetMap.get(KAFKA_OFFSET_KEY) + "'. Offset values for sink connectors need " +
+ "to be integers.", e);
+ }
+ parsedOffsetMap.put(tp, offset);
+ }
+ }
+
+ return parsedOffsetMap;
+ }
}
diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/integration/MonitorableSinkConnector.java b/connect/runtime/src/test/java/org/apache/kafka/connect/integration/MonitorableSinkConnector.java
index 8309d723130b..ffc5885ba77a 100644
--- a/connect/runtime/src/test/java/org/apache/kafka/connect/integration/MonitorableSinkConnector.java
+++ b/connect/runtime/src/test/java/org/apache/kafka/connect/integration/MonitorableSinkConnector.java
@@ -43,6 +43,9 @@ public class MonitorableSinkConnector extends SampleSinkConnector {
private static final Logger log = LoggerFactory.getLogger(MonitorableSinkConnector.class);
+ // Boolean valued configuration that determines whether MonitorableSinkConnector::alterOffsets should return true or false
+ public static final String ALTER_OFFSETS_RESULT = "alter.offsets.result";
+
private String connectorName;
private Map commonConfigs;
private ConnectorHandle connectorHandle;
@@ -84,6 +87,11 @@ public ConfigDef config() {
return new ConfigDef();
}
+ @Override
+ public boolean alterOffsets(Map connectorConfig, Map offsets) {
+ return Boolean.parseBoolean(connectorConfig.get(ALTER_OFFSETS_RESULT));
+ }
+
public static class MonitorableSinkTask extends SinkTask {
private String connectorName;
diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/integration/MonitorableSourceConnector.java b/connect/runtime/src/test/java/org/apache/kafka/connect/integration/MonitorableSourceConnector.java
index cf7a1f408f10..88fdb5a7d21d 100644
--- a/connect/runtime/src/test/java/org/apache/kafka/connect/integration/MonitorableSourceConnector.java
+++ b/connect/runtime/src/test/java/org/apache/kafka/connect/integration/MonitorableSourceConnector.java
@@ -67,6 +67,9 @@ public class MonitorableSourceConnector extends SampleSourceConnector {
public static final String TRANSACTION_BOUNDARIES_NULL = "null";
public static final String TRANSACTION_BOUNDARIES_FAIL = "fail";
+ // Boolean valued configuration that determines whether MonitorableSourceConnector::alterOffsets should return true or false
+ public static final String ALTER_OFFSETS_RESULT = "alter.offsets.result";
+
private String connectorName;
private ConnectorHandle connectorHandle;
private Map commonConfigs;
@@ -147,6 +150,11 @@ public ConnectorTransactionBoundaries canDefineTransactionBoundaries(Map connectorConfig, Map, Map> offsets) {
+ return Boolean.parseBoolean(connectorConfig.get(ALTER_OFFSETS_RESULT));
+ }
+
public static String taskId(String connectorName, int taskId) {
return connectorName + "-" + taskId;
}
diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/integration/OffsetsApiIntegrationTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/integration/OffsetsApiIntegrationTest.java
index 51fe6a38ea50..b52b55a7b2a8 100644
--- a/connect/runtime/src/test/java/org/apache/kafka/connect/integration/OffsetsApiIntegrationTest.java
+++ b/connect/runtime/src/test/java/org/apache/kafka/connect/integration/OffsetsApiIntegrationTest.java
@@ -21,6 +21,7 @@
import org.apache.kafka.clients.admin.ConsumerGroupListing;
import org.apache.kafka.connect.runtime.ConnectorConfig;
import org.apache.kafka.connect.runtime.SourceConnectorConfig;
+import org.apache.kafka.connect.runtime.distributed.DistributedConfig;
import org.apache.kafka.connect.runtime.rest.entities.ConnectorOffset;
import org.apache.kafka.connect.runtime.rest.entities.ConnectorOffsets;
import org.apache.kafka.connect.runtime.rest.errors.ConnectRestException;
@@ -35,8 +36,12 @@
import org.junit.Test;
import org.junit.experimental.categories.Category;
+import javax.ws.rs.core.Response;
+import java.util.ArrayList;
import java.util.Collection;
+import java.util.Collections;
import java.util.HashMap;
+import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.TimeUnit;
@@ -51,6 +56,8 @@
import static org.apache.kafka.connect.runtime.WorkerConfig.KEY_CONVERTER_CLASS_CONFIG;
import static org.apache.kafka.connect.runtime.WorkerConfig.OFFSET_COMMIT_INTERVAL_MS_CONFIG;
import static org.apache.kafka.connect.runtime.WorkerConfig.VALUE_CONVERTER_CLASS_CONFIG;
+import static org.hamcrest.CoreMatchers.containsString;
+import static org.hamcrest.MatcherAssert.assertThat;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertThrows;
import static org.junit.Assert.assertTrue;
@@ -65,13 +72,15 @@ public class OffsetsApiIntegrationTest {
private static final String TOPIC = "test-topic";
private static final Integer NUM_TASKS = 2;
private static final long OFFSET_COMMIT_INTERVAL_MS = TimeUnit.SECONDS.toMillis(1);
+ private static final long OFFSET_READ_TIMEOUT_MS = TimeUnit.SECONDS.toMillis(30);
private static final int NUM_WORKERS = 3;
+ private Map workerProps;
private EmbeddedConnectCluster connect;
@Before
public void setup() {
// setup Connect worker properties
- Map workerProps = new HashMap<>();
+ workerProps = new HashMap<>();
workerProps.put(OFFSET_COMMIT_INTERVAL_MS_CONFIG, String.valueOf(OFFSET_COMMIT_INTERVAL_MS));
// build a Connect cluster backed by Kafka and Zk
@@ -147,20 +156,8 @@ private void getAndVerifySinkConnectorOffsets(Map connectorConfi
connect.assertions().assertConnectorAndAtLeastNumTasksAreRunning(CONNECTOR_NAME, NUM_TASKS,
"Connector tasks did not start in time.");
- TestUtils.waitForCondition(() -> {
- ConnectorOffsets offsets = connect.connectorOffsets(CONNECTOR_NAME);
- // There should be 5 topic partitions
- if (offsets.offsets().size() != 5) {
- return false;
- }
- for (ConnectorOffset offset: offsets.offsets()) {
- assertEquals("test-topic", offset.partition().get(SinkUtils.KAFKA_TOPIC_KEY));
- if ((Integer) offset.offset().get(SinkUtils.KAFKA_OFFSET_KEY) != 10) {
- return false;
- }
- }
- return true;
- }, "Sink connector consumer group offsets should catch up to the topic end offsets");
+ waitForExpectedSinkConnectorOffsets(CONNECTOR_NAME, "test-topic", 5, 10,
+ "Sink connector consumer group offsets should catch up to the topic end offsets");
// Produce 10 more messages to each partition
for (int partition = 0; partition < 5; partition++) {
@@ -169,20 +166,8 @@ private void getAndVerifySinkConnectorOffsets(Map connectorConfi
}
}
- TestUtils.waitForCondition(() -> {
- ConnectorOffsets offsets = connect.connectorOffsets(CONNECTOR_NAME);
- // There should be 5 topic partitions
- if (offsets.offsets().size() != 5) {
- return false;
- }
- for (ConnectorOffset offset: offsets.offsets()) {
- assertEquals("test-topic", offset.partition().get(SinkUtils.KAFKA_TOPIC_KEY));
- if ((Integer) offset.offset().get(SinkUtils.KAFKA_OFFSET_KEY) != 20) {
- return false;
- }
- }
- return true;
- }, "Sink connector consumer group offsets should catch up to the topic end offsets");
+ waitForExpectedSinkConnectorOffsets(CONNECTOR_NAME, "test-topic", 5, 20,
+ "Sink connector consumer group offsets should catch up to the topic end offsets");
}
@Test
@@ -220,39 +205,373 @@ private void getAndVerifySourceConnectorOffsets(Map connectorCon
connect.assertions().assertConnectorAndAtLeastNumTasksAreRunning(CONNECTOR_NAME, NUM_TASKS,
"Connector tasks did not start in time.");
- TestUtils.waitForCondition(() -> {
- ConnectorOffsets offsets = connect.connectorOffsets(CONNECTOR_NAME);
- // The MonitorableSourceConnector has a source partition per task
- if (offsets.offsets().size() != NUM_TASKS) {
- return false;
- }
- for (ConnectorOffset offset : offsets.offsets()) {
- assertTrue(((String) offset.partition().get("task.id")).startsWith(CONNECTOR_NAME));
- if ((Integer) offset.offset().get("saved") != 10) {
- return false;
- }
- }
- return true;
- }, "Source connector offsets should reflect the expected number of records produced");
+ waitForExpectedSourceConnectorOffsets(connect, CONNECTOR_NAME, NUM_TASKS, 10,
+ "Source connector offsets should reflect the expected number of records produced");
// Each task should produce 10 more records
connectorConfigs.put(MonitorableSourceConnector.MAX_MESSAGES_PRODUCED_CONFIG, "20");
connect.configureConnector(CONNECTOR_NAME, connectorConfigs);
- TestUtils.waitForCondition(() -> {
- ConnectorOffsets offsets = connect.connectorOffsets(CONNECTOR_NAME);
- // The MonitorableSourceConnector has a source partition per task
- if (offsets.offsets().size() != NUM_TASKS) {
- return false;
- }
- for (ConnectorOffset offset : offsets.offsets()) {
- assertTrue(((String) offset.partition().get("task.id")).startsWith(CONNECTOR_NAME));
- if ((Integer) offset.offset().get("saved") != 20) {
- return false;
- }
+ waitForExpectedSourceConnectorOffsets(connect, CONNECTOR_NAME, NUM_TASKS, 20,
+ "Source connector offsets should reflect the expected number of records produced");
+ }
+
+ @Test
+ public void testAlterOffsetsNonExistentConnector() throws Exception {
+ ConnectRestException e = assertThrows(ConnectRestException.class,
+ () -> connect.alterConnectorOffsets("non-existent-connector", new ConnectorOffsets(Collections.singletonList(
+ new ConnectorOffset(Collections.emptyMap(), Collections.emptyMap())))));
+ assertEquals(404, e.errorCode());
+ }
+
+ @Test
+ public void testAlterOffsetsNonStoppedConnector() throws Exception {
+ // Create source connector
+ connect.configureConnector(CONNECTOR_NAME, baseSourceConnectorConfigs());
+ connect.assertions().assertConnectorAndAtLeastNumTasksAreRunning(CONNECTOR_NAME, NUM_TASKS,
+ "Connector tasks did not start in time.");
+
+ List offsets = new ArrayList<>();
+ // The MonitorableSourceConnector has a source partition per task
+ for (int i = 0; i < NUM_TASKS; i++) {
+ offsets.add(
+ new ConnectorOffset(Collections.singletonMap("task.id", CONNECTOR_NAME + "-" + i),
+ Collections.singletonMap("saved", 5))
+ );
+ }
+
+ // Try altering offsets for a running connector
+ ConnectRestException e = assertThrows(ConnectRestException.class,
+ () -> connect.alterConnectorOffsets(CONNECTOR_NAME, new ConnectorOffsets(offsets)));
+ assertEquals(400, e.errorCode());
+
+ connect.pauseConnector(CONNECTOR_NAME);
+ connect.assertions().assertConnectorAndExactlyNumTasksArePaused(
+ CONNECTOR_NAME,
+ NUM_TASKS,
+ "Connector did not pause in time"
+ );
+
+ // Try altering offsets for a paused (not stopped) connector
+ e = assertThrows(ConnectRestException.class,
+ () -> connect.alterConnectorOffsets(CONNECTOR_NAME, new ConnectorOffsets(offsets)));
+ assertEquals(400, e.errorCode());
+ }
+
+ @Test
+ public void testAlterSinkConnectorOffsets() throws Exception {
+ alterAndVerifySinkConnectorOffsets(baseSinkConnectorConfigs(), connect.kafka());
+ }
+
+ @Test
+ public void testAlterSinkConnectorOffsetsOverriddenConsumerGroupId() throws Exception {
+ Map connectorConfigs = baseSinkConnectorConfigs();
+ connectorConfigs.put(ConnectorConfig.CONNECTOR_CLIENT_CONSUMER_OVERRIDES_PREFIX + CommonClientConfigs.GROUP_ID_CONFIG,
+ "overridden-group-id");
+ alterAndVerifySinkConnectorOffsets(connectorConfigs, connect.kafka());
+ // Ensure that the overridden consumer group ID was the one actually used
+ try (Admin admin = connect.kafka().createAdminClient()) {
+ Collection consumerGroups = admin.listConsumerGroups().all().get();
+ assertTrue(consumerGroups.stream().anyMatch(consumerGroupListing -> "overridden-group-id".equals(consumerGroupListing.groupId())));
+ assertTrue(consumerGroups.stream().noneMatch(consumerGroupListing -> SinkUtils.consumerGroupId(CONNECTOR_NAME).equals(consumerGroupListing.groupId())));
+ }
+ }
+
+ @Test
+ public void testAlterSinkConnectorOffsetsDifferentKafkaClusterTargeted() throws Exception {
+ EmbeddedKafkaCluster kafkaCluster = new EmbeddedKafkaCluster(1, new Properties());
+
+ try (AutoCloseable ignored = kafkaCluster::stop) {
+ kafkaCluster.start();
+
+ Map connectorConfigs = baseSinkConnectorConfigs();
+ connectorConfigs.put(ConnectorConfig.CONNECTOR_CLIENT_CONSUMER_OVERRIDES_PREFIX + CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG,
+ kafkaCluster.bootstrapServers());
+ connectorConfigs.put(ConnectorConfig.CONNECTOR_CLIENT_ADMIN_OVERRIDES_PREFIX + CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG,
+ kafkaCluster.bootstrapServers());
+
+ alterAndVerifySinkConnectorOffsets(connectorConfigs, kafkaCluster);
+ }
+ }
+
+ private void alterAndVerifySinkConnectorOffsets(Map connectorConfigs, EmbeddedKafkaCluster kafkaCluster) throws Exception {
+ int numPartitions = 3;
+ int numMessages = 10;
+ kafkaCluster.createTopic(TOPIC, numPartitions);
+
+ // Produce numMessages messages to each partition
+ for (int partition = 0; partition < numPartitions; partition++) {
+ for (int message = 0; message < numMessages; message++) {
+ kafkaCluster.produce(TOPIC, partition, "key", "value");
}
- return true;
- }, "Source connector offsets should reflect the expected number of records produced");
+ }
+ // Create sink connector
+ connect.configureConnector(CONNECTOR_NAME, connectorConfigs);
+ connect.assertions().assertConnectorAndAtLeastNumTasksAreRunning(CONNECTOR_NAME, NUM_TASKS,
+ "Connector tasks did not start in time.");
+
+ waitForExpectedSinkConnectorOffsets(CONNECTOR_NAME, "test-topic", numPartitions, numMessages,
+ "Sink connector consumer group offsets should catch up to the topic end offsets");
+
+ connect.stopConnector(CONNECTOR_NAME);
+ connect.assertions().assertConnectorIsStopped(
+ CONNECTOR_NAME,
+ "Connector did not stop in time"
+ );
+
+ // Delete the offset of one partition; alter the offsets of the others
+ List offsetsToAlter = new ArrayList<>();
+ Map partition = new HashMap<>();
+ partition.put(SinkUtils.KAFKA_TOPIC_KEY, TOPIC);
+ partition.put(SinkUtils.KAFKA_PARTITION_KEY, 0);
+ offsetsToAlter.add(new ConnectorOffset(partition, null));
+
+ for (int i = 1; i < numPartitions; i++) {
+ partition = new HashMap<>();
+ partition.put(SinkUtils.KAFKA_TOPIC_KEY, TOPIC);
+ partition.put(SinkUtils.KAFKA_PARTITION_KEY, i);
+ offsetsToAlter.add(new ConnectorOffset(partition, Collections.singletonMap(SinkUtils.KAFKA_OFFSET_KEY, 5)));
+ }
+
+ String response = connect.alterConnectorOffsets(CONNECTOR_NAME, new ConnectorOffsets(offsetsToAlter));
+ assertThat(response, containsString("The Connect framework-managed offsets for this connector have been altered successfully. " +
+ "However, if this connector manages offsets externally, they will need to be manually altered in the system that the connector uses."));
+
+ waitForExpectedSinkConnectorOffsets(CONNECTOR_NAME, "test-topic", numPartitions - 1, 5,
+ "Sink connector consumer group offsets should reflect the altered offsets");
+
+ // Update the connector's configs; this time expect SinkConnector::alterOffsets to return true
+ connectorConfigs.put(MonitorableSinkConnector.ALTER_OFFSETS_RESULT, "true");
+ connect.configureConnector(CONNECTOR_NAME, connectorConfigs);
+
+ // Alter offsets again while the connector is still in a stopped state
+ offsetsToAlter.clear();
+ for (int i = 1; i < numPartitions; i++) {
+ partition = new HashMap<>();
+ partition.put(SinkUtils.KAFKA_TOPIC_KEY, TOPIC);
+ partition.put(SinkUtils.KAFKA_PARTITION_KEY, i);
+ offsetsToAlter.add(new ConnectorOffset(partition, Collections.singletonMap(SinkUtils.KAFKA_OFFSET_KEY, 3)));
+ }
+
+ response = connect.alterConnectorOffsets(CONNECTOR_NAME, new ConnectorOffsets(offsetsToAlter));
+ assertThat(response, containsString("The offsets for this connector have been altered successfully"));
+
+ waitForExpectedSinkConnectorOffsets(CONNECTOR_NAME, "test-topic", numPartitions - 1, 3,
+ "Sink connector consumer group offsets should reflect the altered offsets");
+
+ // Resume the connector and expect its offsets to catch up to the latest offsets
+ connect.resumeConnector(CONNECTOR_NAME);
+ connect.assertions().assertConnectorAndExactlyNumTasksAreRunning(
+ CONNECTOR_NAME,
+ NUM_TASKS,
+ "Connector tasks did not resume in time"
+ );
+ waitForExpectedSinkConnectorOffsets(CONNECTOR_NAME, "test-topic", numPartitions, 10,
+ "Sink connector consumer group offsets should catch up to the topic end offsets");
+ }
+
+ @Test
+ public void testAlterSinkConnectorOffsetsZombieSinkTasks() throws Exception {
+ connect.kafka().createTopic(TOPIC, 1);
+
+ // Produce 10 messages
+ for (int message = 0; message < 10; message++) {
+ connect.kafka().produce(TOPIC, 0, "key", "value");
+ }
+
+ // Configure a sink connector whose sink task blocks in its stop method
+ Map connectorConfigs = new HashMap<>();
+ connectorConfigs.put(CONNECTOR_CLASS_CONFIG, BlockingConnectorTest.BlockingSinkConnector.class.getName());
+ connectorConfigs.put(TOPICS_CONFIG, TOPIC);
+ connectorConfigs.put("block", "Task::stop");
+
+ connect.configureConnector(CONNECTOR_NAME, connectorConfigs);
+ connect.assertions().assertConnectorAndAtLeastNumTasksAreRunning(CONNECTOR_NAME, 1,
+ "Connector tasks did not start in time.");
+
+ connect.stopConnector(CONNECTOR_NAME);
+
+ // Try to delete the offsets for the single topic partition
+ Map partition = new HashMap<>();
+ partition.put(SinkUtils.KAFKA_TOPIC_KEY, TOPIC);
+ partition.put(SinkUtils.KAFKA_PARTITION_KEY, 0);
+ List offsetsToAlter = Collections.singletonList(new ConnectorOffset(partition, null));
+
+ ConnectRestException e = assertThrows(ConnectRestException.class,
+ () -> connect.alterConnectorOffsets(CONNECTOR_NAME, new ConnectorOffsets(offsetsToAlter)));
+ assertThat(e.getMessage(), containsString("zombie sink task"));
+ }
+
+ @Test
+ public void testAlterSinkConnectorOffsetsInvalidRequestBody() throws Exception {
+ // Create a sink connector and stop it
+ connect.configureConnector(CONNECTOR_NAME, baseSinkConnectorConfigs());
+ connect.assertions().assertConnectorAndAtLeastNumTasksAreRunning(CONNECTOR_NAME, NUM_TASKS,
+ "Connector tasks did not start in time.");
+ connect.stopConnector(CONNECTOR_NAME);
+ connect.assertions().assertConnectorIsStopped(
+ CONNECTOR_NAME,
+ "Connector did not stop in time"
+ );
+ String url = connect.endpointForResource(String.format("connectors/%s/offsets", CONNECTOR_NAME));
+
+ String content = "{}";
+ try (Response response = connect.requestPatch(url, content)) {
+ assertEquals(400, response.getStatus());
+ assertThat(response.getEntity().toString(), containsString("Partitions / offsets need to be provided for an alter offsets request"));
+ }
+
+ content = "{\"offsets\": []}";
+ try (Response response = connect.requestPatch(url, content)) {
+ assertEquals(400, response.getStatus());
+ assertThat(response.getEntity().toString(), containsString("Partitions / offsets need to be provided for an alter offsets request"));
+ }
+
+ content = "{\"offsets\": [{}]}";
+ try (Response response = connect.requestPatch(url, content)) {
+ assertEquals(400, response.getStatus());
+ assertThat(response.getEntity().toString(), containsString("The partition for a sink connector offset cannot be null or missing"));
+ }
+
+ content = "{\"offsets\": [{\"partition\": null, \"offset\": null}]}";
+ try (Response response = connect.requestPatch(url, content)) {
+ assertEquals(400, response.getStatus());
+ assertThat(response.getEntity().toString(), containsString("The partition for a sink connector offset cannot be null or missing"));
+ }
+
+ content = "{\"offsets\": [{\"partition\": {}, \"offset\": null}]}";
+ try (Response response = connect.requestPatch(url, content)) {
+ assertEquals(400, response.getStatus());
+ assertThat(response.getEntity().toString(), containsString("The partition for a sink connector offset must contain the keys 'kafka_topic' and 'kafka_partition'"));
+ }
+
+ content = "{\"offsets\": [{\"partition\": {\"kafka_topic\": \"test\", \"kafka_partition\": \"not a number\"}, \"offset\": null}]}";
+ try (Response response = connect.requestPatch(url, content)) {
+ assertEquals(400, response.getStatus());
+ assertThat(response.getEntity().toString(), containsString("Partition values for sink connectors need to be integers"));
+ }
+
+ content = "{\"offsets\": [{\"partition\": {\"kafka_topic\": \"test\", \"kafka_partition\": 1}, \"offset\": {}}]}";
+ try (Response response = connect.requestPatch(url, content)) {
+ assertEquals(400, response.getStatus());
+ assertThat(response.getEntity().toString(), containsString("The offset for a sink connector should either be null or contain the key 'kafka_offset'"));
+ }
+
+ content = "{\"offsets\": [{\"partition\": {\"kafka_topic\": \"test\", \"kafka_partition\": 1}, \"offset\": {\"kafka_offset\": \"not a number\"}}]}";
+ try (Response response = connect.requestPatch(url, content)) {
+ assertEquals(400, response.getStatus());
+ assertThat(response.getEntity().toString(), containsString("Offset values for sink connectors need to be integers"));
+ }
+ }
+
+ @Test
+ public void testAlterSourceConnectorOffsets() throws Exception {
+ alterAndVerifySourceConnectorOffsets(connect, baseSourceConnectorConfigs());
+ }
+
+ @Test
+ public void testAlterSourceConnectorOffsetsCustomOffsetsTopic() throws Exception {
+ Map connectorConfigs = baseSourceConnectorConfigs();
+ connectorConfigs.put(SourceConnectorConfig.OFFSETS_TOPIC_CONFIG, "custom-offsets-topic");
+ alterAndVerifySourceConnectorOffsets(connect, connectorConfigs);
+ }
+
+ @Test
+ public void testAlterSourceConnectorOffsetsDifferentKafkaClusterTargeted() throws Exception {
+ EmbeddedKafkaCluster kafkaCluster = new EmbeddedKafkaCluster(1, new Properties());
+
+ try (AutoCloseable ignored = kafkaCluster::stop) {
+ kafkaCluster.start();
+
+ Map connectorConfigs = baseSourceConnectorConfigs();
+ connectorConfigs.put(ConnectorConfig.CONNECTOR_CLIENT_PRODUCER_OVERRIDES_PREFIX + CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG,
+ kafkaCluster.bootstrapServers());
+ connectorConfigs.put(ConnectorConfig.CONNECTOR_CLIENT_ADMIN_OVERRIDES_PREFIX + CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG,
+ kafkaCluster.bootstrapServers());
+
+ alterAndVerifySourceConnectorOffsets(connect, connectorConfigs);
+ }
+ }
+
+ @Test
+ public void testAlterSourceConnectorOffsetsExactlyOnceSupportEnabled() throws Exception {
+ Properties brokerProps = new Properties();
+ brokerProps.put("transaction.state.log.replication.factor", "1");
+ brokerProps.put("transaction.state.log.min.isr", "1");
+ workerProps.put(DistributedConfig.EXACTLY_ONCE_SOURCE_SUPPORT_CONFIG, "enabled");
+ EmbeddedConnectCluster exactlyOnceSupportEnabledConnectCluster = new EmbeddedConnectCluster.Builder()
+ .name("connect-cluster")
+ .brokerProps(brokerProps)
+ .numWorkers(NUM_WORKERS)
+ .workerProps(workerProps)
+ .build();
+ exactlyOnceSupportEnabledConnectCluster.start();
+
+ try (AutoCloseable ignored = exactlyOnceSupportEnabledConnectCluster::stop) {
+ alterAndVerifySourceConnectorOffsets(exactlyOnceSupportEnabledConnectCluster, baseSourceConnectorConfigs());
+ }
+ }
+
+ public void alterAndVerifySourceConnectorOffsets(EmbeddedConnectCluster connect, Map connectorConfigs) throws Exception {
+ // Create source connector
+ connect.configureConnector(CONNECTOR_NAME, connectorConfigs);
+ connect.assertions().assertConnectorAndAtLeastNumTasksAreRunning(CONNECTOR_NAME, NUM_TASKS,
+ "Connector tasks did not start in time.");
+
+ waitForExpectedSourceConnectorOffsets(connect, CONNECTOR_NAME, NUM_TASKS, 10,
+ "Source connector offsets should reflect the expected number of records produced");
+
+ connect.stopConnector(CONNECTOR_NAME);
+ connect.assertions().assertConnectorIsStopped(
+ CONNECTOR_NAME,
+ "Connector did not stop in time"
+ );
+
+ List offsetsToAlter = new ArrayList<>();
+ // The MonitorableSourceConnector has a source partition per task
+ for (int i = 0; i < NUM_TASKS; i++) {
+ offsetsToAlter.add(
+ new ConnectorOffset(Collections.singletonMap("task.id", CONNECTOR_NAME + "-" + i),
+ Collections.singletonMap("saved", 5))
+ );
+ }
+
+ String response = connect.alterConnectorOffsets(CONNECTOR_NAME, new ConnectorOffsets(offsetsToAlter));
+ assertThat(response, containsString("The Connect framework-managed offsets for this connector have been altered successfully. " +
+ "However, if this connector manages offsets externally, they will need to be manually altered in the system that the connector uses."));
+
+ waitForExpectedSourceConnectorOffsets(connect, CONNECTOR_NAME, NUM_TASKS, 5,
+ "Source connector offsets should reflect the altered offsets");
+
+ // Update the connector's configs; this time expect SourceConnector::alterOffsets to return true
+ connectorConfigs.put(MonitorableSourceConnector.ALTER_OFFSETS_RESULT, "true");
+ connect.configureConnector(CONNECTOR_NAME, connectorConfigs);
+
+ // Alter offsets again while connector is in stopped state
+ offsetsToAlter = new ArrayList<>();
+ // The MonitorableSourceConnector has a source partition per task
+ for (int i = 0; i < NUM_TASKS; i++) {
+ offsetsToAlter.add(
+ new ConnectorOffset(Collections.singletonMap("task.id", CONNECTOR_NAME + "-" + i),
+ Collections.singletonMap("saved", 7))
+ );
+ }
+
+ response = connect.alterConnectorOffsets(CONNECTOR_NAME, new ConnectorOffsets(offsetsToAlter));
+ assertThat(response, containsString("The offsets for this connector have been altered successfully"));
+
+ waitForExpectedSourceConnectorOffsets(connect, CONNECTOR_NAME, NUM_TASKS, 7,
+ "Source connector offsets should reflect the altered offsets");
+
+ // Resume the connector and expect its offsets to catch up to the latest offsets
+ connect.resumeConnector(CONNECTOR_NAME);
+ connect.assertions().assertConnectorAndExactlyNumTasksAreRunning(
+ CONNECTOR_NAME,
+ NUM_TASKS,
+ "Connector tasks did not resume in time"
+ );
+ waitForExpectedSourceConnectorOffsets(connect, CONNECTOR_NAME, NUM_TASKS, 10,
+ "Source connector offsets should reflect the expected number of records produced");
}
private Map baseSinkConnectorConfigs() {
@@ -278,4 +597,70 @@ private Map baseSourceConnectorConfigs() {
props.put(DEFAULT_TOPIC_CREATION_PREFIX + PARTITIONS_CONFIG, "1");
return props;
}
+
+ /**
+ * Verify whether the actual consumer group offsets for a sink connector match the expected offsets. The verification
+ * is done using the `GET /connectors/{connector}/offsets` REST API which is repeatedly queried until the offsets match
+ * or the {@link #OFFSET_READ_TIMEOUT_MS timeout} is reached. Note that this assumes the following:
+ *
+ * - The sink connector is consuming from a single Kafka topic
+ * - The expected offset for each partition in the topic is the same
+ *
+ *
+ * @param connectorName the name of the sink connector whose offsets are to be verified
+ * @param expectedTopic the name of the Kafka topic that the sink connector is consuming from
+ * @param expectedPartitions the number of partitions that exist for the Kafka topic
+ * @param expectedOffset the expected consumer group offset for each partition
+ * @param conditionDetails the condition that we're waiting to achieve (for example: Sink connector should process
+ * 10 records)
+ * @throws InterruptedException if the thread is interrupted while waiting for the actual offsets to match the expected offsets
+ */
+ private void waitForExpectedSinkConnectorOffsets(String connectorName, String expectedTopic, int expectedPartitions,
+ int expectedOffset, String conditionDetails) throws InterruptedException {
+ TestUtils.waitForCondition(() -> {
+ ConnectorOffsets offsets = connect.connectorOffsets(connectorName);
+ if (offsets.offsets().size() != expectedPartitions) {
+ return false;
+ }
+ for (ConnectorOffset offset: offsets.offsets()) {
+ assertEquals(expectedTopic, offset.partition().get(SinkUtils.KAFKA_TOPIC_KEY));
+ if ((Integer) offset.offset().get(SinkUtils.KAFKA_OFFSET_KEY) != expectedOffset) {
+ return false;
+ }
+ }
+ return true;
+ }, OFFSET_READ_TIMEOUT_MS, conditionDetails);
+ }
+
+ /**
+ * Verify whether the actual offsets for a source connector match the expected offsets. The verification is done using the
+ * `GET /connectors/{connector}/offsets` REST API which is repeatedly queried until the offsets match or the
+ * {@link #OFFSET_READ_TIMEOUT_MS timeout} is reached. Note that this assumes that the source connector is a
+ * {@link MonitorableSourceConnector}
+ *
+ * @param connect the Connect cluster that is running the source connector
+ * @param connectorName the name of the source connector whose offsets are to be verified
+ * @param numTasks the number of tasks for the source connector
+ * @param expectedOffset the expected offset for each source partition
+ * @param conditionDetails the condition that we're waiting to achieve (for example: Source connector should process
+ * 10 records)
+ * @throws InterruptedException if the thread is interrupted while waiting for the actual offsets to match the expected offsets
+ */
+ private void waitForExpectedSourceConnectorOffsets(EmbeddedConnectCluster connect, String connectorName, int numTasks,
+ int expectedOffset, String conditionDetails) throws InterruptedException {
+ TestUtils.waitForCondition(() -> {
+ ConnectorOffsets offsets = connect.connectorOffsets(connectorName);
+ // The MonitorableSourceConnector has a source partition per task
+ if (offsets.offsets().size() != numTasks) {
+ return false;
+ }
+ for (ConnectorOffset offset : offsets.offsets()) {
+ assertTrue(((String) offset.partition().get("task.id")).startsWith(CONNECTOR_NAME));
+ if ((Integer) offset.offset().get("saved") != expectedOffset) {
+ return false;
+ }
+ }
+ return true;
+ }, OFFSET_READ_TIMEOUT_MS, conditionDetails);
+ }
}
diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerTest.java
index 0614656aee48..fa9d87c29e12 100644
--- a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerTest.java
+++ b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerTest.java
@@ -18,11 +18,16 @@
import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.clients.admin.AlterConsumerGroupOffsetsOptions;
+import org.apache.kafka.clients.admin.AlterConsumerGroupOffsetsResult;
+import org.apache.kafka.clients.admin.DeleteConsumerGroupOffsetsOptions;
+import org.apache.kafka.clients.admin.DeleteConsumerGroupOffsetsResult;
import org.apache.kafka.clients.admin.FenceProducersResult;
import org.apache.kafka.clients.admin.ListConsumerGroupOffsetsOptions;
import org.apache.kafka.clients.admin.ListConsumerGroupOffsetsResult;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
+import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.KafkaFuture;
@@ -32,6 +37,7 @@
import org.apache.kafka.common.config.ConfigException;
import org.apache.kafka.common.config.provider.MockFileConfigProvider;
import org.apache.kafka.common.errors.ClusterAuthorizationException;
+import org.apache.kafka.common.internals.KafkaFutureImpl;
import org.apache.kafka.common.metrics.MetricsReporter;
import org.apache.kafka.common.metrics.stats.Avg;
import org.apache.kafka.common.utils.MockTime;
@@ -54,6 +60,7 @@
import org.apache.kafka.connect.runtime.isolation.Plugins.ClassLoaderUsage;
import org.apache.kafka.connect.runtime.rest.entities.ConnectorOffsets;
import org.apache.kafka.connect.runtime.rest.entities.ConnectorStateInfo;
+import org.apache.kafka.connect.runtime.rest.entities.Message;
import org.apache.kafka.connect.runtime.standalone.StandaloneConfig;
import org.apache.kafka.connect.sink.SinkConnector;
import org.apache.kafka.connect.sink.SinkRecord;
@@ -67,7 +74,9 @@
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.OffsetStorageWriter;
import org.apache.kafka.connect.storage.StatusBackingStore;
+import org.apache.kafka.connect.util.Callback;
import org.apache.kafka.connect.util.ConnectorTaskId;
import org.apache.kafka.connect.util.FutureCallback;
import org.apache.kafka.connect.util.ParameterizedTest;
@@ -78,6 +87,7 @@
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
+import org.mockito.AdditionalAnswers;
import org.mockito.ArgumentCaptor;
import org.mockito.Mock;
import org.mockito.MockedConstruction;
@@ -92,6 +102,7 @@
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
@@ -100,6 +111,7 @@
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Function;
@@ -134,6 +146,7 @@
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.ArgumentMatchers.anyMap;
+import static org.mockito.ArgumentMatchers.anySet;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.ArgumentMatchers.same;
@@ -144,6 +157,7 @@
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.mockConstructionWithAnswer;
import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.timeout;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.verifyNoMoreInteractions;
@@ -589,7 +603,7 @@ public void testAddRemoveSourceTask() {
Map origProps = Collections.singletonMap(TaskConfig.TASK_CLASS_CONFIG, TestSourceTask.class.getName());
worker = new Worker(WORKER_ID, new MockTime(), plugins, config, offsetBackingStore, executorService,
- noneConnectorClientConfigOverridePolicy);
+ noneConnectorClientConfigOverridePolicy, null);
worker.herder = herder;
worker.start();
@@ -628,7 +642,7 @@ public void testAddRemoveSinkTask() {
Map origProps = Collections.singletonMap(TaskConfig.TASK_CLASS_CONFIG, TestSinkTask.class.getName());
worker = new Worker(WORKER_ID, new MockTime(), plugins, config, offsetBackingStore, executorService,
- noneConnectorClientConfigOverridePolicy);
+ noneConnectorClientConfigOverridePolicy, null);
worker.herder = herder;
worker.start();
@@ -688,7 +702,7 @@ public void testAddRemoveExactlyOnceSourceTask() {
Map origProps = Collections.singletonMap(TaskConfig.TASK_CLASS_CONFIG, TestSourceTask.class.getName());
worker = new Worker(WORKER_ID, new MockTime(), plugins, config, offsetBackingStore, executorService,
- noneConnectorClientConfigOverridePolicy);
+ noneConnectorClientConfigOverridePolicy, null);
worker.herder = herder;
worker.start();
@@ -746,7 +760,8 @@ public void testTaskStatusMetricsStatuses() {
config,
offsetBackingStore,
executorService,
- noneConnectorClientConfigOverridePolicy);
+ noneConnectorClientConfigOverridePolicy,
+ null);
worker.herder = herder;
@@ -872,7 +887,7 @@ public void testCleanupTasksOnStop() {
TaskConfig taskConfig = new TaskConfig(origProps);
worker = new Worker(WORKER_ID, new MockTime(), plugins, config, offsetBackingStore, executorService,
- noneConnectorClientConfigOverridePolicy);
+ noneConnectorClientConfigOverridePolicy, null);
worker.herder = herder;
worker.start();
assertStatistics(worker, 0, 0);
@@ -916,7 +931,7 @@ public void testConverterOverrides() {
mockExecutorFakeSubmit(WorkerTask.class);
worker = new Worker(WORKER_ID, new MockTime(), plugins, config, offsetBackingStore, executorService,
- noneConnectorClientConfigOverridePolicy);
+ noneConnectorClientConfigOverridePolicy, null);
worker.herder = herder;
worker.start();
assertStatistics(worker, 0, 0);
@@ -1286,7 +1301,7 @@ public void testOffsetStoreForRegularSourceConnector() {
connectorProps.put(ConnectorConfig.TASKS_MAX_CONFIG, "1");
SourceConnectorConfig sourceConfig = new SourceConnectorConfig(plugins, connectorProps, enableTopicCreation);
// With no connector-specific offsets topic in the config, we should only use the worker-global offsets store
- ConnectorOffsetBackingStore connectorStore = worker.offsetStoreForRegularSourceConnector(sourceConfig, CONNECTOR_ID, sourceConnector);
+ ConnectorOffsetBackingStore connectorStore = worker.offsetStoreForRegularSourceConnector(sourceConfig, CONNECTOR_ID, sourceConnector, null);
assertTrue(connectorStore.hasWorkerGlobalStore());
assertFalse(connectorStore.hasConnectorSpecificStore());
@@ -1294,7 +1309,7 @@ public void testOffsetStoreForRegularSourceConnector() {
sourceConfig = new SourceConnectorConfig(plugins, connectorProps, enableTopicCreation);
// With a connector-specific offsets topic in the config (whose name differs from the worker's offsets topic), we should use both a
// connector-specific store and the worker-global store
- connectorStore = worker.offsetStoreForRegularSourceConnector(sourceConfig, CONNECTOR_ID, sourceConnector);
+ connectorStore = worker.offsetStoreForRegularSourceConnector(sourceConfig, CONNECTOR_ID, sourceConnector, null);
assertTrue(connectorStore.hasWorkerGlobalStore());
assertTrue(connectorStore.hasConnectorSpecificStore());
@@ -1302,7 +1317,7 @@ public void testOffsetStoreForRegularSourceConnector() {
sourceConfig = new SourceConnectorConfig(plugins, connectorProps, enableTopicCreation);
// With a connector-specific offsets topic in the config whose name matches the worker's offsets topic, and no overridden bootstrap.servers
// for the connector, we should only use a connector-specific offsets store
- connectorStore = worker.offsetStoreForRegularSourceConnector(sourceConfig, CONNECTOR_ID, sourceConnector);
+ connectorStore = worker.offsetStoreForRegularSourceConnector(sourceConfig, CONNECTOR_ID, sourceConnector, null);
assertFalse(connectorStore.hasWorkerGlobalStore());
assertTrue(connectorStore.hasConnectorSpecificStore());
@@ -1310,7 +1325,7 @@ public void testOffsetStoreForRegularSourceConnector() {
sourceConfig = new SourceConnectorConfig(plugins, connectorProps, enableTopicCreation);
// With a connector-specific offsets topic in the config whose name matches the worker's offsets topic, and an overridden bootstrap.servers
// for the connector that exactly matches the worker's, we should only use a connector-specific offsets store
- connectorStore = worker.offsetStoreForRegularSourceConnector(sourceConfig, CONNECTOR_ID, sourceConnector);
+ connectorStore = worker.offsetStoreForRegularSourceConnector(sourceConfig, CONNECTOR_ID, sourceConnector, null);
assertFalse(connectorStore.hasWorkerGlobalStore());
assertTrue(connectorStore.hasConnectorSpecificStore());
@@ -1318,7 +1333,7 @@ public void testOffsetStoreForRegularSourceConnector() {
sourceConfig = new SourceConnectorConfig(plugins, connectorProps, enableTopicCreation);
// With a connector-specific offsets topic in the config whose name matches the worker's offsets topic, and an overridden bootstrap.servers
// for the connector that doesn't match the worker's, we should use both a connector-specific store and the worker-global store
- connectorStore = worker.offsetStoreForRegularSourceConnector(sourceConfig, CONNECTOR_ID, sourceConnector);
+ connectorStore = worker.offsetStoreForRegularSourceConnector(sourceConfig, CONNECTOR_ID, sourceConnector, null);
assertTrue(connectorStore.hasWorkerGlobalStore());
assertTrue(connectorStore.hasConnectorSpecificStore());
@@ -1326,7 +1341,7 @@ public void testOffsetStoreForRegularSourceConnector() {
sourceConfig = new SourceConnectorConfig(plugins, connectorProps, enableTopicCreation);
// With no connector-specific offsets topic in the config, even with an overridden bootstrap.servers
// for the connector that doesn't match the worker's, we should still only use the worker-global offsets store
- connectorStore = worker.offsetStoreForRegularSourceConnector(sourceConfig, CONNECTOR_ID, sourceConnector);
+ connectorStore = worker.offsetStoreForRegularSourceConnector(sourceConfig, CONNECTOR_ID, sourceConnector, null);
assertTrue(connectorStore.hasWorkerGlobalStore());
assertFalse(connectorStore.hasConnectorSpecificStore());
@@ -1361,7 +1376,7 @@ public void testOffsetStoreForExactlyOnceSourceConnector() {
connectorProps.put(ConnectorConfig.TASKS_MAX_CONFIG, "1");
SourceConnectorConfig sourceConfig = new SourceConnectorConfig(plugins, connectorProps, enableTopicCreation);
// With no connector-specific offsets topic in the config, we should only use a connector-specific offsets store
- ConnectorOffsetBackingStore connectorStore = worker.offsetStoreForExactlyOnceSourceConnector(sourceConfig, CONNECTOR_ID, sourceConnector);
+ ConnectorOffsetBackingStore connectorStore = worker.offsetStoreForExactlyOnceSourceConnector(sourceConfig, CONNECTOR_ID, sourceConnector, null);
assertFalse(connectorStore.hasWorkerGlobalStore());
assertTrue(connectorStore.hasConnectorSpecificStore());
@@ -1369,7 +1384,7 @@ public void testOffsetStoreForExactlyOnceSourceConnector() {
sourceConfig = new SourceConnectorConfig(plugins, connectorProps, enableTopicCreation);
// With a connector-specific offsets topic in the config (whose name differs from the worker's offsets topic), we should use both a
// connector-specific store and the worker-global store
- connectorStore = worker.offsetStoreForExactlyOnceSourceConnector(sourceConfig, CONNECTOR_ID, sourceConnector);
+ connectorStore = worker.offsetStoreForExactlyOnceSourceConnector(sourceConfig, CONNECTOR_ID, sourceConnector, null);
assertTrue(connectorStore.hasWorkerGlobalStore());
assertTrue(connectorStore.hasConnectorSpecificStore());
@@ -1377,7 +1392,7 @@ public void testOffsetStoreForExactlyOnceSourceConnector() {
sourceConfig = new SourceConnectorConfig(plugins, connectorProps, enableTopicCreation);
// With a connector-specific offsets topic in the config whose name matches the worker's offsets topic, and no overridden bootstrap.servers
// for the connector, we should only use a connector-specific store
- connectorStore = worker.offsetStoreForExactlyOnceSourceConnector(sourceConfig, CONNECTOR_ID, sourceConnector);
+ connectorStore = worker.offsetStoreForExactlyOnceSourceConnector(sourceConfig, CONNECTOR_ID, sourceConnector, null);
assertFalse(connectorStore.hasWorkerGlobalStore());
assertTrue(connectorStore.hasConnectorSpecificStore());
@@ -1385,7 +1400,7 @@ public void testOffsetStoreForExactlyOnceSourceConnector() {
sourceConfig = new SourceConnectorConfig(plugins, connectorProps, enableTopicCreation);
// With a connector-specific offsets topic in the config whose name matches the worker's offsets topic, and an overridden bootstrap.servers
// for the connector that exactly matches the worker's, we should only use a connector-specific store
- connectorStore = worker.offsetStoreForExactlyOnceSourceConnector(sourceConfig, CONNECTOR_ID, sourceConnector);
+ connectorStore = worker.offsetStoreForExactlyOnceSourceConnector(sourceConfig, CONNECTOR_ID, sourceConnector, null);
assertFalse(connectorStore.hasWorkerGlobalStore());
assertTrue(connectorStore.hasConnectorSpecificStore());
@@ -1393,7 +1408,7 @@ public void testOffsetStoreForExactlyOnceSourceConnector() {
sourceConfig = new SourceConnectorConfig(plugins, connectorProps, enableTopicCreation);
// With a connector-specific offsets topic in the config whose name matches the worker's offsets topic, and an overridden bootstrap.servers
// for the connector that doesn't match the worker's, we should use both a connector-specific store and the worker-global store
- connectorStore = worker.offsetStoreForExactlyOnceSourceConnector(sourceConfig, CONNECTOR_ID, sourceConnector);
+ connectorStore = worker.offsetStoreForExactlyOnceSourceConnector(sourceConfig, CONNECTOR_ID, sourceConnector, null);
assertTrue(connectorStore.hasWorkerGlobalStore());
assertTrue(connectorStore.hasConnectorSpecificStore());
@@ -1401,7 +1416,7 @@ public void testOffsetStoreForExactlyOnceSourceConnector() {
sourceConfig = new SourceConnectorConfig(plugins, connectorProps, enableTopicCreation);
// With no connector-specific offsets topic in the config and an overridden bootstrap.servers
// for the connector that doesn't match the worker's, we should use both a connector-specific store and the worker-global store
- connectorStore = worker.offsetStoreForExactlyOnceSourceConnector(sourceConfig, CONNECTOR_ID, sourceConnector);
+ connectorStore = worker.offsetStoreForExactlyOnceSourceConnector(sourceConfig, CONNECTOR_ID, sourceConnector, null);
assertTrue(connectorStore.hasWorkerGlobalStore());
assertTrue(connectorStore.hasConnectorSpecificStore());
@@ -1635,7 +1650,7 @@ public void testExecutorServiceShutdown() throws InterruptedException {
worker = new Worker(WORKER_ID, new MockTime(), plugins, config,
offsetBackingStore, executorService,
- noneConnectorClientConfigOverridePolicy);
+ noneConnectorClientConfigOverridePolicy, null);
worker.start();
assertEquals(Collections.emptySet(), worker.connectorNames());
@@ -1655,7 +1670,7 @@ public void testExecutorServiceShutdownWhenTerminationFails() throws Interrupted
when(executorService.awaitTermination(1000L, TimeUnit.MILLISECONDS)).thenReturn(false);
worker = new Worker(WORKER_ID, new MockTime(), plugins, config,
offsetBackingStore, executorService,
- noneConnectorClientConfigOverridePolicy);
+ noneConnectorClientConfigOverridePolicy, null);
worker.start();
assertEquals(Collections.emptySet(), worker.connectorNames());
@@ -1676,7 +1691,7 @@ public void testExecutorServiceShutdownWhenTerminationThrowsException() throws I
when(executorService.awaitTermination(1000L, TimeUnit.MILLISECONDS)).thenThrow(new InterruptedException("interrupt"));
worker = new Worker(WORKER_ID, new MockTime(), plugins, config,
offsetBackingStore, executorService,
- noneConnectorClientConfigOverridePolicy);
+ noneConnectorClientConfigOverridePolicy, null);
worker.start();
assertEquals(Collections.emptySet(), worker.connectorNames());
@@ -1692,6 +1707,11 @@ public void testExecutorServiceShutdownWhenTerminationThrowsException() throws I
@SuppressWarnings("unchecked")
public void testZombieFencing() {
Admin admin = mock(Admin.class);
+ AtomicReference> adminConfig = new AtomicReference<>();
+ Function, Admin> mockAdminConstructor = actualAdminConfig -> {
+ adminConfig.set(actualAdminConfig);
+ return admin;
+ };
FenceProducersResult fenceProducersResult = mock(FenceProducersResult.class);
KafkaFuture fenceProducersFuture = mock(KafkaFuture.class);
KafkaFuture expectedZombieFenceFuture = mock(KafkaFuture.class);
@@ -1703,21 +1723,15 @@ public void testZombieFencing() {
mockGenericIsolation();
worker = new Worker(WORKER_ID, new MockTime(), plugins, config, offsetBackingStore, executorService,
- allConnectorClientConfigOverridePolicy);
+ allConnectorClientConfigOverridePolicy, mockAdminConstructor);
worker.herder = herder;
worker.start();
Map connectorConfig = anyConnectorConfigMap();
connectorConfig.put(CONNECTOR_CLIENT_ADMIN_OVERRIDES_PREFIX + RETRY_BACKOFF_MS_CONFIG, "4761");
- AtomicReference> adminConfig = new AtomicReference<>();
- Function, Admin> mockAdminConstructor = actualAdminConfig -> {
- adminConfig.set(actualAdminConfig);
- return admin;
- };
-
KafkaFuture actualZombieFenceFuture =
- worker.fenceZombies(CONNECTOR_ID, 12, connectorConfig, mockAdminConstructor);
+ worker.fenceZombies(CONNECTOR_ID, 12, connectorConfig);
assertEquals(expectedZombieFenceFuture, actualZombieFenceFuture);
assertNotNull(adminConfig.get());
@@ -1737,15 +1751,14 @@ public void testGetSinkConnectorOffsets() throws Exception {
String connectorClass = SampleSinkConnector.class.getName();
connectorProps.put(CONNECTOR_CLASS_CONFIG, connectorClass);
+ Admin admin = mock(Admin.class);
worker = new Worker(WORKER_ID, new MockTime(), plugins, config, offsetBackingStore, executorService,
- allConnectorClientConfigOverridePolicy);
+ allConnectorClientConfigOverridePolicy, config -> admin);
worker.start();
-
- Admin admin = mock(Admin.class);
mockAdminListConsumerGroupOffsets(admin, Collections.singletonMap(new TopicPartition("test-topic", 0), new OffsetAndMetadata(10)), null);
FutureCallback cb = new FutureCallback<>();
- worker.sinkConnectorOffsets(CONNECTOR_ID, sinkConnector, connectorProps, cb, config -> admin);
+ worker.sinkConnectorOffsets(CONNECTOR_ID, sinkConnector, connectorProps, cb);
ConnectorOffsets offsets = cb.get(1000, TimeUnit.MILLISECONDS);
assertEquals(1, offsets.offsets().size());
@@ -1754,7 +1767,7 @@ public void testGetSinkConnectorOffsets() throws Exception {
assertEquals("test-topic", offsets.offsets().get(0).partition().get(SinkUtils.KAFKA_TOPIC_KEY));
verify(admin).listConsumerGroupOffsets(eq(SinkUtils.consumerGroupId(CONNECTOR_ID)), any(ListConsumerGroupOffsetsOptions.class));
- verify(admin).close();
+ verify(admin, timeout(1000)).close();
verifyKafkaClusterId();
}
@@ -1765,15 +1778,15 @@ public void testGetSinkConnectorOffsetsAdminClientSynchronousError() {
String connectorClass = SampleSinkConnector.class.getName();
connectorProps.put(CONNECTOR_CLASS_CONFIG, connectorClass);
+ Admin admin = mock(Admin.class);
worker = new Worker(WORKER_ID, new MockTime(), plugins, config, offsetBackingStore, executorService,
- allConnectorClientConfigOverridePolicy);
+ allConnectorClientConfigOverridePolicy, config -> admin);
worker.start();
- Admin admin = mock(Admin.class);
when(admin.listConsumerGroupOffsets(anyString(), any(ListConsumerGroupOffsetsOptions.class))).thenThrow(new ClusterAuthorizationException("Test exception"));
FutureCallback cb = new FutureCallback<>();
- worker.sinkConnectorOffsets(CONNECTOR_ID, sinkConnector, connectorProps, cb, config -> admin);
+ worker.sinkConnectorOffsets(CONNECTOR_ID, sinkConnector, connectorProps, cb);
ExecutionException e = assertThrows(ExecutionException.class, () -> cb.get(1000, TimeUnit.MILLISECONDS));
assertEquals(ConnectException.class, e.getCause().getClass());
@@ -1789,20 +1802,20 @@ public void testGetSinkConnectorOffsetsAdminClientAsynchronousError() {
String connectorClass = SampleSinkConnector.class.getName();
connectorProps.put(CONNECTOR_CLASS_CONFIG, connectorClass);
+ Admin admin = mock(Admin.class);
worker = new Worker(WORKER_ID, new MockTime(), plugins, config, offsetBackingStore, executorService,
- allConnectorClientConfigOverridePolicy);
+ allConnectorClientConfigOverridePolicy, config -> admin);
worker.start();
- Admin admin = mock(Admin.class);
mockAdminListConsumerGroupOffsets(admin, null, new ClusterAuthorizationException("Test exception"));
FutureCallback cb = new FutureCallback<>();
- worker.sinkConnectorOffsets(CONNECTOR_ID, sinkConnector, connectorProps, cb, config -> admin);
+ worker.sinkConnectorOffsets(CONNECTOR_ID, sinkConnector, connectorProps, cb);
ExecutionException e = assertThrows(ExecutionException.class, () -> cb.get(1000, TimeUnit.MILLISECONDS));
assertEquals(ConnectException.class, e.getCause().getClass());
verify(admin).listConsumerGroupOffsets(eq(SinkUtils.consumerGroupId(CONNECTOR_ID)), any(ListConsumerGroupOffsetsOptions.class));
- verify(admin).close();
+ verify(admin, timeout(1000)).close();
verifyKafkaClusterId();
}
@@ -1831,7 +1844,7 @@ public void testGetSourceConnectorOffsets() throws Exception {
return null;
});
worker = new Worker(WORKER_ID, new MockTime(), plugins, config, offsetBackingStore, executorService,
- allConnectorClientConfigOverridePolicy);
+ allConnectorClientConfigOverridePolicy, null);
worker.start();
Set> connectorPartitions =
@@ -1853,7 +1866,6 @@ public void testGetSourceConnectorOffsets() throws Exception {
assertEquals("partitionValue", offsets.offsets().get(0).partition().get("partitionKey"));
assertEquals("offsetValue", offsets.offsets().get(0).offset().get("offsetKey"));
- verify(offsetStore).configure(config);
verify(offsetStore).start();
verify(offsetReader).close();
verify(offsetStore).stop();
@@ -1872,7 +1884,7 @@ public void testGetSourceConnectorOffsetsError() {
return null;
});
worker = new Worker(WORKER_ID, new MockTime(), plugins, config, offsetBackingStore, executorService,
- allConnectorClientConfigOverridePolicy);
+ allConnectorClientConfigOverridePolicy, null);
worker.start();
when(offsetStore.connectorPartitions(CONNECTOR_ID)).thenThrow(new ConnectException("Test exception"));
@@ -1882,13 +1894,352 @@ public void testGetSourceConnectorOffsetsError() {
ExecutionException e = assertThrows(ExecutionException.class, () -> cb.get(1000, TimeUnit.MILLISECONDS));
assertEquals(ConnectException.class, e.getCause().getClass());
- verify(offsetStore).configure(config);
verify(offsetStore).start();
verify(offsetReader).close();
verify(offsetStore).stop();
verifyKafkaClusterId();
}
+ @Test
+ public void testAlterOffsetsConnectorDoesNotSupportOffsetAlteration() {
+ mockKafkaClusterId();
+
+ worker = new Worker(WORKER_ID, new MockTime(), plugins, config, offsetBackingStore, Executors.newSingleThreadExecutor(),
+ allConnectorClientConfigOverridePolicy, null);
+ worker.start();
+
+ mockGenericIsolation();
+ when(plugins.newConnector(anyString())).thenReturn(sourceConnector);
+ when(plugins.withClassLoader(any(ClassLoader.class), any(Runnable.class))).thenAnswer(AdditionalAnswers.returnsSecondArg());
+ when(sourceConnector.alterOffsets(eq(connectorProps), anyMap())).thenThrow(new UnsupportedOperationException("This connector doesn't " +
+ "support altering of offsets"));
+
+ FutureCallback cb = new FutureCallback<>();
+ worker.alterConnectorOffsets(CONNECTOR_ID, connectorProps,
+ Collections.singletonMap(Collections.singletonMap("partitionKey", "partitionValue"), Collections.singletonMap("offsetKey", "offsetValue")),
+ cb);
+
+ ExecutionException e = assertThrows(ExecutionException.class, () -> cb.get(1000, TimeUnit.MILLISECONDS));
+ assertEquals(ConnectException.class, e.getCause().getClass());
+ assertEquals("Failed to alter offsets for connector " + CONNECTOR_ID + " because it doesn't support external modification of offsets",
+ e.getCause().getMessage());
+
+ verifyGenericIsolation();
+ verifyKafkaClusterId();
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testAlterOffsetsSourceConnector() throws Exception {
+ mockKafkaClusterId();
+ worker = new Worker(WORKER_ID, new MockTime(), plugins, config, offsetBackingStore, Executors.newSingleThreadExecutor(),
+ allConnectorClientConfigOverridePolicy, null);
+ worker.start();
+
+ when(plugins.withClassLoader(any(ClassLoader.class), any(Runnable.class))).thenAnswer(AdditionalAnswers.returnsSecondArg());
+ when(sourceConnector.alterOffsets(eq(connectorProps), anyMap())).thenReturn(true);
+ ConnectorOffsetBackingStore offsetStore = mock(ConnectorOffsetBackingStore.class);
+ KafkaProducer producer = mock(KafkaProducer.class);
+ OffsetStorageWriter offsetWriter = mock(OffsetStorageWriter.class);
+
+ Map, Map> partitionOffsets = new HashMap<>();
+ partitionOffsets.put(Collections.singletonMap("partitionKey", "partitionValue"), Collections.singletonMap("offsetKey", "offsetValue"));
+ partitionOffsets.put(Collections.singletonMap("partitionKey", "partitionValue2"), Collections.singletonMap("offsetKey", "offsetValue"));
+
+ when(offsetWriter.doFlush(any())).thenAnswer(invocation -> {
+ invocation.getArgument(0, Callback.class).onCompletion(null, null);
+ return null;
+ });
+
+ FutureCallback cb = new FutureCallback<>();
+ worker.alterSourceConnectorOffsets(CONNECTOR_ID, sourceConnector, connectorProps, partitionOffsets, offsetStore, producer,
+ offsetWriter, Thread.currentThread().getContextClassLoader(), cb);
+ assertEquals("The offsets for this connector have been altered successfully", cb.get(1000, TimeUnit.MILLISECONDS).message());
+
+ verify(offsetStore).start();
+ partitionOffsets.forEach((partition, offset) -> verify(offsetWriter).offset(partition, offset));
+ verify(offsetWriter).beginFlush();
+ verify(offsetWriter).doFlush(any());
+ verify(offsetStore, timeout(1000)).stop();
+ verifyKafkaClusterId();
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testAlterOffsetsSourceConnectorError() throws Exception {
+ mockKafkaClusterId();
+ worker = new Worker(WORKER_ID, new MockTime(), plugins, config, offsetBackingStore, Executors.newSingleThreadExecutor(),
+ allConnectorClientConfigOverridePolicy, null);
+ worker.start();
+
+ when(plugins.withClassLoader(any(ClassLoader.class), any(Runnable.class))).thenAnswer(AdditionalAnswers.returnsSecondArg());
+ when(sourceConnector.alterOffsets(eq(connectorProps), anyMap())).thenReturn(true);
+ ConnectorOffsetBackingStore offsetStore = mock(ConnectorOffsetBackingStore.class);
+ KafkaProducer producer = mock(KafkaProducer.class);
+ OffsetStorageWriter offsetWriter = mock(OffsetStorageWriter.class);
+
+ Map, Map> partitionOffsets = new HashMap<>();
+ partitionOffsets.put(Collections.singletonMap("partitionKey", "partitionValue"), Collections.singletonMap("offsetKey", "offsetValue"));
+ partitionOffsets.put(Collections.singletonMap("partitionKey", "partitionValue2"), Collections.singletonMap("offsetKey", "offsetValue"));
+
+ when(offsetWriter.doFlush(any())).thenAnswer(invocation -> {
+ invocation.getArgument(0, Callback.class).onCompletion(new RuntimeException("Test exception"), null);
+ return null;
+ });
+
+ FutureCallback cb = new FutureCallback<>();
+ worker.alterSourceConnectorOffsets(CONNECTOR_ID, sourceConnector, connectorProps, partitionOffsets, offsetStore, producer,
+ offsetWriter, Thread.currentThread().getContextClassLoader(), cb);
+ ExecutionException e = assertThrows(ExecutionException.class, () -> cb.get(1000, TimeUnit.MILLISECONDS).message());
+ assertEquals(ConnectException.class, e.getCause().getClass());
+
+ verify(offsetStore).start();
+ partitionOffsets.forEach((partition, offset) -> verify(offsetWriter).offset(partition, offset));
+ verify(offsetWriter).beginFlush();
+ verify(offsetWriter).doFlush(any());
+ verify(offsetStore, timeout(1000)).stop();
+ verifyKafkaClusterId();
+ }
+
+ @Test
+ public void testAlterOffsetsSinkConnectorNoResets() throws Exception {
+ @SuppressWarnings("unchecked")
+ ArgumentCaptor> alterOffsetsMapCapture = ArgumentCaptor.forClass(Map.class);
+ Map, Map> partitionOffsets = new HashMap<>();
+ Map partition1 = new HashMap<>();
+ partition1.put(SinkUtils.KAFKA_TOPIC_KEY, "test_topic");
+ partition1.put(SinkUtils.KAFKA_PARTITION_KEY, "10");
+ partitionOffsets.put(partition1, Collections.singletonMap(SinkUtils.KAFKA_OFFSET_KEY, 500));
+ Map partition2 = new HashMap<>();
+ partition2.put(SinkUtils.KAFKA_TOPIC_KEY, "test_topic");
+ partition2.put(SinkUtils.KAFKA_PARTITION_KEY, "20");
+ partitionOffsets.put(partition2, Collections.singletonMap(SinkUtils.KAFKA_OFFSET_KEY, 100));
+
+ // A null value for deleteOffsetsSetCapture indicates that we don't expect any call to Admin::deleteConsumerGroupOffsets
+ alterOffsetsSinkConnector(partitionOffsets, alterOffsetsMapCapture, null);
+
+ assertEquals(2, alterOffsetsMapCapture.getValue().size());
+ assertEquals(500, alterOffsetsMapCapture.getValue().get(new TopicPartition("test_topic", 10)).offset());
+ assertEquals(100, alterOffsetsMapCapture.getValue().get(new TopicPartition("test_topic", 20)).offset());
+ }
+
+ @Test
+ public void testAlterOffsetSinkConnectorOnlyResets() throws Exception {
+ @SuppressWarnings("unchecked")
+ ArgumentCaptor> deleteOffsetsSetCapture = ArgumentCaptor.forClass(Set.class);
+ Map, Map> partitionOffsets = new HashMap<>();
+ Map partition1 = new HashMap<>();
+ partition1.put(SinkUtils.KAFKA_TOPIC_KEY, "test_topic");
+ partition1.put(SinkUtils.KAFKA_PARTITION_KEY, "10");
+ partitionOffsets.put(partition1, null);
+ Map partition2 = new HashMap<>();
+ partition2.put(SinkUtils.KAFKA_TOPIC_KEY, "test_topic");
+ partition2.put(SinkUtils.KAFKA_PARTITION_KEY, "20");
+ partitionOffsets.put(partition2, null);
+
+ // A null value for alterOffsetsMapCapture indicates that we don't expect any call to Admin::alterConsumerGroupOffsets
+ alterOffsetsSinkConnector(partitionOffsets, null, deleteOffsetsSetCapture);
+
+ Set expectedTopicPartitionsForOffsetDelete = new HashSet<>();
+ expectedTopicPartitionsForOffsetDelete.add(new TopicPartition("test_topic", 10));
+ expectedTopicPartitionsForOffsetDelete.add(new TopicPartition("test_topic", 20));
+
+ // Verify that contents are equal without caring about order
+ assertEquals(expectedTopicPartitionsForOffsetDelete, deleteOffsetsSetCapture.getValue());
+ }
+
+ @Test
+ public void testAlterOffsetsSinkConnectorAltersAndResets() throws Exception {
+ @SuppressWarnings("unchecked")
+ ArgumentCaptor> alterOffsetsMapCapture = ArgumentCaptor.forClass(Map.class);
+ @SuppressWarnings("unchecked")
+ ArgumentCaptor> deleteOffsetsSetCapture = ArgumentCaptor.forClass(Set.class);
+ Map, Map> partitionOffsets = new HashMap<>();
+ Map partition1 = new HashMap<>();
+ partition1.put(SinkUtils.KAFKA_TOPIC_KEY, "test_topic");
+ partition1.put(SinkUtils.KAFKA_PARTITION_KEY, "10");
+ partitionOffsets.put(partition1, Collections.singletonMap(SinkUtils.KAFKA_OFFSET_KEY, "100"));
+ Map partition2 = new HashMap<>();
+ partition2.put(SinkUtils.KAFKA_TOPIC_KEY, "test_topic");
+ partition2.put(SinkUtils.KAFKA_PARTITION_KEY, "20");
+ partitionOffsets.put(partition2, null);
+
+ alterOffsetsSinkConnector(partitionOffsets, alterOffsetsMapCapture, deleteOffsetsSetCapture);
+
+ assertEquals(1, alterOffsetsMapCapture.getValue().size());
+ assertEquals(100, alterOffsetsMapCapture.getValue().get(new TopicPartition("test_topic", 10)).offset());
+
+ Set expectedTopicPartitionsForOffsetDelete = Collections.singleton(new TopicPartition("test_topic", 20));
+ assertEquals(expectedTopicPartitionsForOffsetDelete, deleteOffsetsSetCapture.getValue());
+ }
+
+ private void alterOffsetsSinkConnector(Map, Map> partitionOffsets,
+ ArgumentCaptor> alterOffsetsMapCapture,
+ ArgumentCaptor> deleteOffsetsSetCapture) throws Exception {
+ mockKafkaClusterId();
+ String connectorClass = SampleSinkConnector.class.getName();
+ connectorProps.put(CONNECTOR_CLASS_CONFIG, connectorClass);
+
+ Admin admin = mock(Admin.class);
+ worker = new Worker(WORKER_ID, new MockTime(), plugins, config, offsetBackingStore, Executors.newCachedThreadPool(),
+ allConnectorClientConfigOverridePolicy, config -> admin);
+ worker.start();
+
+ when(plugins.withClassLoader(any(ClassLoader.class), any(Runnable.class))).thenAnswer(AdditionalAnswers.returnsSecondArg());
+ when(sinkConnector.alterOffsets(eq(connectorProps), anyMap())).thenReturn(true);
+
+ // If alterOffsetsMapCapture is null, then we won't stub any of the following methods resulting in test failures in case
+ // offsets for certain topic partitions are actually attempted to be altered.
+ if (alterOffsetsMapCapture != null) {
+ AlterConsumerGroupOffsetsResult alterConsumerGroupOffsetsResult = mock(AlterConsumerGroupOffsetsResult.class);
+ when(admin.alterConsumerGroupOffsets(anyString(), alterOffsetsMapCapture.capture(), any(AlterConsumerGroupOffsetsOptions.class)))
+ .thenReturn(alterConsumerGroupOffsetsResult);
+ KafkaFuture alterFuture = KafkaFuture.completedFuture(null);
+ when(alterConsumerGroupOffsetsResult.all()).thenReturn(alterFuture);
+ }
+
+ // If deleteOffsetsSetCapture is null, then we won't stub any of the following methods resulting in test failures in case
+ // offsets for certain topic partitions are actually attempted to be deleted.
+ if (deleteOffsetsSetCapture != null) {
+ DeleteConsumerGroupOffsetsResult deleteConsumerGroupOffsetsResult = mock(DeleteConsumerGroupOffsetsResult.class);
+ when(admin.deleteConsumerGroupOffsets(anyString(), deleteOffsetsSetCapture.capture(), any(DeleteConsumerGroupOffsetsOptions.class)))
+ .thenReturn(deleteConsumerGroupOffsetsResult);
+ KafkaFuture deleteFuture = KafkaFuture.completedFuture(null);
+ when(deleteConsumerGroupOffsetsResult.all()).thenReturn(deleteFuture);
+ }
+
+ FutureCallback cb = new FutureCallback<>();
+ worker.alterSinkConnectorOffsets(CONNECTOR_ID, sinkConnector, connectorProps, partitionOffsets,
+ Thread.currentThread().getContextClassLoader(), cb);
+ assertEquals("The offsets for this connector have been altered successfully", cb.get(1000, TimeUnit.MILLISECONDS).message());
+
+ verify(admin, timeout(1000)).close();
+ verifyKafkaClusterId();
+ }
+
+ @Test
+ public void testAlterOffsetsSinkConnectorAlterOffsetsError() throws Exception {
+ mockKafkaClusterId();
+ String connectorClass = SampleSinkConnector.class.getName();
+ connectorProps.put(CONNECTOR_CLASS_CONFIG, connectorClass);
+
+ Admin admin = mock(Admin.class);
+ worker = new Worker(WORKER_ID, new MockTime(), plugins, config, offsetBackingStore, Executors.newSingleThreadExecutor(),
+ allConnectorClientConfigOverridePolicy, config -> admin);
+ worker.start();
+
+ when(plugins.withClassLoader(any(ClassLoader.class), any(Runnable.class))).thenAnswer(AdditionalAnswers.returnsSecondArg());
+ when(sinkConnector.alterOffsets(eq(connectorProps), anyMap())).thenReturn(true);
+
+ AlterConsumerGroupOffsetsResult alterConsumerGroupOffsetsResult = mock(AlterConsumerGroupOffsetsResult.class);
+ when(admin.alterConsumerGroupOffsets(anyString(), anyMap(), any(AlterConsumerGroupOffsetsOptions.class)))
+ .thenReturn(alterConsumerGroupOffsetsResult);
+ KafkaFutureImpl alterFuture = new KafkaFutureImpl<>();
+ alterFuture.completeExceptionally(new ClusterAuthorizationException("Test exception"));
+ when(alterConsumerGroupOffsetsResult.all()).thenReturn(alterFuture);
+
+ Map partition1 = new HashMap<>();
+ partition1.put(SinkUtils.KAFKA_TOPIC_KEY, "test_topic");
+ partition1.put(SinkUtils.KAFKA_PARTITION_KEY, "10");
+ Map, Map> partitionOffsets = Collections.singletonMap(partition1,
+ Collections.singletonMap(SinkUtils.KAFKA_OFFSET_KEY, "100"));
+
+ FutureCallback cb = new FutureCallback<>();
+ worker.alterSinkConnectorOffsets(CONNECTOR_ID, sinkConnector, connectorProps, partitionOffsets,
+ Thread.currentThread().getContextClassLoader(), cb);
+
+ ExecutionException e = assertThrows(ExecutionException.class, () -> cb.get(1000, TimeUnit.MILLISECONDS));
+ assertEquals(ConnectException.class, e.getCause().getClass());
+
+ verify(admin, timeout(1000)).close();
+ verifyNoMoreInteractions(admin);
+ verifyKafkaClusterId();
+ }
+
+ @Test
+ public void testAlterOffsetsSinkConnectorDeleteOffsetsError() throws Exception {
+ mockKafkaClusterId();
+ String connectorClass = SampleSinkConnector.class.getName();
+ connectorProps.put(CONNECTOR_CLASS_CONFIG, connectorClass);
+
+ Admin admin = mock(Admin.class);
+ worker = new Worker(WORKER_ID, new MockTime(), plugins, config, offsetBackingStore, Executors.newSingleThreadExecutor(),
+ allConnectorClientConfigOverridePolicy, config -> admin);
+ worker.start();
+
+ when(plugins.withClassLoader(any(ClassLoader.class), any(Runnable.class))).thenAnswer(AdditionalAnswers.returnsSecondArg());
+ when(sinkConnector.alterOffsets(eq(connectorProps), anyMap())).thenReturn(true);
+
+ AlterConsumerGroupOffsetsResult alterConsumerGroupOffsetsResult = mock(AlterConsumerGroupOffsetsResult.class);
+ when(admin.alterConsumerGroupOffsets(anyString(), anyMap(), any(AlterConsumerGroupOffsetsOptions.class)))
+ .thenReturn(alterConsumerGroupOffsetsResult);
+ KafkaFuture alterFuture = KafkaFuture.completedFuture(null);
+ when(alterConsumerGroupOffsetsResult.all()).thenReturn(alterFuture);
+
+ DeleteConsumerGroupOffsetsResult deleteConsumerGroupOffsetsResult = mock(DeleteConsumerGroupOffsetsResult.class);
+ when(admin.deleteConsumerGroupOffsets(anyString(), anySet(), any(DeleteConsumerGroupOffsetsOptions.class)))
+ .thenReturn(deleteConsumerGroupOffsetsResult);
+ KafkaFutureImpl deleteFuture = new KafkaFutureImpl<>();
+ deleteFuture.completeExceptionally(new ClusterAuthorizationException("Test exception"));
+ when(deleteConsumerGroupOffsetsResult.all()).thenReturn(deleteFuture);
+
+ Map, Map> partitionOffsets = new HashMap<>();
+ Map partition1 = new HashMap<>();
+ partition1.put(SinkUtils.KAFKA_TOPIC_KEY, "test_topic");
+ partition1.put(SinkUtils.KAFKA_PARTITION_KEY, "10");
+ partitionOffsets.put(partition1, Collections.singletonMap(SinkUtils.KAFKA_OFFSET_KEY, "100"));
+ Map partition2 = new HashMap<>();
+ partition2.put(SinkUtils.KAFKA_TOPIC_KEY, "test_topic");
+ partition2.put(SinkUtils.KAFKA_PARTITION_KEY, "20");
+ partitionOffsets.put(partition2, null);
+
+ FutureCallback cb = new FutureCallback<>();
+ worker.alterSinkConnectorOffsets(CONNECTOR_ID, sinkConnector, connectorProps, partitionOffsets,
+ Thread.currentThread().getContextClassLoader(), cb);
+
+ ExecutionException e = assertThrows(ExecutionException.class, () -> cb.get(1000, TimeUnit.MILLISECONDS));
+ assertEquals(ConnectException.class, e.getCause().getClass());
+
+ verify(admin, timeout(1000)).close();
+ verifyNoMoreInteractions(admin);
+ verifyKafkaClusterId();
+ }
+
+ @Test
+ public void testAlterOffsetsSinkConnectorSynchronousError() throws Exception {
+ mockKafkaClusterId();
+ String connectorClass = SampleSinkConnector.class.getName();
+ connectorProps.put(CONNECTOR_CLASS_CONFIG, connectorClass);
+
+ Admin admin = mock(Admin.class);
+ worker = new Worker(WORKER_ID, new MockTime(), plugins, config, offsetBackingStore, Executors.newSingleThreadExecutor(),
+ allConnectorClientConfigOverridePolicy, config -> admin);
+ worker.start();
+
+ when(plugins.withClassLoader(any(ClassLoader.class), any(Runnable.class))).thenAnswer(AdditionalAnswers.returnsSecondArg());
+ when(sinkConnector.alterOffsets(eq(connectorProps), anyMap())).thenReturn(true);
+
+ when(admin.alterConsumerGroupOffsets(anyString(), anyMap(), any(AlterConsumerGroupOffsetsOptions.class)))
+ .thenThrow(new RuntimeException("Test Exception"));
+
+ Map, Map> partitionOffsets = new HashMap<>();
+ Map partition1 = new HashMap<>();
+ partition1.put(SinkUtils.KAFKA_TOPIC_KEY, "test_topic");
+ partition1.put(SinkUtils.KAFKA_PARTITION_KEY, "10");
+ partitionOffsets.put(partition1, Collections.singletonMap(SinkUtils.KAFKA_OFFSET_KEY, "100"));
+
+ FutureCallback cb = new FutureCallback<>();
+ worker.alterSinkConnectorOffsets(CONNECTOR_ID, sinkConnector, connectorProps, partitionOffsets,
+ Thread.currentThread().getContextClassLoader(), cb);
+
+ ExecutionException e = assertThrows(ExecutionException.class, () -> cb.get(1000, TimeUnit.MILLISECONDS));
+ assertEquals(ConnectException.class, e.getCause().getClass());
+
+ verify(admin, timeout(1000)).close();
+ verifyNoMoreInteractions(admin);
+ verifyKafkaClusterId();
+ }
+
private void assertStatusMetrics(long expected, String metricName) {
MetricGroup statusMetrics = worker.connectorStatusMetricsGroup().metricGroup(TASK_ID.connector());
if (expected == 0L) {
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 33b8f966e2ed..d5d1bbab66eb 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
@@ -52,6 +52,7 @@
import org.apache.kafka.connect.runtime.rest.entities.ConnectorOffsets;
import org.apache.kafka.connect.runtime.rest.entities.ConnectorStateInfo;
import org.apache.kafka.connect.runtime.rest.entities.ConnectorType;
+import org.apache.kafka.connect.runtime.rest.entities.Message;
import org.apache.kafka.connect.runtime.rest.entities.TaskInfo;
import org.apache.kafka.connect.runtime.rest.errors.BadRequestException;
import org.apache.kafka.connect.runtime.rest.errors.ConnectRestException;
@@ -220,12 +221,24 @@ public class DistributedHerderTest {
private static final ClusterConfigState SNAPSHOT_STOPPED_CONN1 = new ClusterConfigState(
1,
null,
- Collections.singletonMap(CONN1, 3),
+ Collections.singletonMap(CONN1, 0),
Collections.singletonMap(CONN1, CONN1_CONFIG),
Collections.singletonMap(CONN1, TargetState.STOPPED),
Collections.emptyMap(), // Stopped connectors should have an empty set of task configs
+ Collections.singletonMap(CONN1, 3),
+ Collections.singletonMap(CONN1, 10),
+ Collections.singleton(CONN1),
+ Collections.emptySet());
+
+ private static final ClusterConfigState SNAPSHOT_STOPPED_CONN1_FENCED = new ClusterConfigState(
+ 1,
+ null,
+ Collections.singletonMap(CONN1, 0),
+ Collections.singletonMap(CONN1, CONN1_CONFIG),
+ Collections.singletonMap(CONN1, TargetState.STOPPED),
Collections.emptyMap(),
- Collections.emptyMap(),
+ Collections.singletonMap(CONN1, 0),
+ Collections.singletonMap(CONN1, 11),
Collections.emptySet(),
Collections.emptySet());
private static final ClusterConfigState SNAPSHOT_UPDATED_CONN1_CONFIG = new ClusterConfigState(
@@ -4123,6 +4136,325 @@ public void testConnectorOffsets() throws Exception {
PowerMock.verifyAll();
}
+ @Test
+ public void testAlterConnectorOffsetsUnknownConnector() throws Exception {
+ // Get the initial assignment
+ EasyMock.expect(member.memberId()).andStubReturn("leader");
+ EasyMock.expect(member.currentProtocolVersion()).andStubReturn(CONNECT_PROTOCOL_V0);
+ expectRebalance(1, Collections.emptyList(), Collections.emptyList(), true);
+ expectConfigRefreshAndSnapshot(SNAPSHOT);
+ member.poll(EasyMock.anyInt());
+ PowerMock.expectLastCall();
+
+ // Now handle the alter connector offsets request
+ member.wakeup();
+ PowerMock.expectLastCall();
+ member.ensureActive();
+ PowerMock.expectLastCall();
+ expectConfigRefreshAndSnapshot(SNAPSHOT);
+ member.poll(EasyMock.anyInt());
+ PowerMock.expectLastCall();
+
+ PowerMock.replayAll();
+
+ herder.tick();
+ FutureCallback callback = new FutureCallback<>();
+ herder.alterConnectorOffsets("connector-does-not-exist", new HashMap<>(), callback);
+ herder.tick();
+ ExecutionException e = assertThrows(ExecutionException.class, () -> callback.get(1000L, TimeUnit.MILLISECONDS));
+ assertTrue(e.getCause() instanceof NotFoundException);
+
+ PowerMock.verifyAll();
+ }
+
+ @Test
+ public void testAlterOffsetsConnectorNotInStoppedState() throws Exception {
+ // Get the initial assignment
+ EasyMock.expect(member.memberId()).andStubReturn("leader");
+ EasyMock.expect(member.currentProtocolVersion()).andStubReturn(CONNECT_PROTOCOL_V0);
+ expectRebalance(1, Collections.emptyList(), Collections.emptyList(), true);
+ expectConfigRefreshAndSnapshot(SNAPSHOT);
+ member.poll(EasyMock.anyInt());
+ PowerMock.expectLastCall();
+
+ // Now handle the alter connector offsets request
+ member.wakeup();
+ PowerMock.expectLastCall();
+ member.ensureActive();
+ PowerMock.expectLastCall();
+ expectConfigRefreshAndSnapshot(SNAPSHOT);
+ member.poll(EasyMock.anyInt());
+ PowerMock.expectLastCall();
+
+ PowerMock.replayAll();
+
+ herder.tick();
+ FutureCallback callback = new FutureCallback<>();
+ herder.alterConnectorOffsets(CONN1, new HashMap<>(), callback);
+ herder.tick();
+ ExecutionException e = assertThrows(ExecutionException.class, () -> callback.get(1000L, TimeUnit.MILLISECONDS));
+ assertTrue(e.getCause() instanceof BadRequestException);
+
+ PowerMock.verifyAll();
+ }
+
+ @Test
+ public void testAlterOffsetsNotLeader() throws Exception {
+ // Get the initial assignment
+ EasyMock.expect(member.memberId()).andStubReturn("member");
+ EasyMock.expect(member.currentProtocolVersion()).andStubReturn(CONNECT_PROTOCOL_V0);
+ expectRebalance(1, Collections.emptyList(), Collections.emptyList(), false);
+ expectConfigRefreshAndSnapshot(SNAPSHOT_STOPPED_CONN1);
+ member.poll(EasyMock.anyInt());
+ PowerMock.expectLastCall();
+
+ // Now handle the alter connector offsets request
+ member.wakeup();
+ PowerMock.expectLastCall();
+ member.ensureActive();
+ PowerMock.expectLastCall();
+ member.poll(EasyMock.anyInt());
+ PowerMock.expectLastCall();
+
+ PowerMock.replayAll();
+
+ herder.tick();
+ FutureCallback callback = new FutureCallback<>();
+ herder.alterConnectorOffsets(CONN1, new HashMap<>(), callback);
+ herder.tick();
+ ExecutionException e = assertThrows(ExecutionException.class, () -> callback.get(1000L, TimeUnit.MILLISECONDS));
+ assertTrue(e.getCause() instanceof NotLeaderException);
+
+ PowerMock.verifyAll();
+ }
+
+ @Test
+ public void testAlterOffsetsSinkConnector() throws Exception {
+ EasyMock.reset(herder);
+ EasyMock.expect(herder.connectorType(EasyMock.anyObject())).andReturn(ConnectorType.SINK).anyTimes();
+ PowerMock.expectPrivate(herder, "updateDeletedConnectorStatus").andVoid().anyTimes();
+ PowerMock.expectPrivate(herder, "updateDeletedTaskStatus").andVoid().anyTimes();
+
+ // Get the initial assignment
+ EasyMock.expect(member.memberId()).andStubReturn("leader");
+ EasyMock.expect(member.currentProtocolVersion()).andStubReturn(CONNECT_PROTOCOL_V0);
+ expectRebalance(1, Collections.emptyList(), Collections.emptyList(), true);
+ expectConfigRefreshAndSnapshot(SNAPSHOT_STOPPED_CONN1);
+ member.poll(EasyMock.anyInt());
+ PowerMock.expectLastCall();
+
+ // Now handle the alter connector offsets request
+ Map, Map> offsets = new HashMap<>();
+ member.wakeup();
+ PowerMock.expectLastCall();
+ member.ensureActive();
+ PowerMock.expectLastCall();
+ member.poll(EasyMock.anyInt());
+ PowerMock.expectLastCall();
+ expectConfigRefreshAndSnapshot(SNAPSHOT_STOPPED_CONN1);
+ Capture> workerCallbackCapture = Capture.newInstance();
+ worker.alterConnectorOffsets(EasyMock.eq(CONN1), EasyMock.eq(CONN1_CONFIG), EasyMock.eq(offsets), capture(workerCallbackCapture));
+ Message msg = new Message("The offsets for this connector have been altered successfully");
+ EasyMock.expectLastCall().andAnswer(() -> {
+ workerCallbackCapture.getValue().onCompletion(null, msg);
+ return null;
+ });
+
+ PowerMock.replayAll();
+
+ herder.tick();
+ FutureCallback callback = new FutureCallback<>();
+ herder.alterConnectorOffsets(CONN1, offsets, callback);
+ herder.tick();
+ assertEquals(msg, callback.get(1000L, TimeUnit.MILLISECONDS));
+ assertEquals("The offsets for this connector have been altered successfully", msg.message());
+
+ PowerMock.verifyAll();
+ }
+
+ @Test
+ public void testAlterOffsetsSourceConnectorExactlyOnceDisabled() throws Exception {
+ // Get the initial assignment
+ EasyMock.expect(member.memberId()).andStubReturn("leader");
+ EasyMock.expect(member.currentProtocolVersion()).andStubReturn(CONNECT_PROTOCOL_V0);
+ expectRebalance(1, Collections.emptyList(), Collections.emptyList(), true);
+ expectConfigRefreshAndSnapshot(SNAPSHOT_STOPPED_CONN1);
+ member.poll(EasyMock.anyInt());
+ PowerMock.expectLastCall();
+
+ // Now handle the alter connector offsets request
+ Map, Map> offsets = new HashMap<>();
+ member.wakeup();
+ PowerMock.expectLastCall();
+ member.ensureActive();
+ PowerMock.expectLastCall();
+ member.poll(EasyMock.anyInt());
+ PowerMock.expectLastCall();
+ expectConfigRefreshAndSnapshot(SNAPSHOT_STOPPED_CONN1);
+ Capture> workerCallbackCapture = Capture.newInstance();
+ worker.alterConnectorOffsets(EasyMock.eq(CONN1), EasyMock.eq(CONN1_CONFIG), EasyMock.eq(offsets), capture(workerCallbackCapture));
+ Message msg = new Message("The offsets for this connector have been altered successfully");
+ EasyMock.expectLastCall().andAnswer(() -> {
+ workerCallbackCapture.getValue().onCompletion(null, msg);
+ return null;
+ });
+
+ PowerMock.replayAll();
+
+ herder.tick();
+ FutureCallback callback = new FutureCallback<>();
+ herder.alterConnectorOffsets(CONN1, offsets, callback);
+ herder.tick();
+ assertEquals(msg, callback.get(1000L, TimeUnit.MILLISECONDS));
+ assertEquals("The offsets for this connector have been altered successfully", msg.message());
+
+ PowerMock.verifyAll();
+ }
+
+ @Test
+ public void testAlterOffsetsSourceConnectorExactlyOnceEnabled() throws Exception {
+ // Setup herder with exactly-once support for source connectors enabled
+ herder = exactlyOnceHerder();
+ rebalanceListener = herder.new RebalanceListener(time);
+ PowerMock.expectPrivate(herder, "updateDeletedConnectorStatus").andVoid().anyTimes();
+ PowerMock.expectPrivate(herder, "updateDeletedTaskStatus").andVoid().anyTimes();
+
+ // Get the initial assignment
+ EasyMock.expect(member.memberId()).andStubReturn("leader");
+ EasyMock.expect(member.currentProtocolVersion()).andStubReturn(CONNECT_PROTOCOL_V0);
+ expectRebalance(1, Collections.emptyList(), Collections.emptyList(), true);
+ expectConfigRefreshAndSnapshot(SNAPSHOT_STOPPED_CONN1);
+ member.poll(EasyMock.anyInt());
+ PowerMock.expectLastCall().anyTimes();
+
+ // Now handle the alter connector offsets request
+ Map, Map> offsets = new HashMap<>();
+ member.wakeup();
+ PowerMock.expectLastCall().anyTimes();
+ member.ensureActive();
+ PowerMock.expectLastCall().anyTimes();
+ expectConfigRefreshAndSnapshot(SNAPSHOT_STOPPED_CONN1);
+ expectConfigRefreshAndSnapshot(SNAPSHOT_STOPPED_CONN1);
+ EasyMock.expect(herder.connectorType(EasyMock.anyObject())).andReturn(ConnectorType.SOURCE).anyTimes();
+
+ // Expect a round of zombie fencing to occur
+ expectConfigRefreshAndSnapshot(SNAPSHOT_STOPPED_CONN1);
+ KafkaFuture workerFencingFuture = EasyMock.mock(KafkaFuture.class);
+ KafkaFuture herderFencingFuture = EasyMock.mock(KafkaFuture.class);
+ EasyMock.expect(worker.fenceZombies(CONN1, SNAPSHOT_STOPPED_CONN1.taskCountRecord(CONN1), CONN1_CONFIG)).andReturn(workerFencingFuture);
+ EasyMock.expect(workerFencingFuture.thenApply(EasyMock.>anyObject())).andReturn(herderFencingFuture);
+
+ // Two fencing callbacks are added - one is in ZombieFencing::start itself to remove the connector from the active
+ // fencing list. The other is the callback passed from DistributedHerder::alterConnectorOffsets in order to
+ // queue up the actual alter offsets request if the zombie fencing succeeds.
+ for (int i = 0; i < 2; i++) {
+ Capture> herderFencingCallback = EasyMock.newCapture();
+ EasyMock.expect(herderFencingFuture.whenComplete(EasyMock.capture(herderFencingCallback))).andAnswer(() -> {
+ herderFencingCallback.getValue().accept(null, null);
+ return null;
+ });
+ }
+
+ Capture> workerCallbackCapture = Capture.newInstance();
+ Message msg = new Message("The offsets for this connector have been altered successfully");
+ worker.alterConnectorOffsets(EasyMock.eq(CONN1), EasyMock.eq(CONN1_CONFIG), EasyMock.eq(offsets), capture(workerCallbackCapture));
+ EasyMock.expectLastCall().andAnswer(() -> {
+ workerCallbackCapture.getValue().onCompletion(null, msg);
+ return null;
+ });
+
+ // Handle the second alter connector offsets request. No zombie fencing request to the worker is expected now since we
+ // already did a round of zombie fencing last time and no new tasks came up in the meanwhile. The config snapshot is
+ // refreshed once at the beginning of the DistributedHerder::alterConnectorOffsets method, once before checking
+ // whether zombie fencing is required, and once before actually proceeding to alter connector offsets.
+ expectConfigRefreshAndSnapshot(SNAPSHOT_STOPPED_CONN1_FENCED);
+ expectConfigRefreshAndSnapshot(SNAPSHOT_STOPPED_CONN1_FENCED);
+ expectConfigRefreshAndSnapshot(SNAPSHOT_STOPPED_CONN1_FENCED);
+ Capture> workerCallbackCapture2 = Capture.newInstance();
+ worker.alterConnectorOffsets(EasyMock.eq(CONN1), EasyMock.eq(CONN1_CONFIG), EasyMock.eq(offsets), capture(workerCallbackCapture2));
+ EasyMock.expectLastCall().andAnswer(() -> {
+ workerCallbackCapture2.getValue().onCompletion(null, msg);
+ return null;
+ });
+
+ PowerMock.replayAll(workerFencingFuture, herderFencingFuture);
+
+ herder.tick();
+ FutureCallback callback = new FutureCallback<>();
+ herder.alterConnectorOffsets(CONN1, offsets, callback);
+ // Process the zombie fencing request
+ herder.tick();
+ // Process the alter offsets request
+ herder.tick();
+ assertEquals(msg, callback.get(1000L, TimeUnit.MILLISECONDS));
+
+ FutureCallback callback2 = new FutureCallback<>();
+ herder.alterConnectorOffsets(CONN1, offsets, callback2);
+ herder.tick();
+ assertEquals(msg, callback.get(1000L, TimeUnit.MILLISECONDS));
+
+ PowerMock.verifyAll();
+ }
+
+ @Test
+ public void testAlterOffsetsSourceConnectorExactlyOnceEnabledZombieFencingFailure() throws Exception {
+ // Setup herder with exactly-once support for source connectors enabled
+ herder = exactlyOnceHerder();
+ rebalanceListener = herder.new RebalanceListener(time);
+ PowerMock.expectPrivate(herder, "updateDeletedConnectorStatus").andVoid().anyTimes();
+ PowerMock.expectPrivate(herder, "updateDeletedTaskStatus").andVoid().anyTimes();
+
+ // Get the initial assignment
+ EasyMock.expect(member.memberId()).andStubReturn("leader");
+ EasyMock.expect(member.currentProtocolVersion()).andStubReturn(CONNECT_PROTOCOL_V0);
+ expectRebalance(1, Collections.emptyList(), Collections.emptyList(), true);
+ expectConfigRefreshAndSnapshot(SNAPSHOT_STOPPED_CONN1);
+ member.poll(EasyMock.anyInt());
+ PowerMock.expectLastCall().anyTimes();
+
+ // Now handle the alter connector offsets request
+ member.wakeup();
+ PowerMock.expectLastCall().anyTimes();
+ member.ensureActive();
+ PowerMock.expectLastCall().anyTimes();
+ expectConfigRefreshAndSnapshot(SNAPSHOT_STOPPED_CONN1);
+ EasyMock.expect(herder.connectorType(EasyMock.anyObject())).andReturn(ConnectorType.SOURCE).anyTimes();
+
+ // Expect a round of zombie fencing to occur
+ expectConfigRefreshAndSnapshot(SNAPSHOT_STOPPED_CONN1);
+ KafkaFuture workerFencingFuture = EasyMock.mock(KafkaFuture.class);
+ KafkaFuture herderFencingFuture = EasyMock.mock(KafkaFuture.class);
+ EasyMock.expect(worker.fenceZombies(CONN1, SNAPSHOT_STOPPED_CONN1.taskCountRecord(CONN1), CONN1_CONFIG)).andReturn(workerFencingFuture);
+ EasyMock.expect(workerFencingFuture.thenApply(EasyMock.>anyObject())).andReturn(herderFencingFuture);
+
+ // Two fencing callbacks are added - one is in ZombieFencing::start itself to remove the connector from the active
+ // fencing list. The other is the callback passed from DistributedHerder::alterConnectorOffsets in order to
+ // queue up the actual alter offsets request if the zombie fencing succeeds.
+ for (int i = 0; i < 2; i++) {
+ Capture> herderFencingCallback = EasyMock.newCapture();
+ EasyMock.expect(herderFencingFuture.whenComplete(EasyMock.capture(herderFencingCallback))).andAnswer(() -> {
+ herderFencingCallback.getValue().accept(null, new ConnectException("Failed to perform zombie fencing"));
+ return null;
+ });
+ }
+
+ PowerMock.replayAll(workerFencingFuture, herderFencingFuture);
+
+ herder.tick();
+ FutureCallback callback = new FutureCallback<>();
+ herder.alterConnectorOffsets(CONN1, new HashMap<>(), callback);
+ // Process the zombie fencing request
+ herder.tick();
+ // Process the alter offsets request
+ herder.tick();
+ ExecutionException e = assertThrows(ExecutionException.class, () -> callback.get(1000L, TimeUnit.MILLISECONDS));
+ assertEquals(ConnectException.class, e.getCause().getClass());
+ assertEquals("Failed to perform zombie fencing for source connector prior to altering offsets",
+ e.getCause().getMessage());
+
+ PowerMock.verifyAll();
+ }
+
private void expectRebalance(final long offset,
final List assignedConnectors,
final List assignedTasks) {
diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/rest/entities/ConnectorOffsetsTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/rest/entities/ConnectorOffsetsTest.java
new file mode 100644
index 000000000000..726915d2e3a1
--- /dev/null
+++ b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/rest/entities/ConnectorOffsetsTest.java
@@ -0,0 +1,52 @@
+/*
+ * 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.runtime.rest.entities;
+
+import org.junit.Test;
+
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.junit.Assert.assertEquals;
+
+public class ConnectorOffsetsTest {
+
+ @Test
+ public void testConnectorOffsetsToMap() {
+ // Using arbitrary partition and offset formats here to demonstrate that source connector offsets don't
+ // follow a standard pattern
+ Map partition1 = new HashMap<>();
+ partition1.put("partitionKey1", "partitionValue");
+ partition1.put("k", 123);
+ Map offset1 = new HashMap<>();
+ offset1.put("offset", 3.14);
+ ConnectorOffset connectorOffset1 = new ConnectorOffset(partition1, offset1);
+
+ Map partition2 = new HashMap<>();
+ partition2.put("partitionKey1", true);
+ Map offset2 = new HashMap<>();
+ offset2.put("offset", new byte[]{0x00, 0x1A});
+ ConnectorOffset connectorOffset2 = new ConnectorOffset(partition2, offset2);
+
+ ConnectorOffsets connectorOffsets = new ConnectorOffsets(Arrays.asList(connectorOffset1, connectorOffset2));
+ Map, Map> connectorOffsetsMap = connectorOffsets.toMap();
+ assertEquals(2, connectorOffsetsMap.size());
+ assertEquals(offset1, connectorOffsetsMap.get(partition1));
+ assertEquals(offset2, connectorOffsetsMap.get(partition2));
+ }
+}
diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/rest/resources/ConnectorsResourceTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/rest/resources/ConnectorsResourceTest.java
index 95890a90497a..e202f35bd9f8 100644
--- a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/rest/resources/ConnectorsResourceTest.java
+++ b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/rest/resources/ConnectorsResourceTest.java
@@ -35,6 +35,7 @@
import org.apache.kafka.connect.runtime.rest.entities.ConnectorStateInfo;
import org.apache.kafka.connect.runtime.rest.entities.ConnectorType;
import org.apache.kafka.connect.runtime.rest.entities.CreateConnectorRequest;
+import org.apache.kafka.connect.runtime.rest.entities.Message;
import org.apache.kafka.connect.runtime.rest.entities.TaskInfo;
import org.apache.kafka.connect.runtime.rest.errors.ConnectRestException;
import org.apache.kafka.connect.util.Callback;
@@ -785,6 +786,58 @@ public void testGetOffsets() throws Throwable {
assertEquals(offsets, connectorsResource.getOffsets(CONNECTOR_NAME));
}
+ @Test
+ public void testAlterOffsetsEmptyOffsets() {
+ assertThrows(BadRequestException.class, () -> connectorsResource.alterConnectorOffsets(
+ false, NULL_HEADERS, CONNECTOR_NAME, new ConnectorOffsets(Collections.emptyList())));
+ }
+
+ @Test
+ public void testAlterOffsetsNotLeader() throws Throwable {
+ Map partition = new HashMap<>();
+ Map offset = new HashMap<>();
+ ConnectorOffset connectorOffset = new ConnectorOffset(partition, offset);
+ ConnectorOffsets body = new ConnectorOffsets(Collections.singletonList(connectorOffset));
+
+ final ArgumentCaptor> cb = ArgumentCaptor.forClass(Callback.class);
+ expectAndCallbackNotLeaderException(cb).when(herder).alterConnectorOffsets(eq(CONNECTOR_NAME), eq(body.toMap()), cb.capture());
+
+ when(restClient.httpRequest(eq(LEADER_URL + "connectors/" + CONNECTOR_NAME + "/offsets?forward=true"), eq("PATCH"), isNull(), eq(body), any()))
+ .thenReturn(new RestClient.HttpResponse<>(200, new HashMap<>(), new Message("")));
+ connectorsResource.alterConnectorOffsets(null, NULL_HEADERS, CONNECTOR_NAME, body);
+ }
+
+ @Test
+ public void testAlterOffsetsConnectorNotFound() {
+ Map partition = new HashMap<>();
+ Map offset = new HashMap<>();
+ ConnectorOffset connectorOffset = new ConnectorOffset(partition, offset);
+ ConnectorOffsets body = new ConnectorOffsets(Collections.singletonList(connectorOffset));
+ final ArgumentCaptor> cb = ArgumentCaptor.forClass(Callback.class);
+ expectAndCallbackException(cb, new NotFoundException("Connector not found"))
+ .when(herder).alterConnectorOffsets(eq(CONNECTOR_NAME), eq(body.toMap()), cb.capture());
+
+ assertThrows(NotFoundException.class, () -> connectorsResource.alterConnectorOffsets(null, NULL_HEADERS, CONNECTOR_NAME, body));
+ }
+
+ @Test
+ public void testAlterOffsets() throws Throwable {
+ Map partition = Collections.singletonMap("partitionKey", "partitionValue");
+ Map offset = Collections.singletonMap("offsetKey", "offsetValue");
+ ConnectorOffset connectorOffset = new ConnectorOffset(partition, offset);
+ ConnectorOffsets body = new ConnectorOffsets(Collections.singletonList(connectorOffset));
+
+ final ArgumentCaptor