chia7712 commented on code in PR #23783:
URL: https://github.com/apache/kafka/pull/23783#discussion_r4251578175
##########
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:
ditto
##########
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) {
Review Comment:
There is a race condition that this does not fix. I opened
https://issues.apache.org/jira/browse/KAFKA-21270 to track it, since I believe
it is rare :)
##########
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedHerder.java:
##########
@@ -1149,6 +1149,9 @@ private void doPutConnectorConfig(
}
log.trace("Handling connector config request {}",
connName);
+ if (initialOffsets != null &&
!refreshConfigSnapshotForInitialOffsets(connName, callback)) {
Review Comment:
`putConnectorConfigChecks` returns directly if this is not the leader, so we
could defer `refreshConfigSnapshotForInitialOffsets` until after the leader
check, right?
--
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]
