diff --git a/checkstyle/suppressions.xml b/checkstyle/suppressions.xml
index a53a50a75fcd3..b4d475ad9aa66 100644
--- a/checkstyle/suppressions.xml
+++ b/checkstyle/suppressions.xml
@@ -115,14 +115,14 @@
+ files="(DistributedHerder|KafkaConfigBackingStore|Values|IncrementalCooperativeAssignor).java"/>
+ files="(RestServer|AbstractHerder|DistributedHerder|Worker).java"/>
@@ -133,7 +133,7 @@
files="(JsonConverter|Values|ConnectHeaders).java"/>
+ files="(KafkaConfigBackingStore|Values|ConnectMetricsRegistry).java"/>
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/Admin.java b/clients/src/main/java/org/apache/kafka/clients/admin/Admin.java
index 377f009a958df..0c795bc5206dc 100644
--- a/clients/src/main/java/org/apache/kafka/clients/admin/Admin.java
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/Admin.java
@@ -1575,6 +1575,29 @@ default ListTransactionsResult listTransactions() {
*/
ListTransactionsResult listTransactions(ListTransactionsOptions options);
+ /**
+ * Fence out all active producers that use any of the provided transactional IDs, with the default options.
+ *
+ * This is a convenience method for {@link #fenceProducers(Collection, FenceProducersOptions)}
+ * with default options. See the overload for more details.
+ *
+ * @param transactionalIds The IDs of the producers to fence.
+ * @return The FenceProducersResult.
+ */
+ default FenceProducersResult fenceProducers(Collection transactionalIds) {
+ return fenceProducers(transactionalIds, new FenceProducersOptions());
+ }
+
+ /**
+ * Fence out all active producers that use any of the provided transactional IDs.
+ *
+ * @param transactionalIds The IDs of the producers to fence.
+ * @param options The options to use when fencing the producers.
+ * @return The FenceProducersResult.
+ */
+ FenceProducersResult fenceProducers(Collection transactionalIds,
+ FenceProducersOptions options);
+
/**
* Get the metrics kept by the adminClient
*/
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/FenceProducersOptions.java b/clients/src/main/java/org/apache/kafka/clients/admin/FenceProducersOptions.java
new file mode 100644
index 0000000000000..4e38281809b83
--- /dev/null
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/FenceProducersOptions.java
@@ -0,0 +1,38 @@
+/*
+ * 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.clients.admin;
+
+import org.apache.kafka.common.annotation.InterfaceStability;
+
+import java.util.Collection;
+
+/**
+ * Options for {@link Admin#fenceProducers(Collection, FenceProducersOptions)}
+ *
+ * The API of this class is evolving. See {@link Admin} for details.
+ */
+@InterfaceStability.Evolving
+public class FenceProducersOptions extends AbstractOptions {
+
+ @Override
+ public String toString() {
+ return "FenceProducersOptions{" +
+ "timeoutMs=" + timeoutMs +
+ '}';
+ }
+}
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/FenceProducersResult.java b/clients/src/main/java/org/apache/kafka/clients/admin/FenceProducersResult.java
new file mode 100644
index 0000000000000..a1a8e66c115c5
--- /dev/null
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/FenceProducersResult.java
@@ -0,0 +1,84 @@
+/*
+ * 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.clients.admin;
+
+import org.apache.kafka.clients.admin.internals.CoordinatorKey;
+import org.apache.kafka.common.KafkaFuture;
+import org.apache.kafka.common.annotation.InterfaceStability;
+import org.apache.kafka.common.utils.ProducerIdAndEpoch;
+
+import java.util.Collection;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+/**
+ * The result of the {@link Admin#fenceProducers(Collection)} call.
+ *
+ * The API of this class is evolving, see {@link Admin} for details.
+ */
+@InterfaceStability.Evolving
+public class FenceProducersResult {
+
+ private final Map> futures;
+
+ FenceProducersResult(Map> futures) {
+ this.futures = futures;
+ }
+
+ /**
+ * Return a map from transactional ID to futures which can be used to check the status of
+ * individual fencings.
+ */
+ public Map> fencedProducers() {
+ return futures.entrySet().stream().collect(Collectors.toMap(
+ e -> e.getKey().idValue,
+ e -> e.getValue().thenApply(p -> null)
+ ));
+ }
+
+ /**
+ * Returns a future that provides the producer ID generated while initializing the given transaction when the request completes.
+ */
+ public KafkaFuture producerId(String transactionalId) {
+ return findAndApply(transactionalId, p -> p.producerId);
+ }
+
+ /**
+ * Returns a future that provides the epoch ID generated while initializing the given transaction when the request completes.
+ */
+ public KafkaFuture epochId(String transactionalId) {
+ return findAndApply(transactionalId, p -> p.epoch);
+ }
+
+ /**
+ * Return a future which succeeds only if all the producer fencings succeed.
+ */
+ public KafkaFuture all() {
+ return KafkaFuture.allOf(futures.values().toArray(new KafkaFuture[0]));
+ }
+
+ private KafkaFuture findAndApply(String transactionalId, KafkaFuture.BaseFunction followup) {
+ CoordinatorKey key = CoordinatorKey.byTransactionalId(transactionalId);
+ KafkaFuture future = futures.get(key);
+ if (future == null) {
+ throw new IllegalArgumentException("TransactionalId " +
+ "`" + transactionalId + "` was not included in the request");
+ }
+ return future.thenApply(followup);
+ }
+}
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java b/clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
index ea4e2d7e5ffb0..03322fdcf1dc8 100644
--- a/clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
@@ -47,6 +47,7 @@
import org.apache.kafka.clients.admin.internals.DescribeConsumerGroupsHandler;
import org.apache.kafka.clients.admin.internals.DescribeProducersHandler;
import org.apache.kafka.clients.admin.internals.DescribeTransactionsHandler;
+import org.apache.kafka.clients.admin.internals.FenceProducersHandler;
import org.apache.kafka.clients.admin.internals.ListConsumerGroupOffsetsHandler;
import org.apache.kafka.clients.admin.internals.ListTransactionsHandler;
import org.apache.kafka.clients.admin.internals.MetadataOperationContext;
@@ -234,6 +235,7 @@
import org.apache.kafka.common.utils.AppInfoParser;
import org.apache.kafka.common.utils.KafkaThread;
import org.apache.kafka.common.utils.LogContext;
+import org.apache.kafka.common.utils.ProducerIdAndEpoch;
import org.apache.kafka.common.utils.Time;
import org.apache.kafka.common.utils.Utils;
import org.slf4j.Logger;
@@ -4390,6 +4392,15 @@ public ListTransactionsResult listTransactions(ListTransactionsOptions options)
return new ListTransactionsResult(future.all());
}
+ @Override
+ public FenceProducersResult fenceProducers(Collection transactionalIds, FenceProducersOptions options) {
+ AdminApiFuture.SimpleAdminApiFuture future =
+ FenceProducersHandler.newFuture(transactionalIds);
+ FenceProducersHandler handler = new FenceProducersHandler(logContext);
+ invokeDriver(handler, future, options.timeoutMs);
+ return new FenceProducersResult(future.all());
+ }
+
private void invokeDriver(
AdminApiHandler handler,
AdminApiFuture future,
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/internals/AbortTransactionHandler.java b/clients/src/main/java/org/apache/kafka/clients/admin/internals/AbortTransactionHandler.java
index c25e4d8d3f479..4963a49c351b0 100644
--- a/clients/src/main/java/org/apache/kafka/clients/admin/internals/AbortTransactionHandler.java
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/internals/AbortTransactionHandler.java
@@ -38,7 +38,7 @@
import static java.util.Collections.singleton;
import static java.util.Collections.singletonList;
-public class AbortTransactionHandler implements AdminApiHandler {
+public class AbortTransactionHandler extends AdminApiHandler.Batched {
private final Logger log;
private final AbortTransactionSpec abortSpec;
private final PartitionLeaderStrategy lookupStrategy;
@@ -69,7 +69,7 @@ public AdminApiLookupStrategy lookupStrategy() {
}
@Override
- public WriteTxnMarkersRequest.Builder buildRequest(
+ public WriteTxnMarkersRequest.Builder buildBatchedRequest(
int brokerId,
Set topicPartitions
) {
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/internals/AdminApiDriver.java b/clients/src/main/java/org/apache/kafka/clients/admin/internals/AdminApiDriver.java
index b5c9ff32f264b..d00db4b18c694 100644
--- a/clients/src/main/java/org/apache/kafka/clients/admin/internals/AdminApiDriver.java
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/internals/AdminApiDriver.java
@@ -288,7 +288,7 @@ private void clearInflightRequest(long currentTimeMs, RequestSpec spec) {
private void collectRequests(
List> requests,
BiMultimap multimap,
- BiFunction, T, AbstractRequest.Builder>> buildRequest
+ BiFunction, T, Collection>> buildRequest
) {
for (Map.Entry> entry : multimap.entrySet()) {
T scope = entry.getKey();
@@ -306,12 +306,19 @@ private void collectRequests(
// Copy the keys to avoid exposing the underlying mutable set
Set copyKeys = Collections.unmodifiableSet(new HashSet<>(keys));
- AbstractRequest.Builder> request = buildRequest.apply(copyKeys, scope);
+ Collection> newRequests = buildRequest.apply(copyKeys, scope);
+ if (newRequests.isEmpty()) {
+ return;
+ }
+
+ // Only process the first request; all the remaining requests will be targeted at the same broker
+ // and we don't want to issue more than one fulfillment request per broker at a time
+ AdminApiHandler.RequestAndKeys newRequest = newRequests.iterator().next();
RequestSpec spec = new RequestSpec<>(
- handler.apiName() + "(api=" + request.apiKey() + ")",
+ handler.apiName() + "(api=" + newRequest.request.apiKey() + ")",
scope,
- copyKeys,
- request,
+ newRequest.keys,
+ newRequest.request,
requestState.nextAllowedRetryMs,
deadlineMs,
requestState.tries
@@ -326,7 +333,7 @@ private void collectLookupRequests(List> requests) {
collectRequests(
requests,
lookupMap,
- (keys, scope) -> handler.lookupStrategy().buildRequest(keys)
+ (keys, scope) -> Collections.singletonList(new AdminApiHandler.RequestAndKeys<>(handler.lookupStrategy().buildRequest(keys), keys))
);
}
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/internals/AdminApiHandler.java b/clients/src/main/java/org/apache/kafka/clients/admin/internals/AdminApiHandler.java
index 9f8d0ac5f07f2..79df965ffb0f9 100644
--- a/clients/src/main/java/org/apache/kafka/clients/admin/internals/AdminApiHandler.java
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/internals/AdminApiHandler.java
@@ -20,10 +20,12 @@
import org.apache.kafka.common.requests.AbstractRequest;
import org.apache.kafka.common.requests.AbstractResponse;
+import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Set;
+import java.util.stream.Collectors;
public interface AdminApiHandler {
@@ -33,16 +35,18 @@ public interface AdminApiHandler {
String apiName();
/**
- * Build the request. The set of keys are derived by {@link AdminApiDriver}
- * during the lookup stage as the set of keys which all map to the same
- * destination broker.
+ * Build the requests necessary for the given keys. The set of keys is derived by
+ * {@link AdminApiDriver} during the lookup stage as the set of keys which all map
+ * to the same destination broker. Handlers can choose to issue a single request for
+ * all of the provided keys (see {@link Batched}, issue one request per key (see
+ * {@link Unbatched}, or implement their own custom grouping logic if necessary.
*
* @param brokerId the target brokerId for the request
* @param keys the set of keys that should be handled by this request
*
- * @return a builder for the request containing the given keys
+ * @return a collection of {@link RequestAndKeys} for the requests containing the given keys
*/
- AbstractRequest.Builder> buildRequest(int brokerId, Set keys);
+ Collection> buildRequest(int brokerId, Set keys);
/**
* Callback that is invoked when a request returns successfully.
@@ -122,4 +126,54 @@ public static ApiResult empty() {
}
}
+ class RequestAndKeys {
+ public final AbstractRequest.Builder> request;
+ public final Set keys;
+
+ public RequestAndKeys(AbstractRequest.Builder> request, Set keys) {
+ this.request = request;
+ this.keys = keys;
+ }
+ }
+
+ /**
+ * An {@link AdminApiHandler} that will group multiple keys into a single request when possible.
+ * Keys will be grouped together whenever they target the same broker. This type of handler
+ * should be used when when interacting with broker APIs that can act on multiple keys at once,
+ * such as describing or listing transactions.
+ */
+ abstract class Batched implements AdminApiHandler {
+ abstract AbstractRequest.Builder> buildBatchedRequest(int brokerId, Set keys);
+
+ @Override
+ public final Collection> buildRequest(int brokerId, Set keys) {
+ return Collections.singleton(new RequestAndKeys<>(buildBatchedRequest(brokerId, keys), keys));
+ }
+ }
+
+ /**
+ * An {@link AdminApiHandler} that will create one request per key, not performing any grouping based
+ * on the targeted broker. This type of handler should only be used for broker APIs that do not accept
+ * multiple keys at once, such as initializing a transactional producer.
+ */
+ abstract class Unbatched implements AdminApiHandler {
+ abstract AbstractRequest.Builder> buildSingleRequest(int brokerId, K key);
+ abstract ApiResult handleSingleResponse(Node broker, K key, AbstractResponse response);
+
+ @Override
+ public final Collection> buildRequest(int brokerId, Set keys) {
+ return keys.stream()
+ .map(key -> new RequestAndKeys<>(buildSingleRequest(brokerId, key), Collections.singleton(key)))
+ .collect(Collectors.toSet());
+ }
+
+ @Override
+ public final ApiResult handleResponse(Node broker, Set keys, AbstractResponse response) {
+ if (keys.size() != 1) {
+ throw new IllegalArgumentException("Unbatched admin handler should only be required to handle responses for a single key at a time");
+ }
+ K key = keys.iterator().next();
+ return handleSingleResponse(broker, key, response);
+ }
+ }
}
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/internals/AlterConsumerGroupOffsetsHandler.java b/clients/src/main/java/org/apache/kafka/clients/admin/internals/AlterConsumerGroupOffsetsHandler.java
index cb7551e55ed21..eab2e2bb73a40 100644
--- a/clients/src/main/java/org/apache/kafka/clients/admin/internals/AlterConsumerGroupOffsetsHandler.java
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/internals/AlterConsumerGroupOffsetsHandler.java
@@ -41,7 +41,7 @@
import org.apache.kafka.common.utils.LogContext;
import org.slf4j.Logger;
-public class AlterConsumerGroupOffsetsHandler implements AdminApiHandler> {
+public class AlterConsumerGroupOffsetsHandler extends AdminApiHandler.Batched> {
private final CoordinatorKey groupId;
private final Map offsets;
@@ -83,7 +83,7 @@ private void validateKeys(Set groupIds) {
}
@Override
- public OffsetCommitRequest.Builder buildRequest(
+ public OffsetCommitRequest.Builder buildBatchedRequest(
int coordinatorId,
Set groupIds
) {
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/internals/DeleteConsumerGroupOffsetsHandler.java b/clients/src/main/java/org/apache/kafka/clients/admin/internals/DeleteConsumerGroupOffsetsHandler.java
index a853eddcb31b1..b68334b55c8ae 100644
--- a/clients/src/main/java/org/apache/kafka/clients/admin/internals/DeleteConsumerGroupOffsetsHandler.java
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/internals/DeleteConsumerGroupOffsetsHandler.java
@@ -38,7 +38,7 @@
import org.apache.kafka.common.utils.LogContext;
import org.slf4j.Logger;
-public class DeleteConsumerGroupOffsetsHandler implements AdminApiHandler> {
+public class DeleteConsumerGroupOffsetsHandler extends AdminApiHandler.Batched> {
private final CoordinatorKey groupId;
private final Set partitions;
@@ -80,7 +80,7 @@ private void validateKeys(Set groupIds) {
}
@Override
- public OffsetDeleteRequest.Builder buildRequest(int coordinatorId, Set groupIds) {
+ public OffsetDeleteRequest.Builder buildBatchedRequest(int coordinatorId, Set groupIds) {
validateKeys(groupIds);
final OffsetDeleteRequestTopicCollection topics = new OffsetDeleteRequestTopicCollection();
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/internals/DeleteConsumerGroupsHandler.java b/clients/src/main/java/org/apache/kafka/clients/admin/internals/DeleteConsumerGroupsHandler.java
index 693d23625e8d3..c05a11b5cfb40 100644
--- a/clients/src/main/java/org/apache/kafka/clients/admin/internals/DeleteConsumerGroupsHandler.java
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/internals/DeleteConsumerGroupsHandler.java
@@ -36,7 +36,7 @@
import org.apache.kafka.common.utils.LogContext;
import org.slf4j.Logger;
-public class DeleteConsumerGroupsHandler implements AdminApiHandler {
+public class DeleteConsumerGroupsHandler extends AdminApiHandler.Batched {
private final Logger log;
private final AdminApiLookupStrategy lookupStrategy;
@@ -71,7 +71,7 @@ private static Set buildKeySet(Collection groupIds) {
}
@Override
- public DeleteGroupsRequest.Builder buildRequest(
+ public DeleteGroupsRequest.Builder buildBatchedRequest(
int coordinatorId,
Set keys
) {
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/internals/DescribeConsumerGroupsHandler.java b/clients/src/main/java/org/apache/kafka/clients/admin/internals/DescribeConsumerGroupsHandler.java
index 5c5022a37eccc..09a2a7b4cdcdf 100644
--- a/clients/src/main/java/org/apache/kafka/clients/admin/internals/DescribeConsumerGroupsHandler.java
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/internals/DescribeConsumerGroupsHandler.java
@@ -51,7 +51,7 @@
import org.apache.kafka.common.utils.Utils;
import org.slf4j.Logger;
-public class DescribeConsumerGroupsHandler implements AdminApiHandler {
+public class DescribeConsumerGroupsHandler extends AdminApiHandler.Batched {
private final boolean includeAuthorizedOperations;
private final Logger log;
@@ -89,7 +89,7 @@ public AdminApiLookupStrategy lookupStrategy() {
}
@Override
- public DescribeGroupsRequest.Builder buildRequest(int coordinatorId, Set keys) {
+ public DescribeGroupsRequest.Builder buildBatchedRequest(int coordinatorId, Set keys) {
List groupIds = keys.stream().map(key -> {
if (key.type != FindCoordinatorRequest.CoordinatorType.GROUP) {
throw new IllegalArgumentException("Invalid transaction coordinator key " + key +
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/internals/DescribeProducersHandler.java b/clients/src/main/java/org/apache/kafka/clients/admin/internals/DescribeProducersHandler.java
index 4b279d5c90c78..8555bf59ab4ed 100644
--- a/clients/src/main/java/org/apache/kafka/clients/admin/internals/DescribeProducersHandler.java
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/internals/DescribeProducersHandler.java
@@ -46,7 +46,7 @@
import java.util.Set;
import java.util.stream.Collectors;
-public class DescribeProducersHandler implements AdminApiHandler {
+public class DescribeProducersHandler extends AdminApiHandler.Batched {
private final Logger log;
private final DescribeProducersOptions options;
private final AdminApiLookupStrategy lookupStrategy;
@@ -82,7 +82,7 @@ public AdminApiLookupStrategy lookupStrategy() {
}
@Override
- public DescribeProducersRequest.Builder buildRequest(
+ public DescribeProducersRequest.Builder buildBatchedRequest(
int brokerId,
Set topicPartitions
) {
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/internals/DescribeTransactionsHandler.java b/clients/src/main/java/org/apache/kafka/clients/admin/internals/DescribeTransactionsHandler.java
index d270145a423da..4a3d7c0a05272 100644
--- a/clients/src/main/java/org/apache/kafka/clients/admin/internals/DescribeTransactionsHandler.java
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/internals/DescribeTransactionsHandler.java
@@ -43,7 +43,7 @@
import java.util.Set;
import java.util.stream.Collectors;
-public class DescribeTransactionsHandler implements AdminApiHandler {
+public class DescribeTransactionsHandler extends AdminApiHandler.Batched {
private final Logger log;
private final AdminApiLookupStrategy lookupStrategy;
@@ -77,7 +77,7 @@ public AdminApiLookupStrategy lookupStrategy() {
}
@Override
- public DescribeTransactionsRequest.Builder buildRequest(
+ public DescribeTransactionsRequest.Builder buildBatchedRequest(
int brokerId,
Set keys
) {
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/internals/FenceProducersHandler.java b/clients/src/main/java/org/apache/kafka/clients/admin/internals/FenceProducersHandler.java
new file mode 100644
index 0000000000000..7b6ff817efac2
--- /dev/null
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/internals/FenceProducersHandler.java
@@ -0,0 +1,146 @@
+/*
+ * 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.clients.admin.internals;
+
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.errors.InvalidProducerEpochException;
+import org.apache.kafka.common.errors.TransactionalIdAuthorizationException;
+import org.apache.kafka.common.errors.TransactionalIdNotFoundException;
+import org.apache.kafka.common.message.InitProducerIdRequestData;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.requests.AbstractResponse;
+import org.apache.kafka.common.requests.FindCoordinatorRequest;
+import org.apache.kafka.common.requests.InitProducerIdRequest;
+import org.apache.kafka.common.requests.InitProducerIdResponse;
+import org.apache.kafka.common.utils.LogContext;
+import org.apache.kafka.common.utils.ProducerIdAndEpoch;
+import org.slf4j.Logger;
+
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+public class FenceProducersHandler extends AdminApiHandler.Unbatched {
+ private final Logger log;
+ private final AdminApiLookupStrategy lookupStrategy;
+
+ public FenceProducersHandler(
+ LogContext logContext
+ ) {
+ this.log = logContext.logger(DescribeTransactionsHandler.class);
+ this.lookupStrategy = new CoordinatorStrategy(FindCoordinatorRequest.CoordinatorType.TRANSACTION, logContext);
+ }
+
+ public static AdminApiFuture.SimpleAdminApiFuture newFuture(
+ Collection transactionalIds
+ ) {
+ return AdminApiFuture.forKeys(buildKeySet(transactionalIds));
+ }
+
+ private static Set buildKeySet(Collection transactionalIds) {
+ return transactionalIds.stream()
+ .map(CoordinatorKey::byTransactionalId)
+ .collect(Collectors.toSet());
+ }
+
+ @Override
+ public String apiName() {
+ return "fenceProducer";
+ }
+
+ @Override
+ public AdminApiLookupStrategy lookupStrategy() {
+ return lookupStrategy;
+ }
+
+ @Override
+ InitProducerIdRequest.Builder buildSingleRequest(int brokerId, CoordinatorKey key) {
+ if (key.type != FindCoordinatorRequest.CoordinatorType.TRANSACTION) {
+ throw new IllegalArgumentException("Invalid group coordinator key " + key +
+ " when building `InitProducerId` request");
+ }
+ InitProducerIdRequestData data = new InitProducerIdRequestData()
+ .setProducerEpoch(ProducerIdAndEpoch.NONE.epoch)
+ .setProducerId(ProducerIdAndEpoch.NONE.producerId)
+ .setTransactionalId(key.idValue)
+ // Set transaction timeout to 1 since it's only being initialized to fence out older producers with the same transactional ID,
+ // and shouldn't be used for any actual record writes
+ .setTransactionTimeoutMs(1);
+ return new InitProducerIdRequest.Builder(data);
+ }
+
+ @Override
+ public ApiResult handleSingleResponse(
+ Node broker,
+ CoordinatorKey key,
+ AbstractResponse abstractResponse
+ ) {
+ InitProducerIdResponse response = (InitProducerIdResponse) abstractResponse;
+
+ Errors error = Errors.forCode(response.data().errorCode());
+ if (error != Errors.NONE) {
+ return handleError(key, error);
+ }
+
+ Map completed = Collections.singletonMap(key, new ProducerIdAndEpoch(
+ response.data().producerId(),
+ response.data().producerEpoch()
+ ));
+
+ return new ApiResult<>(completed, Collections.emptyMap(), Collections.emptyList());
+ }
+
+ private ApiResult handleError(CoordinatorKey transactionalIdKey, Errors error) {
+ switch (error) {
+ case TRANSACTIONAL_ID_AUTHORIZATION_FAILED:
+ case CLUSTER_AUTHORIZATION_FAILED:
+ return ApiResult.failed(transactionalIdKey, new TransactionalIdAuthorizationException(
+ "InitProducerId request for transactionalId `" + transactionalIdKey.idValue + "` " +
+ "failed due to authorization failure"));
+
+ case TRANSACTIONAL_ID_NOT_FOUND:
+ return ApiResult.failed(transactionalIdKey, new TransactionalIdNotFoundException(
+ "InitProducerId request for transactionalId `" + transactionalIdKey.idValue + "` " +
+ "failed because the ID could not be found"));
+
+ case INVALID_PRODUCER_EPOCH:
+ return ApiResult.failed(transactionalIdKey, new InvalidProducerEpochException(
+ "InitProducerId request with " + transactionalIdKey.idValue + " failed due an invalid producer epoch"));
+
+ case COORDINATOR_LOAD_IN_PROGRESS:
+ // If the coordinator is in the middle of loading, then we just need to retry
+ log.debug("InitProducerId request for transactionalId `{}` failed because the " +
+ "coordinator is still in the process of loading state. Will retry",
+ transactionalIdKey.idValue);
+ return ApiResult.empty();
+
+ case NOT_COORDINATOR:
+ case COORDINATOR_NOT_AVAILABLE:
+ // If the coordinator is unavailable or there was a coordinator change, then we unmap
+ // the key so that we retry the `FindCoordinator` request
+ log.debug("InitProducerId request for transactionalId `{}` returned error {}. Will attempt " +
+ "to find the coordinator again and retry", transactionalIdKey.idValue, error);
+ return ApiResult.unmapped(Collections.singletonList(transactionalIdKey));
+
+ default:
+ return ApiResult.failed(transactionalIdKey, error.exception("InitProducerId request for " +
+ "transactionalId `" + transactionalIdKey.idValue + "` failed due to unexpected error"));
+ }
+ }
+}
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/internals/ListConsumerGroupOffsetsHandler.java b/clients/src/main/java/org/apache/kafka/clients/admin/internals/ListConsumerGroupOffsetsHandler.java
index b1d2e9dc69665..b591548954b96 100644
--- a/clients/src/main/java/org/apache/kafka/clients/admin/internals/ListConsumerGroupOffsetsHandler.java
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/internals/ListConsumerGroupOffsetsHandler.java
@@ -36,7 +36,7 @@
import org.apache.kafka.common.utils.LogContext;
import org.slf4j.Logger;
-public class ListConsumerGroupOffsetsHandler implements AdminApiHandler> {
+public class ListConsumerGroupOffsetsHandler extends AdminApiHandler.Batched> {
private final CoordinatorKey groupId;
private final List partitions;
@@ -78,7 +78,7 @@ private void validateKeys(Set groupIds) {
}
@Override
- public OffsetFetchRequest.Builder buildRequest(int coordinatorId, Set groupIds) {
+ public OffsetFetchRequest.Builder buildBatchedRequest(int coordinatorId, Set groupIds) {
validateKeys(groupIds);
// Set the flag to false as for admin client request,
// we don't need to wait for any pending offset state to clear.
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/internals/ListTransactionsHandler.java b/clients/src/main/java/org/apache/kafka/clients/admin/internals/ListTransactionsHandler.java
index d60580c85bf22..ca249bca7f4a4 100644
--- a/clients/src/main/java/org/apache/kafka/clients/admin/internals/ListTransactionsHandler.java
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/internals/ListTransactionsHandler.java
@@ -35,7 +35,7 @@
import java.util.Set;
import java.util.stream.Collectors;
-public class ListTransactionsHandler implements AdminApiHandler> {
+public class ListTransactionsHandler extends AdminApiHandler.Batched> {
private final Logger log;
private final ListTransactionsOptions options;
private final AllBrokersStrategy lookupStrategy;
@@ -64,7 +64,7 @@ public AdminApiLookupStrategy lookupStrategy() {
}
@Override
- public ListTransactionsRequest.Builder buildRequest(
+ public ListTransactionsRequest.Builder buildBatchedRequest(
int brokerId,
Set keys
) {
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/internals/RemoveMembersFromConsumerGroupHandler.java b/clients/src/main/java/org/apache/kafka/clients/admin/internals/RemoveMembersFromConsumerGroupHandler.java
index 90b3865d0bd65..83aee36587d3a 100644
--- a/clients/src/main/java/org/apache/kafka/clients/admin/internals/RemoveMembersFromConsumerGroupHandler.java
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/internals/RemoveMembersFromConsumerGroupHandler.java
@@ -35,7 +35,7 @@
import org.apache.kafka.common.utils.LogContext;
import org.slf4j.Logger;
-public class RemoveMembersFromConsumerGroupHandler implements AdminApiHandler> {
+public class RemoveMembersFromConsumerGroupHandler extends AdminApiHandler.Batched> {
private final CoordinatorKey groupId;
private final List members;
@@ -79,7 +79,7 @@ private void validateKeys(
}
@Override
- public LeaveGroupRequest.Builder buildRequest(int coordinatorId, Set groupIds) {
+ public LeaveGroupRequest.Builder buildBatchedRequest(int coordinatorId, Set groupIds) {
validateKeys(groupIds);
return new LeaveGroupRequest.Builder(groupId.idValue, members);
}
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/InitProducerIdRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/InitProducerIdRequest.java
index 5c24b41b351df..9d92f0e5351dd 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/InitProducerIdRequest.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/InitProducerIdRequest.java
@@ -27,7 +27,7 @@
public class InitProducerIdRequest extends AbstractRequest {
public static class Builder extends AbstractRequest.Builder {
- private final InitProducerIdRequestData data;
+ public final InitProducerIdRequestData data;
public Builder(InitProducerIdRequestData data) {
super(ApiKeys.INIT_PRODUCER_ID);
diff --git a/clients/src/main/java/org/apache/kafka/common/utils/Utils.java b/clients/src/main/java/org/apache/kafka/common/utils/Utils.java
index 0685d1d49b76a..741fd332fa31c 100755
--- a/clients/src/main/java/org/apache/kafka/common/utils/Utils.java
+++ b/clients/src/main/java/org/apache/kafka/common/utils/Utils.java
@@ -1415,4 +1415,21 @@ public static Map initializeMap(Collection keys, Supplier val
return res;
}
+ /**
+ * Get an array containing all of the {@link Object#toString names} of a given enumerable type.
+ * @param enumClass the enum class; may not be null
+ * @return an array with the names of every value for the enum class; never null, but may be empty
+ * if there are no values defined for the enum
+ */
+ public static String[] enumOptions(Class extends Enum>> enumClass) {
+ Objects.requireNonNull(enumClass);
+ if (!enumClass.isEnum()) {
+ throw new IllegalArgumentException("Class " + enumClass + " is not an enumerable type");
+ }
+
+ return Stream.of(enumClass.getEnumConstants())
+ .map(Object::toString)
+ .toArray(String[]::new);
+ }
+
}
diff --git a/clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java b/clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java
index fcd93a96b3989..c936bab6d6aae 100644
--- a/clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java
@@ -109,6 +109,7 @@
import org.apache.kafka.common.message.IncrementalAlterConfigsResponseData;
import org.apache.kafka.common.message.IncrementalAlterConfigsResponseData.AlterConfigsResourceResponse;
import org.apache.kafka.common.message.LeaveGroupRequestData;
+import org.apache.kafka.common.message.InitProducerIdResponseData;
import org.apache.kafka.common.message.LeaveGroupRequestData.MemberIdentity;
import org.apache.kafka.common.message.LeaveGroupResponseData;
import org.apache.kafka.common.message.LeaveGroupResponseData.MemberResponse;
@@ -167,6 +168,8 @@
import org.apache.kafka.common.requests.FindCoordinatorRequest;
import org.apache.kafka.common.requests.FindCoordinatorResponse;
import org.apache.kafka.common.requests.IncrementalAlterConfigsResponse;
+import org.apache.kafka.common.requests.InitProducerIdRequest;
+import org.apache.kafka.common.requests.InitProducerIdResponse;
import org.apache.kafka.common.requests.JoinGroupRequest;
import org.apache.kafka.common.requests.LeaveGroupRequest;
import org.apache.kafka.common.requests.LeaveGroupResponse;
@@ -6281,6 +6284,30 @@ public void testClientSideTimeoutAfterFailureToReceiveResponse() throws Exceptio
}
}
+ @Test
+ public void testFenceProducers() throws Exception {
+ try (AdminClientUnitTestEnv env = mockClientEnv()) {
+ String transactionalId = "copyCat";
+ Node transactionCoordinator = env.cluster().nodes().iterator().next();
+
+ env.kafkaClient().prepareResponse(prepareFindCoordinatorResponse(Errors.NONE, transactionalId, transactionCoordinator));
+
+ InitProducerIdResponseData initProducerIdResponseData = new InitProducerIdResponseData()
+ .setProducerId(4761)
+ .setProducerEpoch((short) 489);
+ env.kafkaClient().prepareResponseFrom(
+ request -> request instanceof InitProducerIdRequest,
+ new InitProducerIdResponse(initProducerIdResponseData),
+ transactionCoordinator
+ );
+
+ FenceProducersResult result = env.adminClient().fenceProducers(Collections.singleton(transactionalId));
+ assertNull(result.all().get());
+ assertEquals(4761, result.producerId(transactionalId).get());
+ assertEquals((short) 489, result.epochId(transactionalId).get());
+ }
+ }
+
private UnregisterBrokerResponse prepareUnregisterBrokerResponse(Errors error, int throttleTimeMs) {
return new UnregisterBrokerResponse(new UnregisterBrokerResponseData()
.setErrorCode(error.code())
diff --git a/clients/src/test/java/org/apache/kafka/clients/admin/MockAdminClient.java b/clients/src/test/java/org/apache/kafka/clients/admin/MockAdminClient.java
index 1593b6328df74..15cdc5ccc4116 100644
--- a/clients/src/test/java/org/apache/kafka/clients/admin/MockAdminClient.java
+++ b/clients/src/test/java/org/apache/kafka/clients/admin/MockAdminClient.java
@@ -1003,6 +1003,11 @@ public ListTransactionsResult listTransactions(ListTransactionsOptions options)
throw new UnsupportedOperationException("Not implemented yet");
}
+ @Override
+ public FenceProducersResult fenceProducers(Collection transactionalIds, FenceProducersOptions options) {
+ throw new UnsupportedOperationException("Not implemented yet");
+ }
+
@Override
synchronized public void close(Duration timeout) {}
diff --git a/clients/src/test/java/org/apache/kafka/clients/admin/internals/AbortTransactionHandlerTest.java b/clients/src/test/java/org/apache/kafka/clients/admin/internals/AbortTransactionHandlerTest.java
index 78b33a7f59dff..68c850583d6ae 100644
--- a/clients/src/test/java/org/apache/kafka/clients/admin/internals/AbortTransactionHandlerTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/admin/internals/AbortTransactionHandlerTest.java
@@ -63,7 +63,7 @@ public void testInvalidBuildRequestCall() {
@Test
public void testValidBuildRequestCall() {
AbortTransactionHandler handler = new AbortTransactionHandler(abortSpec, logContext);
- WriteTxnMarkersRequest.Builder request = handler.buildRequest(1, singleton(topicPartition));
+ WriteTxnMarkersRequest.Builder request = handler.buildBatchedRequest(1, singleton(topicPartition));
assertEquals(1, request.data.markers().size());
WriteTxnMarkersRequestData.WritableTxnMarker markerRequest = request.data.markers().get(0);
diff --git a/clients/src/test/java/org/apache/kafka/clients/admin/internals/AdminApiDriverTest.java b/clients/src/test/java/org/apache/kafka/clients/admin/internals/AdminApiDriverTest.java
index 6ff393fddd65e..2c664e857fced 100644
--- a/clients/src/test/java/org/apache/kafka/clients/admin/internals/AdminApiDriverTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/admin/internals/AdminApiDriverTest.java
@@ -734,7 +734,7 @@ public void reset() {
}
}
- private static class MockAdminApiHandler implements AdminApiHandler {
+ private static class MockAdminApiHandler extends AdminApiHandler.Batched {
private final Map, ApiResult> expectedRequests = new HashMap<>();
private final MockLookupStrategy lookupStrategy;
@@ -757,7 +757,7 @@ public void expectRequest(Set keys, ApiResult result) {
}
@Override
- public AbstractRequest.Builder> buildRequest(int brokerId, Set keys) {
+ public AbstractRequest.Builder> buildBatchedRequest(int brokerId, Set keys) {
// The request is just a placeholder in these tests
assertTrue(expectedRequests.containsKey(keys), "Unexpected fulfillment request for keys " + keys);
return new MetadataRequest.Builder(Collections.emptyList(), false);
diff --git a/clients/src/test/java/org/apache/kafka/clients/admin/internals/AllBrokersStrategyIntegrationTest.java b/clients/src/test/java/org/apache/kafka/clients/admin/internals/AllBrokersStrategyIntegrationTest.java
index 2b989058082fb..b2e0a0d285b06 100644
--- a/clients/src/test/java/org/apache/kafka/clients/admin/internals/AllBrokersStrategyIntegrationTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/admin/internals/AllBrokersStrategyIntegrationTest.java
@@ -215,7 +215,7 @@ private MetadataResponse responseWithBrokers(Set brokerIds) {
return new MetadataResponse(response, ApiKeys.METADATA.latestVersion());
}
- private class MockApiHandler implements AdminApiHandler {
+ private class MockApiHandler extends AdminApiHandler.Batched {
private final AllBrokersStrategy allBrokersStrategy = new AllBrokersStrategy(logContext);
@Override
@@ -224,7 +224,7 @@ public String apiName() {
}
@Override
- public AbstractRequest.Builder> buildRequest(
+ public AbstractRequest.Builder> buildBatchedRequest(
int brokerId,
Set keys
) {
diff --git a/clients/src/test/java/org/apache/kafka/clients/admin/internals/AlterConsumerGroupOffsetsHandlerTest.java b/clients/src/test/java/org/apache/kafka/clients/admin/internals/AlterConsumerGroupOffsetsHandlerTest.java
index c0ea2ba9f0e7e..80453cc9b85d4 100644
--- a/clients/src/test/java/org/apache/kafka/clients/admin/internals/AlterConsumerGroupOffsetsHandlerTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/admin/internals/AlterConsumerGroupOffsetsHandlerTest.java
@@ -62,7 +62,7 @@ public void setUp() {
@Test
public void testBuildRequest() {
AlterConsumerGroupOffsetsHandler handler = new AlterConsumerGroupOffsetsHandler(groupId, partitions, logContext);
- OffsetCommitRequest request = handler.buildRequest(-1, singleton(CoordinatorKey.byGroupId(groupId))).build();
+ OffsetCommitRequest request = handler.buildBatchedRequest(-1, singleton(CoordinatorKey.byGroupId(groupId))).build();
assertEquals(groupId, request.data().groupId());
assertEquals(2, request.data().topics().size());
assertEquals(2, request.data().topics().get(0).partitions().size());
diff --git a/clients/src/test/java/org/apache/kafka/clients/admin/internals/DeleteConsumerGroupOffsetsHandlerTest.java b/clients/src/test/java/org/apache/kafka/clients/admin/internals/DeleteConsumerGroupOffsetsHandlerTest.java
index b4aea93c3f3f6..629ca89801559 100644
--- a/clients/src/test/java/org/apache/kafka/clients/admin/internals/DeleteConsumerGroupOffsetsHandlerTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/admin/internals/DeleteConsumerGroupOffsetsHandlerTest.java
@@ -59,7 +59,7 @@ public class DeleteConsumerGroupOffsetsHandlerTest {
@Test
public void testBuildRequest() {
DeleteConsumerGroupOffsetsHandler handler = new DeleteConsumerGroupOffsetsHandler(groupId, tps, logContext);
- OffsetDeleteRequest request = handler.buildRequest(1, singleton(CoordinatorKey.byGroupId(groupId))).build();
+ OffsetDeleteRequest request = handler.buildBatchedRequest(1, singleton(CoordinatorKey.byGroupId(groupId))).build();
assertEquals(groupId, request.data().groupId());
assertEquals(2, request.data().topics().size());
assertEquals(2, request.data().topics().find("t0").partitions().size());
diff --git a/clients/src/test/java/org/apache/kafka/clients/admin/internals/DeleteConsumerGroupsHandlerTest.java b/clients/src/test/java/org/apache/kafka/clients/admin/internals/DeleteConsumerGroupsHandlerTest.java
index 8d3a2376fde9a..3e7cead6f5d40 100644
--- a/clients/src/test/java/org/apache/kafka/clients/admin/internals/DeleteConsumerGroupsHandlerTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/admin/internals/DeleteConsumerGroupsHandlerTest.java
@@ -45,7 +45,7 @@ public class DeleteConsumerGroupsHandlerTest {
@Test
public void testBuildRequest() {
DeleteConsumerGroupsHandler handler = new DeleteConsumerGroupsHandler(logContext);
- DeleteGroupsRequest request = handler.buildRequest(1, singleton(CoordinatorKey.byGroupId(groupId1))).build();
+ DeleteGroupsRequest request = handler.buildBatchedRequest(1, singleton(CoordinatorKey.byGroupId(groupId1))).build();
assertEquals(1, request.data().groupsNames().size());
assertEquals(groupId1, request.data().groupsNames().get(0));
}
diff --git a/clients/src/test/java/org/apache/kafka/clients/admin/internals/DescribeConsumerGroupsHandlerTest.java b/clients/src/test/java/org/apache/kafka/clients/admin/internals/DescribeConsumerGroupsHandlerTest.java
index aef207aca6a75..892a81e8c0e11 100644
--- a/clients/src/test/java/org/apache/kafka/clients/admin/internals/DescribeConsumerGroupsHandlerTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/admin/internals/DescribeConsumerGroupsHandlerTest.java
@@ -69,12 +69,12 @@ public class DescribeConsumerGroupsHandlerTest {
@Test
public void testBuildRequest() {
DescribeConsumerGroupsHandler handler = new DescribeConsumerGroupsHandler(false, logContext);
- DescribeGroupsRequest request = handler.buildRequest(1, keys).build();
+ DescribeGroupsRequest request = handler.buildBatchedRequest(1, keys).build();
assertEquals(2, request.data().groups().size());
assertFalse(request.data().includeAuthorizedOperations());
handler = new DescribeConsumerGroupsHandler(true, logContext);
- request = handler.buildRequest(1, keys).build();
+ request = handler.buildBatchedRequest(1, keys).build();
assertEquals(2, request.data().groups().size());
assertTrue(request.data().includeAuthorizedOperations());
}
diff --git a/clients/src/test/java/org/apache/kafka/clients/admin/internals/DescribeProducersHandlerTest.java b/clients/src/test/java/org/apache/kafka/clients/admin/internals/DescribeProducersHandlerTest.java
index 8daed06ddf133..0f39b4dd01633 100644
--- a/clients/src/test/java/org/apache/kafka/clients/admin/internals/DescribeProducersHandlerTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/admin/internals/DescribeProducersHandlerTest.java
@@ -118,7 +118,7 @@ public void testBuildRequest() {
);
int brokerId = 3;
- DescribeProducersRequest.Builder request = handler.buildRequest(brokerId, topicPartitions);
+ DescribeProducersRequest.Builder request = handler.buildBatchedRequest(brokerId, topicPartitions);
List topics = request.data.topics();
diff --git a/clients/src/test/java/org/apache/kafka/clients/admin/internals/DescribeTransactionsHandlerTest.java b/clients/src/test/java/org/apache/kafka/clients/admin/internals/DescribeTransactionsHandlerTest.java
index 04eac89a373ac..7ffda2b00e014 100644
--- a/clients/src/test/java/org/apache/kafka/clients/admin/internals/DescribeTransactionsHandlerTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/admin/internals/DescribeTransactionsHandlerTest.java
@@ -156,7 +156,7 @@ private void assertLookup(
Set transactionalIds
) {
Set keys = coordinatorKeys(transactionalIds);
- DescribeTransactionsRequest.Builder request = handler.buildRequest(1, keys);
+ DescribeTransactionsRequest.Builder request = handler.buildBatchedRequest(1, keys);
assertEquals(transactionalIds, new HashSet<>(request.data.transactionalIds()));
}
diff --git a/clients/src/test/java/org/apache/kafka/clients/admin/internals/FenceProducersHandlerTest.java b/clients/src/test/java/org/apache/kafka/clients/admin/internals/FenceProducersHandlerTest.java
new file mode 100644
index 0000000000000..08e856ec3fa16
--- /dev/null
+++ b/clients/src/test/java/org/apache/kafka/clients/admin/internals/FenceProducersHandlerTest.java
@@ -0,0 +1,145 @@
+/*
+ * 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.clients.admin.internals;
+
+import org.apache.kafka.clients.admin.internals.AdminApiHandler.ApiResult;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.message.InitProducerIdResponseData;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.requests.InitProducerIdRequest;
+import org.apache.kafka.common.requests.InitProducerIdResponse;
+import org.apache.kafka.common.utils.LogContext;
+import org.apache.kafka.common.utils.ProducerIdAndEpoch;
+import org.junit.jupiter.api.Test;
+
+import java.util.Set;
+
+import static java.util.Collections.emptyList;
+import static java.util.Collections.emptyMap;
+import static java.util.Collections.singleton;
+import static java.util.Collections.singletonList;
+import static org.apache.kafka.common.utils.Utils.mkSet;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class FenceProducersHandlerTest {
+ private final LogContext logContext = new LogContext();
+ private final Node node = new Node(1, "host", 1234);
+
+ @Test
+ public void testBuildRequest() {
+ FenceProducersHandler handler = new FenceProducersHandler(logContext);
+ mkSet("foo", "bar", "baz").forEach(transactionalId -> assertLookup(handler, transactionalId));
+ }
+
+ @Test
+ public void testHandleSuccessfulResponse() {
+ String transactionalId = "foo";
+ CoordinatorKey key = CoordinatorKey.byTransactionalId(transactionalId);
+
+ FenceProducersHandler handler = new FenceProducersHandler(logContext);
+
+ short epoch = 57;
+ long producerId = 7;
+ InitProducerIdResponse response = new InitProducerIdResponse(new InitProducerIdResponseData()
+ .setProducerEpoch(epoch)
+ .setProducerId(producerId));
+
+ ApiResult result = handler.handleSingleResponse(
+ node, key, response);
+
+ assertEquals(emptyList(), result.unmappedKeys);
+ assertEquals(emptyMap(), result.failedKeys);
+ assertEquals(singleton(key), result.completedKeys.keySet());
+
+ ProducerIdAndEpoch expected = new ProducerIdAndEpoch(producerId, epoch);
+ assertEquals(expected, result.completedKeys.get(key));
+ }
+
+ @Test
+ public void testHandleErrorResponse() {
+ String transactionalId = "foo";
+ FenceProducersHandler handler = new FenceProducersHandler(logContext);
+ assertFatalError(handler, transactionalId, Errors.TRANSACTIONAL_ID_AUTHORIZATION_FAILED);
+ assertFatalError(handler, transactionalId, Errors.TRANSACTIONAL_ID_NOT_FOUND);
+ assertFatalError(handler, transactionalId, Errors.UNKNOWN_SERVER_ERROR);
+ assertFatalError(handler, transactionalId, Errors.INVALID_PRODUCER_EPOCH);
+ assertRetriableError(handler, transactionalId, Errors.COORDINATOR_LOAD_IN_PROGRESS);
+ assertUnmappedKey(handler, transactionalId, Errors.NOT_COORDINATOR);
+ assertUnmappedKey(handler, transactionalId, Errors.COORDINATOR_NOT_AVAILABLE);
+ }
+
+ private void assertFatalError(
+ FenceProducersHandler handler,
+ String transactionalId,
+ Errors error
+ ) {
+ CoordinatorKey key = CoordinatorKey.byTransactionalId(transactionalId);
+ ApiResult result = handleResponseError(handler, transactionalId, error);
+ assertEquals(emptyList(), result.unmappedKeys);
+ assertEquals(mkSet(key), result.failedKeys.keySet());
+
+ Throwable throwable = result.failedKeys.get(key);
+ assertTrue(error.exception().getClass().isInstance(throwable));
+ }
+
+ private void assertRetriableError(
+ FenceProducersHandler handler,
+ String transactionalId,
+ Errors error
+ ) {
+ ApiResult result = handleResponseError(handler, transactionalId, error);
+ assertEquals(emptyList(), result.unmappedKeys);
+ assertEquals(emptyMap(), result.failedKeys);
+ }
+
+ private void assertUnmappedKey(
+ FenceProducersHandler handler,
+ String transactionalId,
+ Errors error
+ ) {
+ CoordinatorKey key = CoordinatorKey.byTransactionalId(transactionalId);
+ ApiResult result = handleResponseError(handler, transactionalId, error);
+ assertEquals(emptyMap(), result.failedKeys);
+ assertEquals(singletonList(key), result.unmappedKeys);
+ }
+
+ private ApiResult handleResponseError(
+ FenceProducersHandler handler,
+ String transactionalId,
+ Errors error
+ ) {
+ int brokerId = 1;
+
+ CoordinatorKey key = CoordinatorKey.byTransactionalId(transactionalId);
+ Set keys = mkSet(key);
+
+ InitProducerIdResponse response = new InitProducerIdResponse(new InitProducerIdResponseData()
+ .setErrorCode(error.code()));
+
+ ApiResult result = handler.handleResponse(node, keys, response);
+ assertEquals(emptyMap(), result.completedKeys);
+ return result;
+ }
+
+ private void assertLookup(FenceProducersHandler handler, String transactionalId) {
+ CoordinatorKey key = CoordinatorKey.byTransactionalId(transactionalId);
+ InitProducerIdRequest.Builder request = handler.buildSingleRequest(1, key);
+ assertEquals(transactionalId, request.data.transactionalId());
+ assertEquals(1, request.data.transactionTimeoutMs());
+ }
+}
diff --git a/clients/src/test/java/org/apache/kafka/clients/admin/internals/ListConsumerGroupOffsetsHandlerTest.java b/clients/src/test/java/org/apache/kafka/clients/admin/internals/ListConsumerGroupOffsetsHandlerTest.java
index 9c9bb1e58adb5..27597ce035b00 100644
--- a/clients/src/test/java/org/apache/kafka/clients/admin/internals/ListConsumerGroupOffsetsHandlerTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/admin/internals/ListConsumerGroupOffsetsHandlerTest.java
@@ -56,7 +56,7 @@ public class ListConsumerGroupOffsetsHandlerTest {
@Test
public void testBuildRequest() {
ListConsumerGroupOffsetsHandler handler = new ListConsumerGroupOffsetsHandler(groupId, tps, logContext);
- OffsetFetchRequest request = handler.buildRequest(1, singleton(CoordinatorKey.byGroupId(groupId))).build();
+ OffsetFetchRequest request = handler.buildBatchedRequest(1, singleton(CoordinatorKey.byGroupId(groupId))).build();
assertEquals(groupId, request.data().groups().get(0).groupId());
assertEquals(2, request.data().groups().get(0).topics().size());
assertEquals(2, request.data().groups().get(0).topics().get(0).partitionIndexes().size());
diff --git a/clients/src/test/java/org/apache/kafka/clients/admin/internals/ListTransactionsHandlerTest.java b/clients/src/test/java/org/apache/kafka/clients/admin/internals/ListTransactionsHandlerTest.java
index a8923d1aa2c9c..3c54fad65e8f9 100644
--- a/clients/src/test/java/org/apache/kafka/clients/admin/internals/ListTransactionsHandlerTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/admin/internals/ListTransactionsHandlerTest.java
@@ -52,7 +52,7 @@ public void testBuildRequestWithoutFilters() {
BrokerKey brokerKey = new BrokerKey(OptionalInt.of(brokerId));
ListTransactionsOptions options = new ListTransactionsOptions();
ListTransactionsHandler handler = new ListTransactionsHandler(options, logContext);
- ListTransactionsRequest request = handler.buildRequest(brokerId, singleton(brokerKey)).build();
+ ListTransactionsRequest request = handler.buildBatchedRequest(brokerId, singleton(brokerKey)).build();
assertEquals(Collections.emptyList(), request.data().producerIdFilters());
assertEquals(Collections.emptyList(), request.data().stateFilters());
}
@@ -65,7 +65,7 @@ public void testBuildRequestWithFilteredProducerId() {
ListTransactionsOptions options = new ListTransactionsOptions()
.filterProducerIds(singleton(filteredProducerId));
ListTransactionsHandler handler = new ListTransactionsHandler(options, logContext);
- ListTransactionsRequest request = handler.buildRequest(brokerId, singleton(brokerKey)).build();
+ ListTransactionsRequest request = handler.buildBatchedRequest(brokerId, singleton(brokerKey)).build();
assertEquals(Collections.singletonList(filteredProducerId), request.data().producerIdFilters());
assertEquals(Collections.emptyList(), request.data().stateFilters());
}
@@ -78,7 +78,7 @@ public void testBuildRequestWithFilteredState() {
ListTransactionsOptions options = new ListTransactionsOptions()
.filterStates(singleton(filteredState));
ListTransactionsHandler handler = new ListTransactionsHandler(options, logContext);
- ListTransactionsRequest request = handler.buildRequest(brokerId, singleton(brokerKey)).build();
+ ListTransactionsRequest request = handler.buildBatchedRequest(brokerId, singleton(brokerKey)).build();
assertEquals(Collections.singletonList(filteredState.toString()), request.data().stateFilters());
assertEquals(Collections.emptyList(), request.data().producerIdFilters());
}
diff --git a/clients/src/test/java/org/apache/kafka/clients/admin/internals/RemoveMembersFromConsumerGroupHandlerTest.java b/clients/src/test/java/org/apache/kafka/clients/admin/internals/RemoveMembersFromConsumerGroupHandlerTest.java
index 6f5dfda5bc307..3ecd1f10ee2f6 100644
--- a/clients/src/test/java/org/apache/kafka/clients/admin/internals/RemoveMembersFromConsumerGroupHandlerTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/admin/internals/RemoveMembersFromConsumerGroupHandlerTest.java
@@ -55,7 +55,7 @@ public class RemoveMembersFromConsumerGroupHandlerTest {
@Test
public void testBuildRequest() {
RemoveMembersFromConsumerGroupHandler handler = new RemoveMembersFromConsumerGroupHandler(groupId, members, logContext);
- LeaveGroupRequest request = handler.buildRequest(1, singleton(CoordinatorKey.byGroupId(groupId))).build();
+ LeaveGroupRequest request = handler.buildBatchedRequest(1, singleton(CoordinatorKey.byGroupId(groupId))).build();
assertEquals(groupId, request.data().groupId());
assertEquals(2, request.data().members().size());
}
diff --git a/connect/api/src/main/java/org/apache/kafka/connect/source/ConnectorTransactionBoundaries.java b/connect/api/src/main/java/org/apache/kafka/connect/source/ConnectorTransactionBoundaries.java
new file mode 100644
index 0000000000000..73746ba0993f3
--- /dev/null
+++ b/connect/api/src/main/java/org/apache/kafka/connect/source/ConnectorTransactionBoundaries.java
@@ -0,0 +1,31 @@
+/*
+ * 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.source;
+
+/**
+ * An enum to represent the level of support for connector-defined transaction boundaries.
+ */
+public enum ConnectorTransactionBoundaries {
+ /**
+ * Signals that a connector can define its own transaction boundaries.
+ */
+ SUPPORTED,
+ /**
+ * Signals that a connector cannot define its own transaction boundaries.
+ */
+ UNSUPPORTED
+}
diff --git a/connect/api/src/main/java/org/apache/kafka/connect/source/ExactlyOnceSupport.java b/connect/api/src/main/java/org/apache/kafka/connect/source/ExactlyOnceSupport.java
new file mode 100644
index 0000000000000..3980410e4b538
--- /dev/null
+++ b/connect/api/src/main/java/org/apache/kafka/connect/source/ExactlyOnceSupport.java
@@ -0,0 +1,31 @@
+/*
+ * 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.source;
+
+/**
+ * An enum to represent the level of support for exactly-once delivery from a source connector.
+ */
+public enum ExactlyOnceSupport {
+ /**
+ * Signals that a connector supports exactly-once delivery.
+ */
+ SUPPORTED,
+ /**
+ * Signals that a connector does not support exactly-once delivery.
+ */
+ UNSUPPORTED;
+}
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 6e9694024d334..206abfad32034 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
@@ -18,6 +18,8 @@
import org.apache.kafka.connect.connector.Connector;
+import java.util.Map;
+
/**
* SourceConnectors implement the connector interface to pull data from another system and send
* it to Kafka.
@@ -28,4 +30,46 @@ public abstract class SourceConnector extends Connector {
protected SourceConnectorContext context() {
return (SourceConnectorContext) context;
}
+
+ /**
+ * Signals whether the connector supports exactly-once delivery guarantees with a proposed configuration.
+ * Developers can assume that worker-level exactly-once support is enabled when this method is invoked.
+ *
+ * For backwards compatibility, the default implementation will return {@code null}, but connector developers are
+ * strongly encouraged to override this method to return a non-null value such as
+ * {@link ExactlyOnceSupport#SUPPORTED SUPPORTED} or {@link ExactlyOnceSupport#UNSUPPORTED UNSUPPORTED}.
+ *
+ *
Similar to {@link #validate(Map) validate}, this method may be called by the runtime before the
+ * {@link #start(Map) start} method is invoked when the connector will be run with exactly-once support.
+ *
+ * @param connectorConfig the configuration that will be used for the connector.
+ * @return {@link ExactlyOnceSupport#SUPPORTED} if the connector can provide exactly-once support with the given
+ * configuration, and {@link ExactlyOnceSupport#UNSUPPORTED} if it cannot. If this method is overridden by a
+ * connector, should not be {@code null}, but if {@code null}, it will be assumed that the connector cannot provide
+ * exactly-once guarantees.
+ * @since 3.2
+ */
+ public ExactlyOnceSupport exactlyOnceSupport(Map connectorConfig) {
+ return null;
+ }
+
+ /**
+ * Signals whether the connector implementation is capable of defining the transaction boundaries for a
+ * connector with the given configuration. This method is called before {@link #start(Map)}, only when the
+ * runtime supports exactly-once and the connector configuration includes {@code transaction.boundary=connector}.
+ *
+ * This method need not be implemented if the connector implementation does not support defining
+ * transaction boundaries.
+ *
+ * @param connectorConfig the configuration that will be used for the connector
+ * @return {@link ConnectorTransactionBoundaries#SUPPORTED} if the connector will define its own transaction boundaries,
+ * or {@link ConnectorTransactionBoundaries#UNSUPPORTED} otherwise. If this method is overridden by a
+ * connector, should not be {@code null}, but if {@code null}, it will be assumed that the connector cannot define its own
+ * transaction boundaries.
+ * @since 3.2
+ * @see TransactionContext
+ */
+ public ConnectorTransactionBoundaries canDefineTransactionBoundaries(Map connectorConfig) {
+ return ConnectorTransactionBoundaries.UNSUPPORTED;
+ }
}
diff --git a/connect/api/src/main/java/org/apache/kafka/connect/source/SourceTask.java b/connect/api/src/main/java/org/apache/kafka/connect/source/SourceTask.java
index f5209e1ccab64..2159e68e8a88a 100644
--- a/connect/api/src/main/java/org/apache/kafka/connect/source/SourceTask.java
+++ b/connect/api/src/main/java/org/apache/kafka/connect/source/SourceTask.java
@@ -16,10 +16,11 @@
*/
package org.apache.kafka.connect.source;
-import org.apache.kafka.connect.connector.Task;
import org.apache.kafka.clients.producer.RecordMetadata;
+import org.apache.kafka.connect.connector.Task;
import java.util.List;
+import java.util.Locale;
import java.util.Map;
/**
@@ -27,6 +28,51 @@
*/
public abstract class SourceTask implements Task {
+ /**
+ * The configuration key that determines how source tasks will define transaction boundaries
+ * when exactly-once support is enabled.
+ */
+ public static final String TRANSACTION_BOUNDARY_CONFIG = "transaction.boundary";
+
+ /**
+ * Represents the permitted values for the {@link #TRANSACTION_BOUNDARY_CONFIG} property.
+ */
+ public enum TransactionBoundary {
+ /**
+ * A new transaction will be started and committed for every batch of records returned by {@link #poll()}.
+ */
+ POLL,
+ /**
+ * Transactions will be started and committed on a user-defined time interval.
+ */
+ INTERVAL,
+ /**
+ * Transactions will be defined by the connector itself, via a {@link TransactionContext}.
+ */
+ CONNECTOR;
+
+ /**
+ * The default transaction boundary style that will be used for source connectors when no style is explicitly
+ * configured.
+ */
+ public static final TransactionBoundary DEFAULT = POLL;
+
+ /**
+ * Parse a {@link TransactionBoundary} from the given string.
+ * @param property the string to parse; should not be null
+ * @return the {@link TransactionBoundary} whose name matches the given string
+ * @throws IllegalArgumentException if there is no transaction boundary type with the given name
+ */
+ public static TransactionBoundary fromProperty(String property) {
+ return TransactionBoundary.valueOf(property.toUpperCase(Locale.ROOT).trim());
+ }
+
+ @Override
+ public String toString() {
+ return name().toLowerCase(Locale.ROOT);
+ }
+ }
+
protected SourceTaskContext context;
/**
@@ -44,16 +90,13 @@ public void initialize(SourceTaskContext context) {
public abstract void start(Map props);
/**
- *
* Poll this source task for new records. If no data is currently available, this method
* should block but return control to the caller regularly (by returning {@code null}) in
* order for the task to transition to the {@code PAUSED} state if requested to do so.
- *
*
* The task will be {@link #stop() stopped} on a separate thread, and when that happens
* this method is expected to unblock, quickly finish up any remaining processing, and
* return.
- *
*
* @return a list of source records
*/
@@ -63,12 +106,10 @@ public void initialize(SourceTaskContext context) {
*
* Commit the offsets, up to the offsets that have been returned by {@link #poll()}. This
* method should block until the commit is complete.
- *
*
* SourceTasks are not required to implement this functionality; Kafka Connect will record offsets
* automatically. This hook is provided for systems that also need to store offsets internally
* in their own system.
- *
*/
public void commit() throws InterruptedException {
// This space intentionally left blank.
@@ -91,17 +132,14 @@ public void commit() throws InterruptedException {
*
* Commit an individual {@link SourceRecord} when the callback from the producer client is received. This method is
* also called when a record is filtered by a transformation, and thus will never be ACK'd by a broker.
- *
*
* This is an alias for {@link #commitRecord(SourceRecord, RecordMetadata)} for backwards compatibility. The default
* implementation of {@link #commitRecord(SourceRecord, RecordMetadata)} just calls this method. It is not necessary
* to override both methods.
- *
*
* SourceTasks are not required to implement this functionality; Kafka Connect will record offsets
* automatically. This hook is provided for systems that also need to store offsets internally
* in their own system.
- *
*
* @param record {@link SourceRecord} that was successfully sent via the producer or filtered by a transformation
* @throws InterruptedException
@@ -115,19 +153,16 @@ public void commitRecord(SourceRecord record) throws InterruptedException {
/**
*
* Commit an individual {@link SourceRecord} when the callback from the producer client is received. This method is
- * also called when a record is filtered by a transformation or when {@link ConnectorConfig} "errors.tolerance" is set to "all"
+ * also called when a record is filtered by a transformation or when "errors.tolerance" is set to "all"
* and thus will never be ACK'd by a broker.
* In both cases {@code metadata} will be null.
- *
*
* SourceTasks are not required to implement this functionality; Kafka Connect will record offsets
* automatically. This hook is provided for systems that also need to store offsets internally
* in their own system.
- *
*
* The default implementation just calls {@link #commitRecord(SourceRecord)}, which is a nop by default. It is
* not necessary to implement both methods.
- *
*
* @param record {@link SourceRecord} that was successfully sent via the producer, filtered by a transformation, or dropped on producer exception
* @param metadata {@link RecordMetadata} record metadata returned from the broker, or null if the record was filtered or if producer exceptions are ignored
diff --git a/connect/api/src/main/java/org/apache/kafka/connect/source/SourceTaskContext.java b/connect/api/src/main/java/org/apache/kafka/connect/source/SourceTaskContext.java
index ddb0a78718351..5bd7889cb2aa5 100644
--- a/connect/api/src/main/java/org/apache/kafka/connect/source/SourceTaskContext.java
+++ b/connect/api/src/main/java/org/apache/kafka/connect/source/SourceTaskContext.java
@@ -38,4 +38,29 @@ public interface SourceTaskContext {
* Get the OffsetStorageReader for this SourceTask.
*/
OffsetStorageReader offsetStorageReader();
+
+ /**
+ * Get a {@link TransactionContext} that can be used to define producer transaction boundaries
+ * when exactly-once support is enabled for the connector.
+ *
+ * This method was added in Apache Kafka 3.2. Source tasks that use this method but want to
+ * maintain backward compatibility so they can also be deployed to older Connect runtimes
+ * should guard the call to this method with a try-catch block, since calling this method will result in a
+ * {@link NoSuchMethodException} or {@link NoClassDefFoundError} when the source connector is deployed to
+ * Connect runtimes older than Kafka 3.2. For example:
+ *
+ * TransactionContext transactionContext;
+ * try {
+ * transactionContext = context.transactionContext();
+ * } catch (NoSuchMethodError | NoClassDefFoundError e) {
+ * transactionContext = null;
+ * }
+ *
+ *
+ * @return the transaction context, or null if the connector was not configured to specify transaction boundaries
+ * @since 3.2
+ */
+ default TransactionContext transactionContext() {
+ return null;
+ }
}
diff --git a/connect/api/src/main/java/org/apache/kafka/connect/source/TransactionContext.java b/connect/api/src/main/java/org/apache/kafka/connect/source/TransactionContext.java
new file mode 100644
index 0000000000000..d8aac108be180
--- /dev/null
+++ b/connect/api/src/main/java/org/apache/kafka/connect/source/TransactionContext.java
@@ -0,0 +1,55 @@
+/*
+ * 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.source;
+
+/**
+ * Provided to source tasks to allow them to define their own producer transaction boundaries when
+ * exactly-once support is enabled.
+ */
+public interface TransactionContext {
+
+ /**
+ * Request a transaction commit after the next batch of records from {@link SourceTask#poll()}
+ * is processed.
+ */
+ void commitTransaction();
+
+ /**
+ * Request a transaction commit after a source record is processed. The source record will be the
+ * last record in the committed transaction.
+ * @param record the record to commit the transaction after; may not be null.
+ */
+ void commitTransaction(SourceRecord record);
+
+ /**
+ * Requests a transaction abort the next batch of records from {@link SourceTask#poll()}. All of
+ * the records in that transaction will be discarded and will not appear in a committed transaction.
+ * However, offsets for that transaction will still be committed. If the data should be reprocessed,
+ * the task should not invoke this method and should instead throw an exception.
+ */
+ void abortTransaction();
+
+ /**
+ * Requests a transaction abort after a source record is processed. The source record will be the
+ * last record in the aborted transaction. All of the records in that transaction will be discarded
+ * and will not appear in a committed transaction. However, offsets for that transaction will still
+ * be committed. If the data should be reprocessed, the task should not invoke this method and
+ * should instead throw an exception.
+ * @param record the record to abort the transaction after; may not be null.
+ */
+ void abortTransaction(SourceRecord record);
+}
diff --git a/connect/basic-auth-extension/src/main/java/org/apache/kafka/connect/rest/basic/auth/extension/JaasBasicAuthFilter.java b/connect/basic-auth-extension/src/main/java/org/apache/kafka/connect/rest/basic/auth/extension/JaasBasicAuthFilter.java
index 0299cbba0b546..d582a0ce9b125 100644
--- a/connect/basic-auth-extension/src/main/java/org/apache/kafka/connect/rest/basic/auth/extension/JaasBasicAuthFilter.java
+++ b/connect/basic-auth-extension/src/main/java/org/apache/kafka/connect/rest/basic/auth/extension/JaasBasicAuthFilter.java
@@ -18,7 +18,11 @@
package org.apache.kafka.connect.rest.basic.auth.extension;
import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashSet;
import java.util.List;
+import java.util.Set;
+import java.util.function.Predicate;
import java.util.regex.Pattern;
import javax.security.auth.login.Configuration;
import javax.ws.rs.HttpMethod;
@@ -45,7 +49,10 @@
public class JaasBasicAuthFilter implements ContainerRequestFilter {
private static final Logger log = LoggerFactory.getLogger(JaasBasicAuthFilter.class);
- private static final Pattern TASK_REQUEST_PATTERN = Pattern.compile("/?connectors/([^/]+)/tasks/?");
+ private static final Set INTERNAL_REQUEST_MATCHERS = new HashSet<>(Arrays.asList(
+ new RequestMatcher("/?connectors/([^/]+)/tasks/?", HttpMethod.POST),
+ new RequestMatcher("/?connectors/[^/]+/fence/?", HttpMethod.PUT)
+ ));
private static final String CONNECT_LOGIN_MODULE = "KafkaConnect";
static final String AUTHORIZATION = "Authorization";
@@ -53,13 +60,29 @@ public class JaasBasicAuthFilter implements ContainerRequestFilter {
// Package-private for testing
final Configuration configuration;
+ private static class RequestMatcher implements Predicate {
+ private final Pattern path;
+ private final String method;
+
+ public RequestMatcher(String path, String method) {
+ this.path = Pattern.compile(path);
+ this.method = method;
+ }
+
+ @Override
+ public boolean test(ContainerRequestContext requestContext) {
+ return requestContext.getMethod().equals(method)
+ && path.matcher(requestContext.getUriInfo().getPath()).matches();
+ }
+ }
+
public JaasBasicAuthFilter(Configuration configuration) {
this.configuration = configuration;
}
@Override
public void filter(ContainerRequestContext requestContext) throws IOException {
- if (isInternalTaskConfigRequest(requestContext)) {
+ if (isInternalRequest(requestContext)) {
log.trace("Skipping authentication for internal request");
return;
}
@@ -82,12 +105,10 @@ public void filter(ContainerRequestContext requestContext) throws IOException {
}
}
- private static boolean isInternalTaskConfigRequest(ContainerRequestContext requestContext) {
- return requestContext.getMethod().equals(HttpMethod.POST)
- && TASK_REQUEST_PATTERN.matcher(requestContext.getUriInfo().getPath()).matches();
+ private boolean isInternalRequest(ContainerRequestContext requestContext) {
+ return INTERNAL_REQUEST_MATCHERS.stream().anyMatch(m -> m.test(requestContext));
}
-
public static class BasicAuthCallBackHandler implements CallbackHandler {
private static final String BASIC = "basic";
diff --git a/connect/basic-auth-extension/src/test/java/org/apache/kafka/connect/rest/basic/auth/extension/JaasBasicAuthFilterTest.java b/connect/basic-auth-extension/src/test/java/org/apache/kafka/connect/rest/basic/auth/extension/JaasBasicAuthFilterTest.java
index 561095f68218a..599ca0baa265c 100644
--- a/connect/basic-auth-extension/src/test/java/org/apache/kafka/connect/rest/basic/auth/extension/JaasBasicAuthFilterTest.java
+++ b/connect/basic-auth-extension/src/test/java/org/apache/kafka/connect/rest/basic/auth/extension/JaasBasicAuthFilterTest.java
@@ -42,6 +42,7 @@
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.atLeastOnce;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@@ -58,7 +59,7 @@ public void testSuccess() throws IOException {
ContainerRequestContext requestContext = setMock("Basic", "user", "password");
jaasBasicAuthFilter.filter(requestContext);
- verify(requestContext).getMethod();
+ verify(requestContext, atLeastOnce()).getMethod();
verify(requestContext).getHeaderString(JaasBasicAuthFilter.AUTHORIZATION);
}
@@ -69,7 +70,7 @@ public void testEmptyCredentialsFile() throws IOException {
ContainerRequestContext requestContext = setMock("Basic", "user", "password");
jaasBasicAuthFilter.filter(requestContext);
- verify(requestContext).getMethod();
+ verify(requestContext, atLeastOnce()).getMethod();
verify(requestContext).getHeaderString(JaasBasicAuthFilter.AUTHORIZATION);
}
@@ -81,7 +82,7 @@ public void testBadCredential() throws IOException {
jaasBasicAuthFilter.filter(requestContext);
verify(requestContext).abortWith(any(Response.class));
- verify(requestContext).getMethod();
+ verify(requestContext, atLeastOnce()).getMethod();
verify(requestContext).getHeaderString(JaasBasicAuthFilter.AUTHORIZATION);
}
@@ -93,7 +94,7 @@ public void testBadPassword() throws IOException {
jaasBasicAuthFilter.filter(requestContext);
verify(requestContext).abortWith(any(Response.class));
- verify(requestContext).getMethod();
+ verify(requestContext, atLeastOnce()).getMethod();
verify(requestContext).getHeaderString(JaasBasicAuthFilter.AUTHORIZATION);
}
@@ -105,7 +106,7 @@ public void testUnknownBearer() throws IOException {
jaasBasicAuthFilter.filter(requestContext);
verify(requestContext).abortWith(any(Response.class));
- verify(requestContext).getMethod();
+ verify(requestContext, atLeastOnce()).getMethod();
verify(requestContext).getHeaderString(JaasBasicAuthFilter.AUTHORIZATION);
}
@@ -117,7 +118,7 @@ public void testUnknownLoginModule() throws IOException {
jaasBasicAuthFilter.filter(requestContext);
verify(requestContext).abortWith(any(Response.class));
- verify(requestContext).getMethod();
+ verify(requestContext, atLeastOnce()).getMethod();
verify(requestContext).getHeaderString(JaasBasicAuthFilter.AUTHORIZATION);
}
@@ -128,7 +129,7 @@ public void testUnknownCredentialsFile() throws IOException {
jaasBasicAuthFilter.filter(requestContext);
verify(requestContext).abortWith(any(Response.class));
- verify(requestContext).getMethod();
+ verify(requestContext, atLeastOnce()).getMethod();
verify(requestContext).getHeaderString(JaasBasicAuthFilter.AUTHORIZATION);
}
@@ -139,14 +140,23 @@ public void testNoFileOption() throws IOException {
jaasBasicAuthFilter.filter(requestContext);
verify(requestContext).abortWith(any(Response.class));
- verify(requestContext).getMethod();
+ verify(requestContext, atLeastOnce()).getMethod();
verify(requestContext).getHeaderString(JaasBasicAuthFilter.AUTHORIZATION);
}
@Test
- public void testPostWithoutAppropriateCredential() throws IOException {
+ public void testInternalTaskConfigEndpointSkipped() throws IOException {
+ testInternalEndpointSkipped("connectors/connName/tasks");
+ }
+
+ @Test
+ public void testInternalZombieFencingEndpointSkipped() throws IOException {
+ testInternalEndpointSkipped("connectors/connName/fence");
+ }
+
+ private void testInternalEndpointSkipped(String endpoint) throws IOException {
UriInfo uriInfo = mock(UriInfo.class);
- when(uriInfo.getPath()).thenReturn("connectors/connName/tasks");
+ when(uriInfo.getPath()).thenReturn(endpoint);
ContainerRequestContext requestContext = mock(ContainerRequestContext.class);
when(requestContext.getMethod()).thenReturn(HttpMethod.POST);
@@ -158,7 +168,7 @@ public void testPostWithoutAppropriateCredential() throws IOException {
jaasBasicAuthFilter.filter(requestContext);
verify(uriInfo).getPath();
- verify(requestContext).getMethod();
+ verify(requestContext, atLeastOnce()).getMethod();
verify(requestContext).getUriInfo();
}
diff --git a/connect/mirror/src/test/java/org/apache/kafka/connect/mirror/integration/MirrorConnectorsIntegrationBaseTest.java b/connect/mirror/src/test/java/org/apache/kafka/connect/mirror/integration/MirrorConnectorsIntegrationBaseTest.java
index 8f692ca911612..2b6b045ae8f6e 100644
--- a/connect/mirror/src/test/java/org/apache/kafka/connect/mirror/integration/MirrorConnectorsIntegrationBaseTest.java
+++ b/connect/mirror/src/test/java/org/apache/kafka/connect/mirror/integration/MirrorConnectorsIntegrationBaseTest.java
@@ -26,6 +26,8 @@
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.config.ConfigResource;
import org.apache.kafka.common.config.TopicConfig;
+import org.apache.kafka.common.network.ListenerName;
+import org.apache.kafka.common.security.auth.SecurityProtocol;
import org.apache.kafka.common.utils.Exit;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.connect.connector.Connector;
@@ -112,7 +114,9 @@ public abstract class MirrorConnectorsIntegrationBaseTest {
protected Properties backupBrokerProps = new Properties();
protected Map primaryWorkerProps = new HashMap<>();
protected Map backupWorkerProps = new HashMap<>();
-
+ protected ListenerName primaryBrokerListener = ListenerName.forSecurityProtocol(SecurityProtocol.PLAINTEXT);
+ protected ListenerName backupBrokerListener = ListenerName.forSecurityProtocol(SecurityProtocol.PLAINTEXT);
+
@BeforeEach
public void startClusters() throws Exception {
startClusters(new HashMap() {{
@@ -171,6 +175,7 @@ public void startClusters(Map additionalMM2Config) throws Except
.brokerProps(primaryBrokerProps)
.workerProps(primaryWorkerProps)
.maskExitProcedures(false)
+ .brokerListener(primaryBrokerListener)
.build();
backup = new EmbeddedConnectCluster.Builder()
@@ -180,6 +185,7 @@ public void startClusters(Map additionalMM2Config) throws Except
.brokerProps(backupBrokerProps)
.workerProps(backupWorkerProps)
.maskExitProcedures(false)
+ .brokerListener(backupBrokerListener)
.build();
primary.start();
@@ -718,9 +724,12 @@ private void createTopics() {
adminClientConfig.put(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG, REQUEST_TIMEOUT_DURATION_MS);
// create these topics before starting the connectors so we don't need to wait for discovery
+ adminClientConfig.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, primary.kafka().bootstrapServers(primaryBrokerListener));
primary.kafka().createTopic("test-topic-1", NUM_PARTITIONS, 1, topicConfig, adminClientConfig);
primary.kafka().createTopic("backup.test-topic-1", 1, 1, emptyMap, adminClientConfig);
primary.kafka().createTopic("heartbeats", 1, 1, emptyMap, adminClientConfig);
+
+ adminClientConfig.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, backup.kafka().bootstrapServers(backupBrokerListener));
backup.kafka().createTopic("test-topic-1", NUM_PARTITIONS, 1, emptyMap, adminClientConfig);
backup.kafka().createTopic("primary.test-topic-1", 1, 1, emptyMap, adminClientConfig);
backup.kafka().createTopic("heartbeats", 1, 1, emptyMap, adminClientConfig);
diff --git a/connect/mirror/src/test/java/org/apache/kafka/connect/mirror/integration/MirrorConnectorsIntegrationSSLTest.java b/connect/mirror/src/test/java/org/apache/kafka/connect/mirror/integration/MirrorConnectorsIntegrationSSLTest.java
index eb2af4842b60a..b1b81f9b923cb 100644
--- a/connect/mirror/src/test/java/org/apache/kafka/connect/mirror/integration/MirrorConnectorsIntegrationSSLTest.java
+++ b/connect/mirror/src/test/java/org/apache/kafka/connect/mirror/integration/MirrorConnectorsIntegrationSSLTest.java
@@ -24,7 +24,9 @@
import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.common.config.SslConfigs;
import org.apache.kafka.common.config.types.Password;
+import org.apache.kafka.common.network.ListenerName;
import org.apache.kafka.common.network.Mode;
+import org.apache.kafka.common.security.auth.SecurityProtocol;
import org.apache.kafka.test.TestSslUtils;
import org.apache.kafka.test.TestUtils;
@@ -59,6 +61,8 @@ public void startClusters() throws Exception {
// set SSL config for producer used by source task in MM2
mm2Props.putAll(sslProps.entrySet().stream().collect(Collectors.toMap(
e -> BACKUP_CLUSTER_ALIAS + ".producer." + e.getKey(), e -> String.valueOf(e.getValue()))));
+
+ backupBrokerListener = ListenerName.forSecurityProtocol(SecurityProtocol.SSL);
super.startClusters();
}
diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/cli/ConnectDistributed.java b/connect/runtime/src/main/java/org/apache/kafka/connect/cli/ConnectDistributed.java
index 8d93e795911b9..ffd17e9195fba 100644
--- a/connect/runtime/src/main/java/org/apache/kafka/connect/cli/ConnectDistributed.java
+++ b/connect/runtime/src/main/java/org/apache/kafka/connect/cli/ConnectDistributed.java
@@ -103,7 +103,7 @@ public Connect startConnect(Map workerProps) {
URI advertisedUrl = rest.advertisedUrl();
String workerId = advertisedUrl.getHost() + ":" + advertisedUrl.getPort();
- // Create the admin client to be shared by all backing stores.
+ // Create the admin client to be shared by all worker-global backing stores.
Map adminProps = new HashMap<>(config.originals());
ConnectUtils.addMetricsContextProperties(adminProps, config, kafkaClusterId);
SharedTopicAdmin sharedAdmin = new SharedTopicAdmin(adminProps);
diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/AbstractHerder.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/AbstractHerder.java
index 89020a4d1741a..baf25ea23bd6f 100644
--- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/AbstractHerder.java
+++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/AbstractHerder.java
@@ -16,6 +16,8 @@
*/
package org.apache.kafka.connect.runtime;
+import org.apache.kafka.clients.admin.AdminClientConfig;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.config.AbstractConfig;
import org.apache.kafka.common.config.Config;
@@ -28,7 +30,8 @@
import org.apache.kafka.connect.connector.policy.ConnectorClientConfigOverridePolicy;
import org.apache.kafka.connect.connector.policy.ConnectorClientConfigRequest;
import org.apache.kafka.connect.errors.NotFoundException;
-import org.apache.kafka.connect.runtime.distributed.ClusterConfigState;
+import org.apache.kafka.connect.sink.SinkConnector;
+import org.apache.kafka.connect.storage.ClusterConfigState;
import org.apache.kafka.connect.runtime.isolation.Plugins;
import org.apache.kafka.connect.runtime.rest.entities.ActiveTopicsInfo;
import org.apache.kafka.connect.runtime.rest.entities.ConfigInfo;
@@ -344,9 +347,11 @@ public ConnectorStateInfo.TaskState taskStatus(ConnectorTaskId id) {
status.workerId(), status.trace());
}
- protected Map validateBasicConnectorConfig(Connector connector,
- ConfigDef configDef,
- Map config) {
+ protected Map validateSinkConnectorConfig(SinkConnector connector, ConfigDef configDef, Map config) {
+ return configDef.validateAll(config);
+ }
+
+ protected Map validateSourceConnectorConfig(SourceConnector connector, ConfigDef configDef, Map config) {
return configDef.validateAll(config);
}
@@ -412,7 +417,23 @@ public Optional buildRestartPlan(RestartRequest request) {
conf == null ? ConnectorType.UNKNOWN : connectorTypeForClass(conf.get(ConnectorConfig.CONNECTOR_CLASS_CONFIG))
);
return Optional.of(new RestartPlan(request, stateInfo));
+ }
+
+ protected boolean connectorUsesConsumer(org.apache.kafka.connect.health.ConnectorType connectorType, Map connProps) {
+ return connectorType == org.apache.kafka.connect.health.ConnectorType.SINK;
+ }
+ protected boolean connectorUsesAdmin(org.apache.kafka.connect.health.ConnectorType connectorType, Map connProps) {
+ if (connectorType == org.apache.kafka.connect.health.ConnectorType.SOURCE) {
+ return SourceConnectorConfig.usesTopicCreation(connProps);
+ } else {
+ return SinkConnectorConfig.hasDlqTopicConfig(connProps);
+ }
+ }
+
+ protected boolean connectorUsesProducer(org.apache.kafka.connect.health.ConnectorType connectorType, Map connProps) {
+ return connectorType == org.apache.kafka.connect.health.ConnectorType.SOURCE
+ || SinkConnectorConfig.hasDlqTopicConfig(connProps);
}
ConfigInfos validateConnectorConfig(Map connectorProps, boolean doLog) {
@@ -426,22 +447,20 @@ ConfigInfos validateConnectorConfig(Map connectorProps, boolean
Connector connector = getConnector(connType);
org.apache.kafka.connect.health.ConnectorType connectorType;
ClassLoader savedLoader = plugins().compareAndSwapLoaders(connector);
+ ConfigDef enrichedConfigDef;
+ Map validatedConnectorConfig;
try {
- ConfigDef baseConfigDef;
if (connector instanceof SourceConnector) {
- baseConfigDef = SourceConnectorConfig.configDef();
connectorType = org.apache.kafka.connect.health.ConnectorType.SOURCE;
+ enrichedConfigDef = ConnectorConfig.enrich(plugins(), SourceConnectorConfig.configDef(), connectorProps, false);
+ validatedConnectorConfig = validateSourceConnectorConfig((SourceConnector) connector, enrichedConfigDef, connectorProps);
} else {
- baseConfigDef = SinkConnectorConfig.configDef();
SinkConnectorConfig.validate(connectorProps);
connectorType = org.apache.kafka.connect.health.ConnectorType.SINK;
+ enrichedConfigDef = ConnectorConfig.enrich(plugins(), SinkConnectorConfig.configDef(), connectorProps, false);
+ validatedConnectorConfig = validateSinkConnectorConfig((SinkConnector) connector, enrichedConfigDef, connectorProps);
}
- ConfigDef enrichedConfigDef = ConnectorConfig.enrich(plugins(), baseConfigDef, connectorProps, false);
- Map validatedConnectorConfig = validateBasicConnectorConfig(
- connector,
- enrichedConfigDef,
- connectorProps
- );
+
connectorProps.entrySet().stream()
.filter(e -> e.getValue() == null)
.map(Map.Entry::getKey)
@@ -449,6 +468,7 @@ ConfigInfos validateConnectorConfig(Map connectorProps, boolean
validatedConnectorConfig.computeIfAbsent(prop, ConfigValue::new)
.addErrorMessage("Null value can not be supplied as the configuration value.")
);
+
List configValues = new ArrayList<>(validatedConnectorConfig.values());
Map configKeys = new LinkedHashMap<>(enrichedConfigDef.configKeys());
Set allGroups = new LinkedHashSet<>(enrichedConfigDef.groups());
@@ -482,38 +502,39 @@ ConfigInfos validateConnectorConfig(Map connectorProps, boolean
ConfigInfos producerConfigInfos = null;
ConfigInfos consumerConfigInfos = null;
ConfigInfos adminConfigInfos = null;
- if (connectorType.equals(org.apache.kafka.connect.health.ConnectorType.SOURCE)) {
- producerConfigInfos = validateClientOverrides(connName,
- ConnectorConfig.CONNECTOR_CLIENT_PRODUCER_OVERRIDES_PREFIX,
- connectorConfig,
- ProducerConfig.configDef(),
- connector.getClass(),
- connectorType,
- ConnectorClientConfigRequest.ClientType.PRODUCER,
- connectorClientConfigOverridePolicy);
- return mergeConfigInfos(connType, configInfos, producerConfigInfos);
- } else {
- consumerConfigInfos = validateClientOverrides(connName,
- ConnectorConfig.CONNECTOR_CLIENT_CONSUMER_OVERRIDES_PREFIX,
- connectorConfig,
- ProducerConfig.configDef(),
- connector.getClass(),
- connectorType,
- ConnectorClientConfigRequest.ClientType.CONSUMER,
- connectorClientConfigOverridePolicy);
- // check if topic for dead letter queue exists
- String topic = connectorProps.get(SinkConnectorConfig.DLQ_TOPIC_NAME_CONFIG);
- if (topic != null && !topic.isEmpty()) {
- adminConfigInfos = validateClientOverrides(connName,
- ConnectorConfig.CONNECTOR_CLIENT_ADMIN_OVERRIDES_PREFIX,
- connectorConfig,
- ProducerConfig.configDef(),
- connector.getClass(),
- connectorType,
- ConnectorClientConfigRequest.ClientType.ADMIN,
- connectorClientConfigOverridePolicy);
- }
+ if (connectorUsesProducer(connectorType, connectorProps)) {
+ producerConfigInfos = validateClientOverrides(
+ connName,
+ ConnectorConfig.CONNECTOR_CLIENT_PRODUCER_OVERRIDES_PREFIX,
+ connectorConfig,
+ ProducerConfig.configDef(),
+ connector.getClass(),
+ connectorType,
+ ConnectorClientConfigRequest.ClientType.PRODUCER,
+ connectorClientConfigOverridePolicy);
+ }
+ if (connectorUsesAdmin(connectorType, connectorProps)) {
+ adminConfigInfos = validateClientOverrides(
+ connName,
+ ConnectorConfig.CONNECTOR_CLIENT_ADMIN_OVERRIDES_PREFIX,
+ connectorConfig,
+ AdminClientConfig.configDef(),
+ connector.getClass(),
+ connectorType,
+ ConnectorClientConfigRequest.ClientType.ADMIN,
+ connectorClientConfigOverridePolicy);
+ }
+ if (connectorUsesConsumer(connectorType, connectorProps)) {
+ consumerConfigInfos = validateClientOverrides(
+ connName,
+ ConnectorConfig.CONNECTOR_CLIENT_CONSUMER_OVERRIDES_PREFIX,
+ connectorConfig,
+ ConsumerConfig.configDef(),
+ connector.getClass(),
+ connectorType,
+ ConnectorClientConfigRequest.ClientType.CONSUMER,
+ connectorClientConfigOverridePolicy);
}
return mergeConfigInfos(connType, configInfos, producerConfigInfos, consumerConfigInfos, adminConfigInfos);
} finally {
@@ -660,7 +681,7 @@ protected Connector getConnector(String connType) {
return tempConnectors.computeIfAbsent(connType, k -> plugins().newConnector(k));
}
- /*
+ /**
* Retrieves ConnectorType for the corresponding connector class
* @param connClass class of the connector
*/
@@ -668,6 +689,15 @@ public ConnectorType connectorTypeForClass(String connClass) {
return ConnectorType.from(getConnector(connClass).getClass());
}
+ /**
+ * Retrieves ConnectorType for the class specified in the connector config
+ * @param connConfig the connector config; may not be null
+ * @return the {@link ConnectorType} of the connector
+ */
+ public ConnectorType connectorTypeForConfig(Map connConfig) {
+ return connectorTypeForClass(connConfig.get(ConnectorConfig.CONNECTOR_CLASS_CONFIG));
+ }
+
/**
* Checks a given {@link ConfigInfos} for validation error messages and adds an exception
* to the given {@link Callback} if any were found.
diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/AbstractWorkerSourceTask.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/AbstractWorkerSourceTask.java
new file mode 100644
index 0000000000000..a3327464a4d84
--- /dev/null
+++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/AbstractWorkerSourceTask.java
@@ -0,0 +1,642 @@
+/*
+ * 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;
+
+import org.apache.kafka.clients.admin.NewTopic;
+import org.apache.kafka.clients.admin.TopicDescription;
+import org.apache.kafka.clients.producer.Callback;
+import org.apache.kafka.clients.producer.Producer;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.clients.producer.RecordMetadata;
+import org.apache.kafka.common.KafkaException;
+import org.apache.kafka.common.header.internals.RecordHeaders;
+import org.apache.kafka.common.metrics.Sensor;
+import org.apache.kafka.common.metrics.stats.Avg;
+import org.apache.kafka.common.metrics.stats.CumulativeSum;
+import org.apache.kafka.common.metrics.stats.Max;
+import org.apache.kafka.common.metrics.stats.Rate;
+import org.apache.kafka.common.metrics.stats.Value;
+import org.apache.kafka.common.utils.Time;
+import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.connect.errors.ConnectException;
+import org.apache.kafka.connect.errors.RetriableException;
+import org.apache.kafka.connect.header.Header;
+import org.apache.kafka.connect.header.Headers;
+import org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator;
+import org.apache.kafka.connect.runtime.errors.Stage;
+import org.apache.kafka.connect.source.SourceRecord;
+import org.apache.kafka.connect.source.SourceTask;
+import org.apache.kafka.connect.source.SourceTaskContext;
+import org.apache.kafka.connect.storage.CloseableOffsetStorageReader;
+import org.apache.kafka.connect.storage.ConnectorOffsetBackingStore;
+import org.apache.kafka.connect.storage.Converter;
+import org.apache.kafka.connect.storage.HeaderConverter;
+import org.apache.kafka.connect.storage.OffsetStorageWriter;
+import org.apache.kafka.connect.storage.StatusBackingStore;
+import org.apache.kafka.connect.util.ConnectUtils;
+import org.apache.kafka.connect.util.ConnectorTaskId;
+import org.apache.kafka.connect.util.TopicAdmin;
+import org.apache.kafka.connect.util.TopicCreation;
+import org.apache.kafka.connect.util.TopicCreationGroup;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.time.Duration;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Executor;
+import java.util.concurrent.TimeUnit;
+
+import static org.apache.kafka.connect.runtime.WorkerConfig.TOPIC_TRACKING_ENABLE_CONFIG;
+
+/**
+ * WorkerTask that contains shared logic for running source tasks with either standard or exactly-once delivery guarantees.
+ */
+public abstract class AbstractWorkerSourceTask extends WorkerTask {
+ private static final Logger log = LoggerFactory.getLogger(AbstractWorkerSourceTask.class);
+
+ private static final long SEND_FAILED_BACKOFF_MS = 100;
+
+ /**
+ * Hook to define custom startup behavior before the calling {@link SourceTask#initialize(SourceTaskContext)}
+ * and {@link SourceTask#start(Map)}.
+ */
+ protected abstract void prepareToInitializeTask();
+
+ /**
+ * Hook to define custom initialization behavior when preparing to begin the poll-convert-send loop for the first time,
+ * or when re-entering the loop after being paused.
+ */
+ protected abstract void prepareToEnterSendLoop();
+
+ /**
+ * Hook to define custom periodic behavior to be performed at the top of every iteration of the poll-convert-send loop.
+ */
+ protected abstract void beginSendIteration();
+
+ /**
+ * Hook to define custom periodic checks for health, metrics, etc. Called whenever {@link SourceTask#poll()} is about to be invoked.
+ */
+ protected abstract void prepareToPollTask();
+
+ /**
+ * Invoked when a record provided by the task has been filtered out by a transform or the converter,
+ * or will be discarded due to failures during transformation or conversion.
+ * @param record the pre-transform record that has been dropped; never null.
+ */
+ protected abstract void recordDropped(SourceRecord record);
+
+ /**
+ * Invoked when a record is about to be dispatched to the producer. May be invoked multiple times for the same
+ * record if retriable errors are encountered.
+ * @param sourceRecord the pre-transform {@link SourceRecord} provided by the source task; never null.
+ * @param producerRecord the {@link ProducerRecord} produced by transforming and converting the
+ * {@code sourceRecord}; never null;
+ * @return a {@link SubmittedRecords.SubmittedRecord} to be {@link SubmittedRecords.SubmittedRecord#ack() acknowledged}
+ * if the corresponding producer record is ack'd by Kafka or {@link SubmittedRecords.SubmittedRecord#drop() dropped}
+ * if synchronously rejected by the producer. Can also be {@link Optional#empty()} if it is not necessary to track the acknowledgment
+ * of individual producer records
+ */
+ protected abstract Optional prepareToSendRecord(
+ SourceRecord sourceRecord,
+ ProducerRecord producerRecord
+ );
+
+ /**
+ * Invoked when a record has been transformed, converted, and dispatched to the producer successfully via
+ * {@link Producer#send}. Does not guarantee that the record has been sent to Kafka or ack'd by the required number
+ * of brokers, but does guarantee that it will never be re-processed.
+ * @param record the pre-transform {@link SourceRecord} that was successfully dispatched to the producer; never null.
+ */
+ protected abstract void recordDispatched(SourceRecord record);
+
+ /**
+ * Invoked when an entire batch of records returned from {@link SourceTask#poll} has been transformed, converted,
+ * and either discarded due to transform/conversion errors, filtered by a transform, or dispatched to the producer
+ * successfully via {@link Producer#send}. Does not guarantee that the records have been sent to Kafka or ack'd by the
+ * required number of brokers, but does guarantee that none of the records in the batch will ever be re-processed during
+ * the lifetime of this task. At most one record batch is polled from the task in between calls to this method.
+ */
+ protected abstract void batchDispatched();
+
+ /**
+ * Invoked when a record has been sent and ack'd by the Kafka cluster. Note that this method may be invoked
+ * concurrently and should therefore be made thread-safe.
+ * @param sourceRecord the pre-transform {@link SourceRecord} that was successfully sent to Kafka; never null.
+ * @param producerRecord the {@link ProducerRecord} produced by transforming and converting the
+ * {@code sourceRecord}; never null;
+ * @param recordMetadata the {@link RecordMetadata} for the corresponding producer record; never null.
+ */
+ protected abstract void recordSent(
+ SourceRecord sourceRecord,
+ ProducerRecord producerRecord,
+ RecordMetadata recordMetadata
+ );
+
+ /**
+ * Invoked when a record given to {@link Producer#send(ProducerRecord, Callback)} has failed with a non-retriable error.
+ * @param synchronous whether the error occurred during the invocation of {@link Producer#send(ProducerRecord, Callback)}.
+ * If {@code false}, indicates that the error was reported asynchronously by the producer by a {@link Callback}
+ * @param producerRecord the {@link ProducerRecord} that the producer failed to send; never null
+ * @param preTransformRecord the pre-transform {@link SourceRecord} that the producer record was derived from; never null
+ * @param e the exception that was either thrown from {@link Producer#send(ProducerRecord, Callback)}, or reported by the producer
+ * via {@link Callback} after the call to {@link Producer#send(ProducerRecord, Callback)} completed
+ */
+ protected abstract void producerSendFailed(
+ boolean synchronous,
+ ProducerRecord producerRecord,
+ SourceRecord preTransformRecord,
+ Exception e
+ );
+
+ /**
+ * Invoked when no more records will be polled from the task or dispatched to the producer. Should attempt to
+ * commit the offsets for any outstanding records when possible.
+ * @param failed whether the task is undergoing a healthy or an unhealthy shutdown
+ */
+ protected abstract void finalOffsetCommit(boolean failed);
+
+
+ protected final WorkerConfig workerConfig;
+ protected final WorkerSourceTaskContext sourceTaskContext;
+ protected final OffsetStorageWriter offsetWriter;
+ protected final Producer producer;
+
+ private final SourceTask task;
+ private final Converter keyConverter;
+ private final Converter valueConverter;
+ private final HeaderConverter headerConverter;
+ private final TransformationChain transformationChain;
+ private final TopicAdmin admin;
+ private final CloseableOffsetStorageReader offsetReader;
+ private final ConnectorOffsetBackingStore offsetStore;
+ private final SourceTaskMetricsGroup sourceTaskMetricsGroup;
+ private final CountDownLatch stopRequestedLatch;
+ private final boolean topicTrackingEnabled;
+ private final TopicCreation topicCreation;
+ private final Executor closeExecutor;
+
+ // Visible for testing
+ List toSend;
+ protected Map taskConfig;
+ protected boolean started = false;
+
+ protected AbstractWorkerSourceTask(ConnectorTaskId id,
+ SourceTask task,
+ TaskStatus.Listener statusListener,
+ TargetState initialState,
+ Converter keyConverter,
+ Converter valueConverter,
+ HeaderConverter headerConverter,
+ TransformationChain transformationChain,
+ WorkerSourceTaskContext sourceTaskContext,
+ Producer producer,
+ TopicAdmin admin,
+ Map topicGroups,
+ CloseableOffsetStorageReader offsetReader,
+ OffsetStorageWriter offsetWriter,
+ ConnectorOffsetBackingStore offsetStore,
+ WorkerConfig workerConfig,
+ ConnectMetrics connectMetrics,
+ ClassLoader loader,
+ Time time,
+ RetryWithToleranceOperator retryWithToleranceOperator,
+ StatusBackingStore statusBackingStore,
+ Executor closeExecutor) {
+
+ super(id, statusListener, initialState, loader, connectMetrics,
+ retryWithToleranceOperator, time, statusBackingStore);
+
+ this.workerConfig = workerConfig;
+ this.task = task;
+ this.keyConverter = keyConverter;
+ this.valueConverter = valueConverter;
+ this.headerConverter = headerConverter;
+ this.transformationChain = transformationChain;
+ this.producer = producer;
+ this.admin = admin;
+ this.offsetReader = offsetReader;
+ this.offsetWriter = offsetWriter;
+ this.offsetStore = offsetStore;
+ this.closeExecutor = closeExecutor;
+ this.sourceTaskContext = sourceTaskContext;
+
+ this.stopRequestedLatch = new CountDownLatch(1);
+ this.sourceTaskMetricsGroup = new SourceTaskMetricsGroup(id, connectMetrics);
+ this.topicTrackingEnabled = workerConfig.getBoolean(TOPIC_TRACKING_ENABLE_CONFIG);
+ this.topicCreation = TopicCreation.newTopicCreation(workerConfig, topicGroups);
+ }
+
+ @Override
+ public void initialize(TaskConfig taskConfig) {
+ try {
+ this.taskConfig = taskConfig.originalsStrings();
+ } catch (Throwable t) {
+ log.error("{} Task failed initialization and will not be started.", this, t);
+ onFailure(t);
+ }
+ }
+
+ @Override
+ protected void initializeAndStart() {
+ prepareToInitializeTask();
+ // If we try to start the task at all by invoking initialize, then count this as
+ // "started" and expect a subsequent call to the task's stop() method
+ // to properly clean up any resources allocated by its initialize() or
+ // start() methods. If the task throws an exception during stop(),
+ // the worst thing that happens is another exception gets logged for an already-
+ // failed task
+ started = true;
+ task.initialize(sourceTaskContext);
+ task.start(taskConfig);
+ log.info("{} Source task finished initialization and start", this);
+ }
+
+ @Override
+ public void cancel() {
+ super.cancel();
+ // Preemptively close the offset reader in case the task is blocked on an offset read.
+ offsetReader.close();
+ // We proactively close the producer here as the main work thread for the task may
+ // be blocked indefinitely in a call to Producer::send if automatic topic creation is
+ // not enabled on either the connector or the Kafka cluster. Closing the producer should
+ // unblock it in that case and allow shutdown to proceed normally.
+ // With a duration of 0, the producer's own shutdown logic should be fairly quick,
+ // but closing user-pluggable classes like interceptors may lag indefinitely. So, we
+ // call close on a separate thread in order to avoid blocking the herder's tick thread.
+ closeExecutor.execute(() -> closeProducer(Duration.ZERO));
+ }
+
+ @Override
+ public void stop() {
+ super.stop();
+ stopRequestedLatch.countDown();
+ }
+
+ @Override
+ public void removeMetrics() {
+ Utils.closeQuietly(sourceTaskMetricsGroup::close, "source task metrics tracker");
+ super.removeMetrics();
+ }
+
+ @Override
+ protected void close() {
+ if (started) {
+ Utils.closeQuietly(task::stop, "source task");
+ }
+
+ closeProducer(Duration.ofSeconds(30));
+
+ if (admin != null) {
+ Utils.closeQuietly(() -> admin.close(Duration.ofSeconds(30)), "source task admin");
+ }
+ Utils.closeQuietly(transformationChain, "transformation chain");
+ Utils.closeQuietly(retryWithToleranceOperator, "retry operator");
+ Utils.closeQuietly(offsetReader, "offset reader");
+ Utils.closeQuietly(offsetStore::stop, "offset backing store");
+ }
+
+ private void closeProducer(Duration duration) {
+ if (producer != null) {
+ Utils.closeQuietly(() -> producer.close(duration), "source task producer");
+ }
+ }
+
+ @Override
+ public void execute() {
+ try {
+ prepareToEnterSendLoop();
+ while (!isStopping()) {
+ beginSendIteration();
+
+ if (shouldPause()) {
+ onPause();
+ if (awaitUnpause()) {
+ onResume();
+ prepareToEnterSendLoop();
+ }
+ continue;
+ }
+
+ if (toSend == null) {
+ prepareToPollTask();
+
+ log.trace("{} Nothing to send to Kafka. Polling source for additional records", this);
+ long start = time.milliseconds();
+ toSend = poll();
+ if (toSend != null) {
+ recordPollReturned(toSend.size(), time.milliseconds() - start);
+ }
+ }
+ if (toSend == null)
+ continue;
+ log.trace("{} About to send {} records to Kafka", this, toSend.size());
+ if (sendRecords()) {
+ batchDispatched();
+ } else {
+ stopRequestedLatch.await(SEND_FAILED_BACKOFF_MS, TimeUnit.MILLISECONDS);
+ }
+ }
+ } catch (InterruptedException e) {
+ // Ignore and allow to exit.
+ } catch (RuntimeException e) {
+ try {
+ finalOffsetCommit(true);
+ } catch (Exception offsetException) {
+ log.error("Failed to commit offsets for already-failing task", offsetException);
+ }
+ throw e;
+ }
+ finalOffsetCommit(false);
+ }
+
+ /**
+ * Try to send a batch of records. If a send fails and is retriable, this saves the remainder of the batch so it can
+ * be retried after backing off. If a send fails and is not retriable, this will throw a ConnectException.
+ * @return true if all messages were sent, false if some need to be retried
+ */
+ // Visible for testing
+ boolean sendRecords() {
+ int processed = 0;
+ recordBatch(toSend.size());
+ final SourceRecordWriteCounter counter =
+ toSend.size() > 0 ? new SourceRecordWriteCounter(toSend.size(), sourceTaskMetricsGroup) : null;
+ for (final SourceRecord preTransformRecord : toSend) {
+ retryWithToleranceOperator.sourceRecord(preTransformRecord);
+ final SourceRecord record = transformationChain.apply(preTransformRecord);
+ final ProducerRecord producerRecord = convertTransformedRecord(record);
+ if (producerRecord == null || retryWithToleranceOperator.failed()) {
+ counter.skipRecord();
+ recordDropped(preTransformRecord);
+ continue;
+ }
+
+ log.trace("{} Appending record to the topic {} with key {}, value {}", this, record.topic(), record.key(), record.value());
+ Optional submittedRecord = prepareToSendRecord(preTransformRecord, producerRecord);
+ try {
+ final String topic = producerRecord.topic();
+ maybeCreateTopic(topic);
+ producer.send(
+ producerRecord,
+ (recordMetadata, e) -> {
+ if (e != null) {
+ log.debug("{} failed to send record to {}: ", AbstractWorkerSourceTask.this, topic, e);
+ log.trace("{} Failed record: {}", AbstractWorkerSourceTask.this, preTransformRecord);
+ producerSendFailed(false, producerRecord, preTransformRecord, e);
+ } else {
+ counter.completeRecord();
+ log.trace("{} Wrote record successfully: topic {} partition {} offset {}",
+ AbstractWorkerSourceTask.this,
+ recordMetadata.topic(), recordMetadata.partition(),
+ recordMetadata.offset());
+ recordSent(preTransformRecord, producerRecord, recordMetadata);
+ submittedRecord.ifPresent(SubmittedRecords.SubmittedRecord::ack);
+ if (topicTrackingEnabled) {
+ recordActiveTopic(producerRecord.topic());
+ }
+ }
+ });
+ // Note that this will cause retries to take place within a transaction
+ } catch (RetriableException | org.apache.kafka.common.errors.RetriableException e) {
+ log.warn("{} Failed to send record to topic '{}' and partition '{}'. Backing off before retrying: ",
+ this, producerRecord.topic(), producerRecord.partition(), e);
+ toSend = toSend.subList(processed, toSend.size());
+ submittedRecord.ifPresent(SubmittedRecords.SubmittedRecord::drop);
+ counter.retryRemaining();
+ return false;
+ } catch (ConnectException e) {
+ log.warn("{} Failed to send record to topic '{}' and partition '{}' due to an unrecoverable exception: ",
+ this, producerRecord.topic(), producerRecord.partition(), e);
+ log.trace("{} Failed to send {} with unrecoverable exception: ", this, producerRecord, e);
+ throw e;
+ } catch (KafkaException e) {
+ producerSendFailed(true, producerRecord, preTransformRecord, e);
+ }
+ processed++;
+ recordDispatched(preTransformRecord);
+ }
+ toSend = null;
+ return true;
+ }
+
+ protected List poll() throws InterruptedException {
+ try {
+ return task.poll();
+ } catch (RetriableException | org.apache.kafka.common.errors.RetriableException e) {
+ log.warn("{} failed to poll records from SourceTask. Will retry operation.", this, e);
+ // Do nothing. Let the framework poll whenever it's ready.
+ return null;
+ }
+ }
+
+ /**
+ * Convert the source record into a producer record.
+ *
+ * @param record the transformed record
+ * @return the producer record which can sent over to Kafka. A null is returned if the input is null or
+ * if an error was encountered during any of the converter stages.
+ */
+ protected ProducerRecord convertTransformedRecord(SourceRecord record) {
+ if (record == null) {
+ return null;
+ }
+
+ RecordHeaders headers = retryWithToleranceOperator.execute(() -> convertHeaderFor(record), Stage.HEADER_CONVERTER, headerConverter.getClass());
+
+ byte[] key = retryWithToleranceOperator.execute(() -> keyConverter.fromConnectData(record.topic(), headers, record.keySchema(), record.key()),
+ Stage.KEY_CONVERTER, keyConverter.getClass());
+
+ byte[] value = retryWithToleranceOperator.execute(() -> valueConverter.fromConnectData(record.topic(), headers, record.valueSchema(), record.value()),
+ Stage.VALUE_CONVERTER, valueConverter.getClass());
+
+ if (retryWithToleranceOperator.failed()) {
+ return null;
+ }
+
+ return new ProducerRecord<>(record.topic(), record.kafkaPartition(),
+ ConnectUtils.checkAndConvertTimestamp(record.timestamp()), key, value, headers);
+ }
+
+ // Due to transformations that may change the destination topic of a record (such as
+ // RegexRouter) topic creation can not be batched for multiple topics
+ private void maybeCreateTopic(String topic) {
+ if (!topicCreation.isTopicCreationRequired(topic)) {
+ log.trace("Topic creation by the connector is disabled or the topic {} was previously created." +
+ "If auto.create.topics.enable is enabled on the broker, " +
+ "the topic will be created with default settings", topic);
+ return;
+ }
+ log.info("The task will send records to topic '{}' for the first time. Checking "
+ + "whether topic exists", topic);
+ Map existing = admin.describeTopics(topic);
+ if (!existing.isEmpty()) {
+ log.info("Topic '{}' already exists.", topic);
+ topicCreation.addTopic(topic);
+ return;
+ }
+
+ log.info("Creating topic '{}'", topic);
+ TopicCreationGroup topicGroup = topicCreation.findFirstGroup(topic);
+ log.debug("Topic '{}' matched topic creation group: {}", topic, topicGroup);
+ NewTopic newTopic = topicGroup.newTopic(topic);
+
+ TopicAdmin.TopicCreationResponse response = admin.createOrFindTopics(newTopic);
+ if (response.isCreated(newTopic.name())) {
+ topicCreation.addTopic(topic);
+ log.info("Created topic '{}' using creation group {}", newTopic, topicGroup);
+ } else if (response.isExisting(newTopic.name())) {
+ topicCreation.addTopic(topic);
+ log.info("Found existing topic '{}'", newTopic);
+ } else {
+ // The topic still does not exist and could not be created, so treat it as a task failure
+ log.warn("Request to create new topic '{}' failed", topic);
+ throw new ConnectException("Task failed to create new topic " + newTopic + ". Ensure "
+ + "that the task is authorized to create topics or that the topic exists and "
+ + "restart the task");
+ }
+ }
+
+ protected RecordHeaders convertHeaderFor(SourceRecord record) {
+ Headers headers = record.headers();
+ RecordHeaders result = new RecordHeaders();
+ if (headers != null) {
+ String topic = record.topic();
+ for (Header header : headers) {
+ String key = header.key();
+ byte[] rawHeader = headerConverter.fromConnectHeader(topic, key, header.schema(), header.value());
+ result.add(key, rawHeader);
+ }
+ }
+ return result;
+ }
+
+ protected void commitTaskRecord(SourceRecord record, RecordMetadata metadata) {
+ try {
+ task.commitRecord(record, metadata);
+ } catch (Throwable t) {
+ log.error("{} Exception thrown while calling task.commitRecord()", this, t);
+ }
+ }
+
+ protected void commitSourceTask() {
+ try {
+ this.task.commit();
+ } catch (Throwable t) {
+ log.error("{} Exception thrown while calling task.commit()", this, t);
+ }
+ }
+
+ protected void recordPollReturned(int numRecordsInBatch, long duration) {
+ sourceTaskMetricsGroup.recordPoll(numRecordsInBatch, duration);
+ }
+
+ SourceTaskMetricsGroup sourceTaskMetricsGroup() {
+ return sourceTaskMetricsGroup;
+ }
+
+ static class SourceRecordWriteCounter {
+ private final SourceTaskMetricsGroup metricsGroup;
+ private final int batchSize;
+ private boolean completed = false;
+ private int counter;
+ public SourceRecordWriteCounter(int batchSize, SourceTaskMetricsGroup metricsGroup) {
+ assert batchSize > 0;
+ assert metricsGroup != null;
+ this.batchSize = batchSize;
+ counter = batchSize;
+ this.metricsGroup = metricsGroup;
+ }
+ public void skipRecord() {
+ if (counter > 0 && --counter == 0) {
+ finishedAllWrites();
+ }
+ }
+ public void completeRecord() {
+ if (counter > 0 && --counter == 0) {
+ finishedAllWrites();
+ }
+ }
+ public void retryRemaining() {
+ finishedAllWrites();
+ }
+ private void finishedAllWrites() {
+ if (!completed) {
+ metricsGroup.recordWrite(batchSize - counter);
+ completed = true;
+ }
+ }
+ }
+
+ static class SourceTaskMetricsGroup {
+ private final ConnectMetrics.MetricGroup metricGroup;
+ private final Sensor sourceRecordPoll;
+ private final Sensor sourceRecordWrite;
+ private final Sensor sourceRecordActiveCount;
+ private final Sensor pollTime;
+ private int activeRecordCount;
+
+ public SourceTaskMetricsGroup(ConnectorTaskId id, ConnectMetrics connectMetrics) {
+ ConnectMetricsRegistry registry = connectMetrics.registry();
+ metricGroup = connectMetrics.group(registry.sourceTaskGroupName(),
+ registry.connectorTagName(), id.connector(),
+ registry.taskTagName(), Integer.toString(id.task()));
+ // remove any previously created metrics in this group to prevent collisions.
+ metricGroup.close();
+
+ sourceRecordPoll = metricGroup.sensor("source-record-poll");
+ sourceRecordPoll.add(metricGroup.metricName(registry.sourceRecordPollRate), new Rate());
+ sourceRecordPoll.add(metricGroup.metricName(registry.sourceRecordPollTotal), new CumulativeSum());
+
+ sourceRecordWrite = metricGroup.sensor("source-record-write");
+ sourceRecordWrite.add(metricGroup.metricName(registry.sourceRecordWriteRate), new Rate());
+ sourceRecordWrite.add(metricGroup.metricName(registry.sourceRecordWriteTotal), new CumulativeSum());
+
+ pollTime = metricGroup.sensor("poll-batch-time");
+ pollTime.add(metricGroup.metricName(registry.sourceRecordPollBatchTimeMax), new Max());
+ pollTime.add(metricGroup.metricName(registry.sourceRecordPollBatchTimeAvg), new Avg());
+
+ sourceRecordActiveCount = metricGroup.sensor("source-record-active-count");
+ sourceRecordActiveCount.add(metricGroup.metricName(registry.sourceRecordActiveCount), new Value());
+ sourceRecordActiveCount.add(metricGroup.metricName(registry.sourceRecordActiveCountMax), new Max());
+ sourceRecordActiveCount.add(metricGroup.metricName(registry.sourceRecordActiveCountAvg), new Avg());
+ }
+
+ void close() {
+ metricGroup.close();
+ }
+
+ void recordPoll(int batchSize, long duration) {
+ sourceRecordPoll.record(batchSize);
+ pollTime.record(duration);
+ activeRecordCount += batchSize;
+ sourceRecordActiveCount.record(activeRecordCount);
+ }
+
+ void recordWrite(int recordCount) {
+ sourceRecordWrite.record(recordCount);
+ activeRecordCount -= recordCount;
+ activeRecordCount = Math.max(0, activeRecordCount);
+ sourceRecordActiveCount.record(activeRecordCount);
+ }
+
+ protected ConnectMetrics.MetricGroup metricGroup() {
+ return metricGroup;
+ }
+ }
+}
diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/ConnectMetricsRegistry.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/ConnectMetricsRegistry.java
index cf567450fe2a0..f301439da8356 100644
--- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/ConnectMetricsRegistry.java
+++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/ConnectMetricsRegistry.java
@@ -112,6 +112,9 @@ public class ConnectMetricsRegistry {
public final MetricNameTemplate dlqProduceRequests;
public final MetricNameTemplate dlqProduceFailures;
public final MetricNameTemplate lastErrorTimestamp;
+ public final MetricNameTemplate transactionSizeMin;
+ public final MetricNameTemplate transactionSizeMax;
+ public final MetricNameTemplate transactionSizeAvg;
public Map connectorStatusMetrics;
@@ -207,6 +210,16 @@ public ConnectMetricsRegistry(Set tags) {
"completely written to Kafka.",
sourceTaskTags);
+ transactionSizeMin = createTemplate("transaction-size-min", SOURCE_TASK_GROUP_NAME,
+ "The number of records in the smallest transaction the task has committed so far. ",
+ sourceTaskTags);
+ transactionSizeMax = createTemplate("transaction-size-max", SOURCE_TASK_GROUP_NAME,
+ "The number of records in the largest transaction the task has committed so far.",
+ sourceTaskTags);
+ transactionSizeAvg = createTemplate("transaction-size-avg", SOURCE_TASK_GROUP_NAME,
+ "The average number of records in the transactions the task has committed so far.",
+ sourceTaskTags);
+
/***** Sink worker task level *****/
Set sinkTaskTags = new LinkedHashSet<>(tags);
sinkTaskTags.add(CONNECTOR_TAG_NAME);
diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/ExactlyOnceWorkerSourceTask.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/ExactlyOnceWorkerSourceTask.java
new file mode 100644
index 0000000000000..fe08819f65676
--- /dev/null
+++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/ExactlyOnceWorkerSourceTask.java
@@ -0,0 +1,525 @@
+/*
+ * 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;
+
+import org.apache.kafka.clients.producer.Producer;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.clients.producer.RecordMetadata;
+import org.apache.kafka.common.errors.InvalidProducerEpochException;
+import org.apache.kafka.common.metrics.Sensor;
+import org.apache.kafka.common.metrics.stats.Avg;
+import org.apache.kafka.common.metrics.stats.Max;
+import org.apache.kafka.common.metrics.stats.Min;
+import org.apache.kafka.common.utils.Time;
+import org.apache.kafka.connect.errors.ConnectException;
+import org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator;
+import org.apache.kafka.connect.source.SourceRecord;
+import org.apache.kafka.connect.source.SourceTask;
+import org.apache.kafka.connect.source.SourceTask.TransactionBoundary;
+import org.apache.kafka.connect.storage.CloseableOffsetStorageReader;
+import org.apache.kafka.connect.storage.ClusterConfigState;
+import org.apache.kafka.connect.storage.ConnectorOffsetBackingStore;
+import org.apache.kafka.connect.storage.Converter;
+import org.apache.kafka.connect.storage.HeaderConverter;
+import org.apache.kafka.connect.storage.OffsetStorageWriter;
+import org.apache.kafka.connect.storage.StatusBackingStore;
+import org.apache.kafka.connect.util.ConnectorTaskId;
+import org.apache.kafka.connect.util.LoggingContext;
+import org.apache.kafka.connect.util.TopicAdmin;
+import org.apache.kafka.connect.util.TopicCreationGroup;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.concurrent.Executor;
+import java.util.concurrent.Future;
+import java.util.concurrent.atomic.AtomicReference;
+
+
+/**
+ * WorkerTask that uses a SourceTask to ingest data into Kafka, with support for exactly-once delivery guarantees.
+ */
+class ExactlyOnceWorkerSourceTask extends AbstractWorkerSourceTask {
+ private static final Logger log = LoggerFactory.getLogger(ExactlyOnceWorkerSourceTask.class);
+
+ private boolean transactionOpen;
+ private final LinkedHashMap commitableRecords;
+
+ private final TransactionManager transactionManager;
+ private final TransactionMetricsGroup transactionMetrics;
+
+ private final ConnectorOffsetBackingStore offsetBackingStore;
+ private final Runnable preProducerCheck;
+ private final Runnable postProducerCheck;
+
+ public ExactlyOnceWorkerSourceTask(ConnectorTaskId id,
+ SourceTask task,
+ TaskStatus.Listener statusListener,
+ TargetState initialState,
+ Converter keyConverter,
+ Converter valueConverter,
+ HeaderConverter headerConverter,
+ TransformationChain transformationChain,
+ Producer producer,
+ TopicAdmin admin,
+ Map topicGroups,
+ CloseableOffsetStorageReader offsetReader,
+ OffsetStorageWriter offsetWriter,
+ ConnectorOffsetBackingStore offsetBackingStore,
+ WorkerConfig workerConfig,
+ ClusterConfigState configState,
+ ConnectMetrics connectMetrics,
+ ClassLoader loader,
+ Time time,
+ RetryWithToleranceOperator retryWithToleranceOperator,
+ StatusBackingStore statusBackingStore,
+ SourceConnectorConfig sourceConfig,
+ Executor closeExecutor,
+ Runnable preProducerCheck,
+ Runnable postProducerCheck) {
+ super(id, task, statusListener, initialState, keyConverter, valueConverter, headerConverter, transformationChain,
+ new WorkerSourceTaskContext(offsetReader, id, configState, buildTransactionContext(sourceConfig)),
+ producer, admin, topicGroups, offsetReader, offsetWriter, offsetBackingStore, workerConfig, connectMetrics,
+ loader, time, retryWithToleranceOperator, statusBackingStore, closeExecutor);
+
+ this.transactionOpen = false;
+ this.commitableRecords = new LinkedHashMap<>();
+ this.offsetBackingStore = offsetBackingStore;
+
+ this.preProducerCheck = preProducerCheck;
+ this.postProducerCheck = postProducerCheck;
+
+ this.transactionManager = buildTransactionManager(workerConfig, sourceConfig, sourceTaskContext.transactionContext());
+ this.transactionMetrics = new TransactionMetricsGroup(id, connectMetrics);
+ }
+
+ private static WorkerTransactionContext buildTransactionContext(SourceConnectorConfig sourceConfig) {
+ return TransactionBoundary.CONNECTOR.equals(sourceConfig.transactionBoundary())
+ ? new WorkerTransactionContext()
+ : null;
+ }
+
+ @Override
+ protected void prepareToInitializeTask() {
+ preProducerCheck.run();
+
+ // Try not to start up the offset store (which has its own producer and consumer) if we've already been shut down at this point
+ if (isStopping())
+ return;
+ offsetBackingStore.start();
+
+ // Try not to initialize the transactional producer (which may accidentally fence out other, later task generations) if we've already
+ // been shut down at this point
+ if (isStopping())
+ return;
+ producer.initTransactions();
+
+ postProducerCheck.run();
+ }
+
+ @Override
+ protected void prepareToEnterSendLoop() {
+ transactionManager.initialize();
+ }
+
+ @Override
+ protected void beginSendIteration() {
+ // No-op
+ }
+
+ @Override
+ protected void prepareToPollTask() {
+ // No-op
+ }
+
+ @Override
+ protected void recordDropped(SourceRecord record) {
+ synchronized (this) {
+ commitableRecords.put(record, null);
+ }
+ transactionManager.maybeCommitTransactionForRecord(record);
+ }
+
+ @Override
+ protected Optional prepareToSendRecord(
+ SourceRecord sourceRecord,
+ ProducerRecord producerRecord
+ ) {
+ if (offsetBackingStore.primaryOffsetsTopic().equals(producerRecord.topic())) {
+ // This is to prevent deadlock that occurs when:
+ // 1. A task provides a record whose topic is the task's offsets topic
+ // 2. That record is dispatched to the task's producer in a transaction that remains open
+ // at least until the worker polls the task again
+ // 3. In the subsequent call to SourceTask::poll, the task requests offsets from the worker
+ // (which requires a read to the end of the offsets topic, and will block until any open
+ // transactions on the topic are either committed or aborted)
+ throw new ConnectException("Source tasks may not produce to their own offsets topics when exactly-once support is enabled");
+ }
+ maybeBeginTransaction();
+ return Optional.empty();
+ }
+
+ @Override
+ protected void recordDispatched(SourceRecord record) {
+ // Offsets are converted & serialized in the OffsetWriter
+ // Important: we only save offsets for the record after it has been accepted by the producer; this way,
+ // we commit those offsets if and only if the record is sent successfully.
+ offsetWriter.offset(record.sourcePartition(), record.sourceOffset());
+ transactionMetrics.addRecord();
+ transactionManager.maybeCommitTransactionForRecord(record);
+ }
+
+ @Override
+ protected void batchDispatched() {
+ transactionManager.maybeCommitTransactionForBatch();
+ }
+
+ @Override
+ protected void recordSent(
+ SourceRecord sourceRecord,
+ ProducerRecord producerRecord,
+ RecordMetadata recordMetadata
+ ) {
+ synchronized (this) {
+ commitableRecords.put(sourceRecord, recordMetadata);
+ }
+ }
+
+ @Override
+ protected void producerSendFailed(
+ boolean synchronous,
+ ProducerRecord producerRecord,
+ SourceRecord preTransformRecord,
+ Exception e
+ ) {
+ if (synchronous) {
+ throw maybeWrapProducerSendException(
+ "Unrecoverable exception trying to send",
+ e
+ );
+ } else {
+ // No-op; all asynchronously-reported producer exceptions should be bubbled up again by Producer::commitTransaction
+ }
+ }
+
+ @Override
+ protected void finalOffsetCommit(boolean failed) {
+ if (failed) {
+ log.debug("Skipping final offset commit as task has failed");
+ return;
+ }
+
+ // It should be safe to commit here even if we were in the middle of retrying on RetriableExceptions in the
+ // send loop since we only track source offsets for records that have been successfully dispatched to the
+ // producer.
+ // Any records that we were retrying on (and any records after them in the batch) won't be included in the
+ // transaction and their offsets won't be committed, but (unless the user has requested connector-defined
+ // transaction boundaries), it's better to commit some data than none.
+ transactionManager.maybeCommitFinalTransaction();
+ }
+
+ @Override
+ protected void onPause() {
+ super.onPause();
+ // Commit the transaction now so that we don't end up with a hanging transaction, or worse, get fenced out
+ // and fail the task once unpaused
+ transactionManager.maybeCommitFinalTransaction();
+ }
+
+ private void maybeBeginTransaction() {
+ if (!transactionOpen) {
+ producer.beginTransaction();
+ transactionOpen = true;
+ }
+ }
+
+ private void commitTransaction() {
+ log.debug("{} Committing offsets", this);
+
+ long started = time.milliseconds();
+
+ // We might have just aborted a transaction, in which case we'll have to begin a new one
+ // in order to commit offsets
+ maybeBeginTransaction();
+
+ AtomicReference flushError = new AtomicReference<>();
+ Future offsetFlush = null;
+ if (offsetWriter.beginFlush()) {
+ // Now we can actually write the offsets to the internal topic.
+ offsetFlush = offsetWriter.doFlush((error, result) -> {
+ if (error != null) {
+ log.error("{} Failed to flush offsets to storage: ", ExactlyOnceWorkerSourceTask.this, error);
+ flushError.compareAndSet(null, error);
+ } else {
+ log.trace("{} Finished flushing offsets to storage", ExactlyOnceWorkerSourceTask.this);
+ }
+ });
+ }
+
+ // Commit the transaction
+ // Blocks until all outstanding records have been sent and ack'd
+ try {
+ producer.commitTransaction();
+ if (offsetFlush != null) {
+ // Although it's guaranteed by the above call to Producer::commitTransaction that all outstanding
+ // records for the task's producer (including those sent to the offsets topic) have been delivered and
+ // ack'd, there is no guarantee that the producer callbacks for those records have been completed. So,
+ // we add this call to Future::get to ensure that these callbacks are invoked successfully before
+ // proceeding.
+ offsetFlush.get();
+ }
+ } catch (Throwable t) {
+ flushError.compareAndSet(null, t);
+ }
+
+ transactionOpen = false;
+
+ Throwable error = flushError.get();
+ if (error != null) {
+ recordCommitFailure(time.milliseconds() - started, null);
+ offsetWriter.cancelFlush();
+ throw maybeWrapProducerSendException(
+ "Failed to flush offsets and/or records for task " + id,
+ error
+ );
+ }
+
+ transactionMetrics.commitTransaction();
+
+ long durationMillis = time.milliseconds() - started;
+ recordCommitSuccess(durationMillis);
+ log.debug("{} Finished commitOffsets successfully in {} ms", this, durationMillis);
+
+ // No need for any synchronization here; all other accesses to this field take place in producer callbacks,
+ // which should all be completed by this point
+ commitableRecords.forEach(this::commitTaskRecord);
+ commitableRecords.clear();
+ commitSourceTask();
+ }
+
+ private RuntimeException maybeWrapProducerSendException(String message, Throwable error) {
+ if (isPossibleTransactionTimeoutError(error)) {
+ return wrapTransactionTimeoutError(error);
+ } else {
+ return new ConnectException(message, error);
+ }
+ }
+
+ private static boolean isPossibleTransactionTimeoutError(Throwable error) {
+ return error instanceof InvalidProducerEpochException
+ || error.getCause() instanceof InvalidProducerEpochException;
+ }
+
+ private ConnectException wrapTransactionTimeoutError(Throwable error) {
+ return new ConnectException(
+ "The task " + id + " was unable to finish writing records to Kafka before its producer transaction expired. "
+ + "It may be necessary to reconfigure this connector in order for it to run healthily with exactly-once support. "
+ + "Options for this include: tune the connector's producer configuration for higher throughput, "
+ + "increase the transaction timeout for the connector's producers, "
+ + "decrease the offset commit interval (if using interval-based transaction boundaries), "
+ + "or use the 'poll' transaction boundary (if the connector is not already configured to use it).",
+ error
+ );
+ }
+
+ @Override
+ public String toString() {
+ return "ExactlyOnceWorkerSourceTask{" +
+ "id=" + id +
+ '}';
+ }
+
+ private abstract class TransactionManager {
+ protected boolean shouldCommitTransactionForRecord(SourceRecord record) {
+ return false;
+ }
+
+ protected boolean shouldCommitTransactionForBatch(long currentTimeMs) {
+ return false;
+ }
+
+ protected boolean shouldCommitFinalTransaction() {
+ return false;
+ }
+
+ /**
+ * Hook to signal that a new transaction cycle has been started. May be invoked
+ * multiple times if the task is paused and then resumed. It can be assumed that
+ * a new transaction is created at least every time an existing transaction is
+ * committed; this is just a hook to notify that a new transaction may have been
+ * created outside of that flow as well.
+ */
+ protected void initialize() {
+ }
+
+ public void maybeCommitTransactionForRecord(SourceRecord record) {
+ maybeCommitTransaction(shouldCommitTransactionForRecord(record));
+ }
+
+ public void maybeCommitTransactionForBatch() {
+ maybeCommitTransaction(shouldCommitTransactionForBatch(time.milliseconds()));
+ }
+
+ public void maybeCommitFinalTransaction() {
+ maybeCommitTransaction(shouldCommitFinalTransaction());
+ }
+
+ private void maybeCommitTransaction(boolean shouldCommit) {
+ if (shouldCommit && (transactionOpen || offsetWriter.willFlush())) {
+ try (LoggingContext loggingContext = LoggingContext.forOffsets(id)) {
+ commitTransaction();
+ }
+ }
+ }
+ }
+
+ private TransactionManager buildTransactionManager(
+ WorkerConfig workerConfig,
+ SourceConnectorConfig sourceConfig,
+ WorkerTransactionContext transactionContext) {
+ TransactionBoundary boundary = sourceConfig.transactionBoundary();
+ switch (boundary) {
+ case POLL:
+ return new TransactionManager() {
+ @Override
+ protected boolean shouldCommitTransactionForBatch(long currentTimeMs) {
+ return true;
+ }
+
+ @Override
+ protected boolean shouldCommitFinalTransaction() {
+ return true;
+ }
+ };
+
+ case INTERVAL:
+ long transactionBoundaryInterval = Optional.ofNullable(sourceConfig.transactionBoundaryInterval())
+ .orElse(workerConfig.offsetCommitInterval());
+ return new TransactionManager() {
+ private final long commitInterval = transactionBoundaryInterval;
+ private long lastCommit;
+
+ @Override
+ public void initialize() {
+ this.lastCommit = time.milliseconds();
+ }
+
+ @Override
+ protected boolean shouldCommitTransactionForBatch(long currentTimeMs) {
+ if (time.milliseconds() >= lastCommit + commitInterval) {
+ lastCommit = time.milliseconds();
+ return true;
+ } else {
+ return false;
+ }
+ }
+
+ @Override
+ protected boolean shouldCommitFinalTransaction() {
+ return true;
+ }
+ };
+
+ case CONNECTOR:
+ Objects.requireNonNull(transactionContext, "Transaction context must be provided when using connector-defined transaction boundaries");
+ return new TransactionManager() {
+ @Override
+ protected boolean shouldCommitFinalTransaction() {
+ return shouldCommitTransactionForBatch(time.milliseconds());
+ }
+
+ @Override
+ protected boolean shouldCommitTransactionForBatch(long currentTimeMs) {
+ if (transactionContext.shouldAbortBatch()) {
+ log.info("Aborting transaction for batch as requested by connector");
+ abortTransaction();
+ // We abort the transaction, which causes all the records up to this point to be dropped, but we still want to
+ // commit offsets so that the task doesn't see the same records all over again
+ return true;
+ }
+ return transactionContext.shouldCommitBatch();
+ }
+
+ @Override
+ protected boolean shouldCommitTransactionForRecord(SourceRecord record) {
+ if (transactionContext.shouldAbortOn(record)) {
+ log.info("Aborting transaction for record on topic {} as requested by connector", record.topic());
+ log.trace("Last record in aborted transaction: {}", record);
+ abortTransaction();
+ // We abort the transaction, which causes all the records up to this point to be dropped, but we still want to
+ // commit offsets so that the task doesn't see the same records all over again
+ return true;
+ }
+ return transactionContext.shouldCommitOn(record);
+ }
+
+ private void abortTransaction() {
+ producer.abortTransaction();
+ transactionMetrics.abortTransaction();
+ transactionOpen = false;
+ }
+ };
+ default:
+ throw new IllegalArgumentException("Unrecognized transaction boundary: " + boundary);
+ }
+ }
+
+ TransactionMetricsGroup transactionMetricsGroup() {
+ return transactionMetrics;
+ }
+
+
+ static class TransactionMetricsGroup {
+ private final Sensor transactionSize;
+ private int size;
+ private final ConnectMetrics.MetricGroup metricGroup;
+
+ public TransactionMetricsGroup(ConnectorTaskId id, ConnectMetrics connectMetrics) {
+ ConnectMetricsRegistry registry = connectMetrics.registry();
+ metricGroup = connectMetrics.group(registry.sourceTaskGroupName(),
+ registry.connectorTagName(), id.connector(),
+ registry.taskTagName(), Integer.toString(id.task()));
+
+ transactionSize = metricGroup.sensor("transaction-size");
+ transactionSize.add(metricGroup.metricName(registry.transactionSizeAvg), new Avg());
+ transactionSize.add(metricGroup.metricName(registry.transactionSizeMin), new Min());
+ transactionSize.add(metricGroup.metricName(registry.transactionSizeMax), new Max());
+ }
+
+ void addRecord() {
+ size++;
+ }
+
+ void abortTransaction() {
+ size = 0;
+ }
+
+ void commitTransaction() {
+ transactionSize.record(size);
+ size = 0;
+ }
+
+ protected ConnectMetrics.MetricGroup metricGroup() {
+ return metricGroup;
+ }
+
+ }
+
+}
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 945797f8ceaee..21da7c6ea8454 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
@@ -137,6 +137,17 @@ public interface Herder {
*/
void putTaskConfigs(String connName, List