C0urante commented on code in PR #15999:
URL: https://github.com/apache/kafka/pull/15999#discussion_r1668803291
##########
connect/mirror/src/main/java/org/apache/kafka/connect/mirror/MirrorCheckpointConfig.java:
##########
@@ -166,6 +167,30 @@ Duration consumerPollTimeout() {
return Duration.ofMillis(getLong(CONSUMER_POLL_TIMEOUT_MILLIS));
}
+ public static Map<String, String> validate(Map<String, String> configs) {
+ Map<String, String> invalidConfigs = new HashMap<>();
+
+ // No point to validate when connector is disabled.
+ if ("false".equals(configs.getOrDefault(ENABLED, "true"))) {
+ return invalidConfigs;
+ }
+
+ boolean emitCheckpointDisabled =
"false".equals(configs.getOrDefault(EMIT_CHECKPOINTS_ENABLED, "true"));
+ boolean syncGroupOffsetsDisabled =
"false".equals(configs.getOrDefault(SYNC_GROUP_OFFSETS_ENABLED, "true"));
+
+ if (emitCheckpointDisabled && syncGroupOffsetsDisabled) {
+ Arrays.asList(SYNC_GROUP_OFFSETS_ENABLED,
EMIT_CHECKPOINTS_ENABLED).forEach(configName -> {
+ invalidConfigs.putIfAbsent(configName,
"MirrorCheckpointConnector can't run with both " + SYNC_GROUP_OFFSETS_ENABLED +
" and " +
Review Comment:
I didn't realize that we disabled the connector from generating task configs
if checkpoints were disabled. In that case, I think we should remove the check
for `sync.group.offsets.enabled` altogether and just check
`emit.checkpoints.enabled`. But generally I agree with @OmniaGM that if we
don't generate task configs for the connector when it's configured that way,
it's best to surface this as a validation error instead of silently degrading
to a no-op.
--
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]
