For backwards compatibility, the default implementation will return {@code null}, but connector authors 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.3
+ */
+ public ExactlyOnceSupport exactlyOnceSupport(Map 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.3
+ * @see TransactionContext
+ */
+ public ConnectorTransactionBoundaries canDefineTransactionBoundaries(Map
* 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.
- *
* 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.
- *
* 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.
- *
* 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.
- * 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.3
+ */
+ 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..f90d75baf4748
--- /dev/null
+++ b/connect/api/src/main/java/org/apache/kafka/connect/source/TransactionContext.java
@@ -0,0 +1,56 @@
+/*
+ * 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 after 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 so than the records in that transaction
+ * are not reprocessed. If the data should instead 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 so that the records in that transaction are not reprocessed. 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);
+}