psiby commented on code in PR #23783:
URL: https://github.com/apache/kafka/pull/23783#discussion_r4256325586
##########
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedHerder.java:
##########
@@ -1209,11 +1224,17 @@ private void setInitialOffsetsThenWriteConfig(String
connName, Map<String, Strin
}
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)) {
+ // The offsets are already durable, so wipe them back
out if the create can't be completed, unless
+ // another request created the connector in the
meantime, since that would reset its offsets.
+ Callback<Created<ConnectorInfo>> wipeOnFailure =
(error, ignored) -> {
+ if (error instanceof AlreadyExistsException) {
+ callback.onCompletion(error, null);
+ } else {
+ wipeInitialOffsetsAfterFailedCreate(connName,
config, error, callback);
+ }
+ };
+ if (!refreshConfigSnapshotForInitialOffsets(connName,
wipeOnFailure)
Review Comment:
Same fix here.
--
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]
