OmniaGM commented on code in PR #15999:
URL: https://github.com/apache/kafka/pull/15999#discussion_r1617187677
##########
connect/mirror/src/main/java/org/apache/kafka/connect/mirror/MirrorCheckpointConnector.java:
##########
@@ -105,6 +108,49 @@ public void stop() {
Utils.closeQuietly(targetAdminClient, "target admin client");
}
+ @Override
+ public Config validate(Map<String, String> props) {
+ List<ConfigValue> configValues = super.validate(props).configValues();
+ String emitCheckpointsValue =
Optional.ofNullable(props.get(MirrorCheckpointConfig.EMIT_CHECKPOINTS_ENABLED)).orElse(Boolean.toString(MirrorCheckpointConfig.EMIT_CHECKPOINTS_ENABLED_DEFAULT));
+ String syncGroupOffsetsValue =
Optional.ofNullable(props.get(MirrorCheckpointConfig.SYNC_GROUP_OFFSETS_ENABLED)).orElse(Boolean.toString(MirrorCheckpointConfig.SYNC_GROUP_OFFSETS_ENABLED_DEFAULT));
+ String emitOffsetSyncsValue =
Optional.ofNullable(props.get(MirrorConnectorConfig.EMIT_OFFSET_SYNCS_ENABLED)).orElse(Boolean.toString(MirrorConnectorConfig.EMIT_OFFSET_SYNCS_ENABLED_DEFAULT));
+
+ if ("false".equals(emitCheckpointsValue) &&
"false".equals(syncGroupOffsetsValue)) {
+ ConfigValue syncGroupOffsets = configValues.stream().filter(prop
-> MirrorCheckpointConfig.SYNC_GROUP_OFFSETS_ENABLED.equals(prop.name()))
+ .findAny()
+ .orElseGet(() -> {
+ ConfigValue result = new
ConfigValue(MirrorCheckpointConfig.SYNC_GROUP_OFFSETS_ENABLED);
+ configValues.add(result);
+ return result;
+ });
+
+ ConfigValue emitCheckpoints = configValues.stream().filter(prop ->
MirrorCheckpointConfig.EMIT_CHECKPOINTS_ENABLED.equals(prop.name()))
+ .findAny()
+ .orElseGet(() -> {
+ ConfigValue result = new
ConfigValue(MirrorCheckpointConfig.EMIT_CHECKPOINTS_ENABLED);
+ configValues.add(result);
+ return result;
+ });
+
+ String errorMessage = "MirrorCheckpointConnector can't run with
both" +
+ MirrorCheckpointConfig.SYNC_GROUP_OFFSETS_ENABLED + ", " +
MirrorCheckpointConfig.EMIT_CHECKPOINTS_ENABLED + "set to false";
+ syncGroupOffsets.addErrorMessage(errorMessage);
+ emitCheckpoints.addErrorMessage(errorMessage);
+ }
+ if ("false".equals(emitOffsetSyncsValue) &&
("true".equals(emitCheckpointsValue) || "true".equals(syncGroupOffsetsValue))) {
+ ConfigValue emitOffsetSyncs = configValues.stream().filter(prop ->
MirrorConnectorConfig.EMIT_OFFSET_SYNCS_ENABLED.equals(prop.name()))
+ .findAny()
+ .orElseGet(() -> {
+ ConfigValue result = new
ConfigValue(MirrorConnectorConfig.EMIT_OFFSET_SYNCS_ENABLED);
+ configValues.add(result);
+ return result;
+ });
+
+ emitOffsetSyncs.addErrorMessage("MirrorCheckpointConnector can't
run while MirrorSourceConnector configured with" +
+ MirrorConnectorConfig.EMIT_OFFSET_SYNCS_ENABLED + "set to
false");
+ }
+ return new Config(configValues);
Review Comment:
Just want to clarify some points regarding connect (to the extend of my
understanding):
1. `Connector::validate` get called by the runtime before `Connector::start`
which will get us to fail faster
2. `Connector::validate` get called in
`AbstractHerder::validateConnectorConfig` which handles connector/task
lifecycle tracking. This also get called from connect REST api `config/validate`
3. While `Connector::start` get called during the reconfiguration of the
plugin in connect as well as at the start
So we can't move the validation to `Connector::start` and we can't move the
initialisation of `MirrorConnectorConfig` into `Connector::validate`.
If we want to use `.getBoolean` then I would suggest having a look into my
last
[commit](https://github.com/apache/kafka/pull/15999/commits/27c07546e6b6983e79b1c5989564b337f84e930c)
where I temporary initialised a connect configs to use the getters from
`AbstractConfig` which all mirror connector configs extends.
--
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]
