chia7712 commented on code in PR #23268:
URL: https://github.com/apache/kafka/pull/23268#discussion_r4232878078
##########
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/AbstractHerder.java:
##########
@@ -1316,6 +1316,66 @@ public void resetConnectorOffsets(String connName,
Callback<Message> callback) {
*/
protected abstract void modifyConnectorOffsets(String connName,
Map<Map<String, ?>, Map<String, ?>> offsets, Callback<Message> cb);
+ /**
+ * Validate the offsets supplied via the {@code initial_offsets} field of
a connector creation request. Performs no
+ * I/O, so that an invalid request is rejected before any of the
connector's existing offsets are wiped.
+ *
+ * @param connName the name of the connector being created
+ * @param initialOffsets the offsets supplied by the user; may not be
{@code null}
+ * @throws BadRequestException if the offsets cannot be used as a
connector's initial offsets
+ */
+ protected void validateInitialOffsets(String connName, Map<Map<String, ?>,
Map<String, ?>> initialOffsets) {
+ if (initialOffsets.isEmpty()) {
+ throw new BadRequestException("The initial offsets for connector "
+ connName + " may not be empty");
+ }
+ // A null offset means "delete this partition's offset" when altering,
but is meaningless on create: every offset
+ // the connector has is wiped before these are written, so there is
nothing left to delete. Issuing the delete
+ // anyway targets the consumer group our own wipe just removed, which
the alter path (unlike the reset path) does
+ // not tolerate. This is deliberately stricter than PATCH
/connectors/{connector}/offsets.
+ // Not Map::containsValue, which throws rather than returning false on
the immutable maps Map.of(...) produces.
+ if (initialOffsets.values().stream().anyMatch(offset -> offset ==
null)) {
+ throw new BadRequestException("The initial offsets for connector "
+ connName + " may not contain null offset "
+ + "values. All existing offsets for the connector are
removed before the initial offsets are written, "
+ + "so a null offset has nothing to delete.");
+ }
+ }
+
+ /**
+ * Wipe every offset the connector currently has and write {@code
initialOffsets} in their place.
+ * <p>
+ * Unlike {@link #alterConnectorOffsets} and {@link
#resetConnectorOffsets}, this skips the usual preconditions (that
+ * the connector exists and is STOPPED with no running tasks): the
connector is being created, so it is not expected
+ * to exist and has no tasks. Its config has not been written yet, so
nothing can be running under this name.
+ */
+ protected void setInitialConnectorOffsets(String connName, Map<String,
String> config,
+ Map<Map<String, ?>, Map<String,
?>> initialOffsets, Callback<Message> cb) {
+ worker.modifyConnectorOffsets(connName, config, initialOffsets, true,
cb);
+ }
+
+ /**
+ * Remove the initial offsets written for a connector whose creation could
not subsequently be completed, so that a
+ * failed creation does not leave offsets behind for a connector that does
not exist, and then complete
+ * {@code callback} with the original failure rather than with any error
from the cleanup itself.
+ */
+ protected void wipeInitialOffsetsAfterFailedCreate(String connName,
Map<String, String> config,
+ Throwable
configWriteError, Callback<Created<ConnectorInfo>> callback) {
+ log.error("Failed to create connector {} after its initial offsets had
been written; "
+ + "removing those offsets so that the failed creation does not
leave them behind", connName, configWriteError);
+ Callback<Message> cleanupCallback = (wipeError, ignored) -> {
+ if (wipeError != null) {
+ log.error("Failed to remove the initial offsets for connector
{} after its creation failed. Offsets may "
+ + "remain for a connector that was not created;
re-submitting the identical creation request is "
+ + "safe and will overwrite them.", connName,
wipeError);
+ }
+ callback.onCompletion(configWriteError, null);
+ };
+ try {
+ worker.modifyConnectorOffsets(connName, config, null,
cleanupCallback);
Review Comment:
What happens if `AlreadyExistsException` is thrown? It appears we still
reset the connector’s offsets, which do not belong to this call.
##########
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedHerder.java:
##########
@@ -1146,33 +1149,112 @@ private void doPutConnectorConfig(
}
log.trace("Handling connector config request {}",
connName);
- if (!isLeader()) {
- callback.onCompletion(new NotLeaderException("Only
the leader can set connector configs.", leaderUrl()), null);
- return null;
- }
- boolean exists = configState.contains(connName);
- if (!allowReplace && exists) {
- callback.onCompletion(new
AlreadyExistsException("Connector " + connName + " already exists"), null);
+ if (!putConnectorConfigChecks(connName, allowReplace,
callback)) {
return null;
}
- log.trace("Submitting connector config {} {} {}",
connName, allowReplace, configState.connectors());
- writeToConfigTopicAsLeader(
- "writing a config for connector " + connName +
" to the config topic",
- () ->
configBackingStore.putConnectorConfig(connName, config, targetState)
- );
-
- // Note that we use the updated connector config
despite the fact that we don't have an updated
- // snapshot yet. The existing task info should still
be accurate.
- ConnectorInfo info = new ConnectorInfo(connName,
config, configState.tasks(connName),
- connectorType(config));
- callback.onCompletion(null, new Created<>(!exists,
info));
+ if (initialOffsets == null) {
+ writeConnectorConfigAndComplete(connName, config,
targetState, null, callback);
+ } else {
+ setInitialOffsetsThenWriteConfig(connName, config,
targetState, initialOffsets, allowReplace, callback);
+ }
return null;
},
forwardErrorAndTickThreadStages(callback)
);
}));
}
+
+ /**
+ * Perform the checks required before writing a connector's config as the
leader, completing {@code callback}
+ * exceptionally if any of them fail.
+ *
+ * @param connName the name of the connector whose config is to be written
+ * @param allowReplace whether an existing connector with this name may be
overwritten
+ * @param callback callback to complete exceptionally if a check fails
+ * @return true if all the checks passed, false otherwise
+ */
+ private boolean putConnectorConfigChecks(String connName, boolean
allowReplace, Callback<Created<ConnectorInfo>> callback) {
+ if (!isLeader()) {
+ callback.onCompletion(new NotLeaderException("Only the leader can
set connector configs.", leaderUrl()), null);
+ return false;
+ }
+ if (!allowReplace && configState.contains(connName)) {
+ callback.onCompletion(new AlreadyExistsException("Connector " +
connName + " already exists"), null);
+ return false;
+ }
+ return true;
+ }
+
+ /**
+ * Write a connector's initial offsets and, once they are durable, write
its config. Must be invoked on the tick thread.
+ * <p>
+ * The offsets are written asynchronously, so the config write is handed
back to the tick thread via a nested request
+ * rather than blocking it; this mirrors the {@link
#modifyConnectorOffsets} plus zombie fencing flow, including the
+ * re-check of preconditions that may have changed while the offsets were
being written.
+ */
+ private void setInitialOffsetsThenWriteConfig(String connName, Map<String,
String> config, TargetState targetState,
+ Map<Map<String, ?>,
Map<String, ?>> initialOffsets, boolean allowReplace,
+
Callback<Created<ConnectorInfo>> callback) {
+ // Validate before anything destructive happens, so an invalid request
can't wipe the connector's offsets. This
+ // performs no I/O and is safe to run on the tick thread; throwing
here is forwarded to the callback by the
+ // enclosing herder request.
+ validateInitialOffsets(connName, initialOffsets);
+
+ try (TickThreadStage stage = new TickThreadStage("setting initial
offsets for connector " + connName)) {
+ setInitialConnectorOffsets(connName, config, initialOffsets,
(offsetsError, message) -> {
+ if (offsetsError != null) {
+ callback.onCompletion(offsetsError, null);
+ return;
+ }
+ addRequest(
+ () -> {
+ // The offsets are already durable, so wipe them back
out if the create can't be completed,
+ // whether because the precondition re-check fails
here or the config write below throws.
+ Callback<Created<ConnectorInfo>> wipeOnFailure =
(error, ignored) ->
+ wipeInitialOffsetsAfterFailedCreate(connName,
config, error, callback);
+ if (!putConnectorConfigChecks(connName, allowReplace,
wipeOnFailure)) {
Review Comment:
The path of `modifyConnectorOffsets` calls `modifyConnectorOffsetsChecks` to
run `refreshConfigSnapshot`. It seems this path should apply that too.
Otherwise, it could see out-of-date configs.
##########
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/entities/CreateConnectorRequest.java:
##########
@@ -21,18 +21,54 @@
import com.fasterxml.jackson.annotation.JsonCreator;
import com.fasterxml.jackson.annotation.JsonProperty;
+import java.util.List;
import java.util.Locale;
import java.util.Map;
public record CreateConnectorRequest(
@JsonProperty("name") String name,
@JsonProperty("config") Map<String, String> config,
- @JsonProperty("initial_state") InitialState initialState
+ @JsonProperty("initial_state") InitialState initialState,
+ @JsonProperty("initial_offsets") List<ConnectorOffset> initialOffsets
) {
+ /**
+ * Canonical constructor, declared explicitly and annotated so that
Jackson unambiguously uses it as the creator
+ * rather than having to choose between this and the convenience
constructor below.
+ */
+ @JsonCreator
+ public CreateConnectorRequest(
Review Comment:
I guess this constructor is needed by some Jackson magic, but could we keep
a single constructor to avoid duplicating code? The duplication is a bit of a
mystery to me.
If there are too many tests relying on the convenience constructor, we could
add a helper method for those tests instead.
##########
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/entities/CreateConnectorRequest.java:
##########
@@ -21,18 +21,54 @@
import com.fasterxml.jackson.annotation.JsonCreator;
import com.fasterxml.jackson.annotation.JsonProperty;
+import java.util.List;
import java.util.Locale;
import java.util.Map;
public record CreateConnectorRequest(
@JsonProperty("name") String name,
@JsonProperty("config") Map<String, String> config,
- @JsonProperty("initial_state") InitialState initialState
+ @JsonProperty("initial_state") InitialState initialState,
+ @JsonProperty("initial_offsets") List<ConnectorOffset> initialOffsets
Review Comment:
Should we add ` @JsonInclude(JsonInclude.Include.NON_NULL)`, like
`ConnectorInfo`?
##########
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/resources/ConnectorsResource.java:
##########
Review Comment:
The KIP states the returned code is 200, but it is 201 actually. I guess the
KIP needs to be updated to match the implementation
--
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]
