mimaison commented on a change in pull request #11773: URL: https://github.com/apache/kafka/pull/11773#discussion_r819517322
########## File path: 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 the next batch of records from {@link SourceTask#poll()}. All of Review comment: Should it be `Requests a transaction abort after the next batch of records`? ########## File path: connect/api/src/main/java/org/apache/kafka/connect/source/SourceTask.java ########## @@ -16,17 +16,63 @@ */ 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; /** * SourceTask is a Task that pulls records from another system for storage in Kafka. */ 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"; Review comment: Is there a reason this config is defined here? Is it just temporary? (I've not looked the the remaining PRs yet) ########## File path: connect/api/src/main/java/org/apache/kafka/connect/source/SourceConnector.java ########## @@ -28,4 +30,46 @@ 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. + * + * <p>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}. + * + * <p>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. Review comment: To be honest, I'd be in favor in having stronger guarantees on the order of method calls and on the overall lifecycle of plugins. At the moment, connectors can't assume very much and we get weird behaviors like described in [KIP-419](https://cwiki.apache.org/confluence/display/KAFKA/KIP-419%3A+Safely+notify+Kafka+Connect+SourceTask+is+stopped). At least here, I like that you were explicit in the Javadoc! -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
