You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by ce...@apache.org on 2023/01/13 14:25:17 UTC
[kafka] 02/02: KAFKA-13709 (follow-up): Avoid mention of 'exactly-once delivery' or 'delivery guarantees' in Connect (#13106)
This is an automated email from the ASF dual-hosted git repository.
cegerton pushed a commit to branch 3.4
in repository https://gitbox.apache.org/repos/asf/kafka.git
commit c7b0c50c0ea9e87596d6a12fa1ef25694c724756
Author: Chris Egerton <ch...@aiven.io>
AuthorDate: Fri Jan 13 09:19:29 2023 -0500
KAFKA-13709 (follow-up): Avoid mention of 'exactly-once delivery' or 'delivery guarantees' in Connect (#13106)
---
.../apache/kafka/connect/source/ExactlyOnceSupport.java | 6 +++---
.../org/apache/kafka/connect/source/SourceConnector.java | 4 ++--
.../kafka/connect/file/FileStreamSourceConnector.java | 4 ++--
.../kafka/connect/runtime/AbstractWorkerSourceTask.java | 3 ++-
.../connect/runtime/ExactlyOnceWorkerSourceTask.java | 2 +-
.../kafka/connect/runtime/SourceConnectorConfig.java | 4 ++--
.../java/org/apache/kafka/connect/runtime/Worker.java | 2 +-
.../connect/runtime/distributed/DistributedHerder.java | 6 +++---
.../integration/ExactlyOnceSourceIntegrationTest.java | 8 ++++----
.../runtime/distributed/DistributedHerderTest.java | 2 +-
docs/connect.html | 16 ++++++++--------
11 files changed, 29 insertions(+), 28 deletions(-)
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
index 0e9cdc0cebd..73f9917c632 100644
--- 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
@@ -17,15 +17,15 @@
package org.apache.kafka.connect.source;
/**
- * An enum to represent the level of support for exactly-once delivery from a source connector.
+ * An enum to represent the level of support for exactly-once semantics from a source connector.
*/
public enum ExactlyOnceSupport {
/**
- * Signals that a connector supports exactly-once delivery.
+ * Signals that a connector supports exactly-once semantics.
*/
SUPPORTED,
/**
- * Signals that a connector does not support exactly-once delivery.
+ * Signals that a connector does not support exactly-once semantics.
*/
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 3bd012f9fbc..c06279df421 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
@@ -32,7 +32,7 @@ public abstract class SourceConnector extends Connector {
}
/**
- * Signals whether the connector supports exactly-once delivery guarantees with a proposed configuration.
+ * Signals whether the connector supports exactly-once semantics with a proposed configuration.
* Connector authors 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 authors are
@@ -46,7 +46,7 @@ public abstract class SourceConnector extends 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.
+ * exactly-once semantics.
* @since 3.3
*/
public ExactlyOnceSupport exactlyOnceSupport(Map<String, String> connectorConfig) {
diff --git a/connect/file/src/main/java/org/apache/kafka/connect/file/FileStreamSourceConnector.java b/connect/file/src/main/java/org/apache/kafka/connect/file/FileStreamSourceConnector.java
index 55090d889c9..eef6da2b91d 100644
--- a/connect/file/src/main/java/org/apache/kafka/connect/file/FileStreamSourceConnector.java
+++ b/connect/file/src/main/java/org/apache/kafka/connect/file/FileStreamSourceConnector.java
@@ -92,9 +92,9 @@ public class FileStreamSourceConnector extends SourceConnector {
public ExactlyOnceSupport exactlyOnceSupport(Map<String, String> props) {
AbstractConfig parsedConfig = new AbstractConfig(CONFIG_DEF, props);
String filename = parsedConfig.getString(FILE_CONFIG);
- // We can provide exactly-once guarantees if reading from a "real" file
+ // We can provide exactly-once semantics if reading from a "real" file
// (as long as the file is only appended to over the lifetime of the connector)
- // If we're reading from stdin, we can't provide exactly-once guarantees
+ // If we're reading from stdin, we can't provide exactly-once semantics
// since we don't even track offsets
return filename != null && !filename.isEmpty()
? ExactlyOnceSupport.SUPPORTED
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
index 671e9aefa7f..9f6104b26a0 100644
--- 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
@@ -69,7 +69,8 @@ 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.
+ * WorkerTask that contains shared logic for running source tasks with either standard semantics
+ * (i.e., either at-least-once or at-most-once) or exactly-once semantics.
*/
public abstract class AbstractWorkerSourceTask extends WorkerTask {
private static final Logger log = LoggerFactory.getLogger(AbstractWorkerSourceTask.class);
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
index d02a5f54286..fe580e66cd8 100644
--- 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
@@ -55,7 +55,7 @@ import java.util.concurrent.atomic.AtomicReference;
/**
- * WorkerTask that uses a SourceTask to ingest data into Kafka, with support for exactly-once delivery guarantees.
+ * WorkerTask that uses a SourceTask to ingest data into Kafka, with support for exactly-once semantics.
*/
class ExactlyOnceWorkerSourceTask extends AbstractWorkerSourceTask {
private static final Logger log = LoggerFactory.getLogger(ExactlyOnceWorkerSourceTask.class);
diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/SourceConnectorConfig.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/SourceConnectorConfig.java
index 2115bda662b..c74407593bd 100644
--- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/SourceConnectorConfig.java
+++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/SourceConnectorConfig.java
@@ -79,8 +79,8 @@ public class SourceConnectorConfig extends ConnectorConfig {
public static final String EXACTLY_ONCE_SUPPORT_CONFIG = "exactly.once.support";
private static final String EXACTLY_ONCE_SUPPORT_DOC = "Permitted values are " + String.join(", ", enumOptions(ExactlyOnceSupportLevel.class)) + ". "
- + "If set to \"" + REQUIRED + "\", forces a preflight check for the connector to ensure that it can provide exactly-once delivery "
- + "with the given configuration. Some connectors may be capable of providing exactly-once delivery but not signal to "
+ + "If set to \"" + REQUIRED + "\", forces a preflight check for the connector to ensure that it can provide exactly-once semantics "
+ + "with the given configuration. Some connectors may be capable of providing exactly-once semantics but not signal to "
+ "Connect that they support this; in that case, documentation for the connector should be consulted carefully before "
+ "creating it, and the value for this property should be set to \"" + REQUESTED + "\". "
+ "Additionally, if the value is set to \"" + REQUIRED + "\" but the worker that performs preflight validation does not have "
diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Worker.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Worker.java
index 342c5e5a05a..f966ea4cb55 100644
--- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Worker.java
+++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Worker.java
@@ -822,7 +822,7 @@ public class Worker {
Map<String, Object> result = baseConsumerConfigs(
connName, defaultClientId, config, connConfig, connectorClass,
connectorClientConfigOverridePolicy, clusterId, ConnectorType.SOURCE);
- // Users can disable this if they want to; it won't affect delivery guarantees since the task isn't exactly-once anyways
+ // Users can disable this if they want to since the task isn't exactly-once anyways
result.putIfAbsent(
ConsumerConfig.ISOLATION_LEVEL_CONFIG,
IsolationLevel.READ_COMMITTED.toString().toLowerCase(Locale.ROOT));
diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedHerder.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedHerder.java
index 1983e88f16b..0f45be163cc 100644
--- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedHerder.java
+++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedHerder.java
@@ -942,18 +942,18 @@ public class DistributedHerder extends AbstractHerder implements Runnable {
if (exactlyOnceSupport == null) {
validationErrorMessage = "The connector does not implement the API required for preflight validation of exactly-once "
+ "source support. Please consult the documentation for the connector to determine whether it supports exactly-once "
- + "guarantees, and then consider reconfiguring the connector to use the value \""
+ + "semantics, and then consider reconfiguring the connector to use the value \""
+ SourceConnectorConfig.ExactlyOnceSupportLevel.REQUESTED
+ "\" for this property (which will disable this preflight check and allow the connector to be created).";
} else if (ExactlyOnceSupport.UNSUPPORTED.equals(exactlyOnceSupport)) {
- validationErrorMessage = "The connector does not support exactly-once delivery guarantees with the provided configuration.";
+ validationErrorMessage = "The connector does not support exactly-once semantics with the provided configuration.";
} else {
throw new ConnectException("Unexpected value returned from SourceConnector::exactlyOnceSupport: " + exactlyOnceSupport);
}
validatedExactlyOnceSupport.addErrorMessage(validationErrorMessage);
}
} catch (Exception e) {
- log.error("Failed while validating connector support for exactly-once guarantees", e);
+ log.error("Failed while validating connector support for exactly-once semantics", e);
String validationErrorMessage = "An unexpected error occurred during validation";
String failureMessage = e.getMessage();
if (failureMessage != null && !failureMessage.trim().isEmpty()) {
diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/integration/ExactlyOnceSourceIntegrationTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/integration/ExactlyOnceSourceIntegrationTest.java
index dd839093b53..a2dd730b344 100644
--- a/connect/runtime/src/test/java/org/apache/kafka/connect/integration/ExactlyOnceSourceIntegrationTest.java
+++ b/connect/runtime/src/test/java/org/apache/kafka/connect/integration/ExactlyOnceSourceIntegrationTest.java
@@ -159,7 +159,7 @@ public class ExactlyOnceSourceIntegrationTest {
}
/**
- * A simple test for the pre-flight validation API for connectors to provide their own delivery guarantees.
+ * A simple test for the pre-flight validation API for connectors to provide their own guarantees for exactly-once semantics.
*/
@Test
public void testPreflightValidation() {
@@ -727,8 +727,8 @@ public class ExactlyOnceSourceIntegrationTest {
* Then, a "soft downgrade" is simulated: the Connect cluster is shut down and reconfigured to disable
* exactly-once support. The cluster is brought up again, the connector is allowed to produce some data,
* the connector is shut down, and this time, the records the connector has produced are inspected for
- * accuracy. Because of the downgrade, exactly-once guarantees are lost, but we check to make sure that
- * the task has maintained exactly-once delivery <i>up to the last-committed record</i>.
+ * accuracy. Because of the downgrade, exactly-once semantics are lost, but we check to make sure that
+ * the task has maintained exactly-once semantics <i>up to the last-committed record</i>.
*/
@Test
public void testSeparateOffsetsTopic() throws Exception {
@@ -858,7 +858,7 @@ public class ExactlyOnceSourceIntegrationTest {
);
assertTrue("Not enough records produced by source connector. Expected at least: " + recordsProduced + " + but got " + sourceRecords.count(),
sourceRecords.count() >= recordsProduced);
- // also have to check which offsets have actually been committed, since we no longer have exactly-once guarantees
+ // also have to check which offsets have actually been committed, since we no longer have exactly-once semantics
offsetRecords = connectorTargetedCluster.consumeAll(
CONSUME_RECORDS_TIMEOUT_MS,
Collections.singletonMap(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed"),
diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/distributed/DistributedHerderTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/distributed/DistributedHerderTest.java
index ab17060c887..1d79d59aa33 100644
--- a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/distributed/DistributedHerderTest.java
+++ b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/distributed/DistributedHerderTest.java
@@ -855,7 +855,7 @@ public class DistributedHerderTest {
List<String> errors = validatedConfigs.get(SourceConnectorConfig.EXACTLY_ONCE_SUPPORT_CONFIG).errorMessages();
assertEquals(
- Collections.singletonList("The connector does not support exactly-once delivery guarantees with the provided configuration."),
+ Collections.singletonList("The connector does not support exactly-once semantics with the provided configuration."),
errors);
PowerMock.verifyAll();
diff --git a/docs/connect.html b/docs/connect.html
index fca18ddf7d1..8709e4c4dae 100644
--- a/docs/connect.html
+++ b/docs/connect.html
@@ -371,15 +371,15 @@ errors.tolerance=all</pre>
<h4><a id="connect_exactlyonce" href="#connect_exactlyonce">Exactly-once support</a></h4>
- <p>Kafka Connect is capable of providing exactly-once delivery guarantees for sink connectors (as of version 0.11.0) and source connectors (as of version 3.3.0). Please note that <b>support for exactly-once delivery is highly dependent on the type of connector you run.</b> Even if you set all the correct worker properties in the configuration for each node in a cluster, if a connector is not designed to, or cannot take advantage of the capabilities of the Kafka Connect framework, exa [...]
+ <p>Kafka Connect is capable of providing exactly-once semantics for sink connectors (as of version 0.11.0) and source connectors (as of version 3.3.0). Please note that <b>support for exactly-once semantics is highly dependent on the type of connector you run.</b> Even if you set all the correct worker properties in the configuration for each node in a cluster, if a connector is not designed to, or cannot take advantage of the capabilities of the Kafka Connect framework, exactly-once [...]
<h5><a id="connect_exactlyoncesink" href="#connect_exactlyoncesink">Sink connectors</a></h5>
- <p>If a sink connector supports exactly-once delivery, to enable exactly-once delivery at the Connect worker level, you must ensure its consumer group is configured to ignore records in aborted transactions. You can do this by setting the worker property <code>consumer.isolation.level</code> to <code>read_committed</code> or, if running a version of Kafka Connect that supports it, using a <a href="#connectconfigs_connector.client.config.override.policy">connector client config overri [...]
+ <p>If a sink connector supports exactly-once semantics, to enable exactly-once at the Connect worker level, you must ensure its consumer group is configured to ignore records in aborted transactions. You can do this by setting the worker property <code>consumer.isolation.level</code> to <code>read_committed</code> or, if running a version of Kafka Connect that supports it, using a <a href="#connectconfigs_connector.client.config.override.policy">connector client config override polic [...]
<h5><a id="connect_exactlyoncesource" href="connect_exactlyoncesource">Source connectors</a></h5>
- <p>If a source connector supports exactly-once delivery, you must configure your Connect cluster to enable framework-level support for exactly-once delivery for source connectors. Additional ACLs may be necessary if running against a secured Kafka cluster. Note that exactly-once support for source connectors is currently only available in distributed mode; standalone Connect workers cannot provide exactly-once guarantees.</p>
+ <p>If a source connector supports exactly-once semantics, you must configure your Connect cluster to enable framework-level support for exactly-once source connectors. Additional ACLs may be necessary if running against a secured Kafka cluster. Note that exactly-once support for source connectors is currently only available in distributed mode; standalone Connect workers cannot provide exactly-once semantics.</p>
<h6>Worker configuration</h6>
@@ -770,24 +770,24 @@ public List<SourceRecord> poll() {
<p>A few additional preflight validation APIs can be implemented by source connector developers.</p>
- <p>Some users may require exactly-once delivery guarantees from a connector. In this case, they may set the <code>exactly.once.support</code> property to <code>required</code> in the configuration for the connector. When this happens, the Kafka Connect framework will ask the connector whether it can provide exactly-once delivery guarantees with the specified configuration. This is done by invoking the <code>exactlyOnceSupport</code> method on the connector.</p>
+ <p>Some users may require exactly-once semantics from a connector. In this case, they may set the <code>exactly.once.support</code> property to <code>required</code> in the configuration for the connector. When this happens, the Kafka Connect framework will ask the connector whether it can provide exactly-once semantics with the specified configuration. This is done by invoking the <code>exactlyOnceSupport</code> method on the connector.</p>
- <p>If a connector doesn't support exactly-once delivery, it should still implement this method to let users know for certain that it cannot provide exactly-once delivery guarantees:</p>
+ <p>If a connector doesn't support exactly-once semantics, it should still implement this method to let users know for certain that it cannot provide exactly-once semantics:</p>
<pre class="brush: java;">
@Override
public ExactlyOnceSupport exactlyOnceSupport(Map<String, String> props) {
- // This connector cannot provide exactly-once delivery guarantees under any conditions
+ // This connector cannot provide exactly-once semantics under any conditions
return ExactlyOnceSupport.UNSUPPORTED;
}
</pre>
- <p>Otherwise, a connector should examine the configuration, and return <code>ExactlyOnceSupport.SUPPORTED</code> if it can provide exactly-once delivery guarantees:</p>
+ <p>Otherwise, a connector should examine the configuration, and return <code>ExactlyOnceSupport.SUPPORTED</code> if it can provide exactly-once semantics:</p>
<pre class="brush: java;">
@Override
public ExactlyOnceSupport exactlyOnceSupport(Map<String, String> props) {
- // This connector can always provide exactly-once delivery guarantees
+ // This connector can always provide exactly-once semantics
return ExactlyOnceSupport.SUPPORTED;
}
</pre>