psiby commented on code in PR #23268:
URL: https://github.com/apache/kafka/pull/23268#discussion_r4246805709
##########
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:
Good point, fixed in #23783. Now we refresh the config snapshot before
precondition checks (both before writing offsets and before writing the config).
--
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]
