chiacyu commented on code in PR #15989:
URL: https://github.com/apache/kafka/pull/15989#discussion_r1623158977
##########
connect/runtime/src/test/java/org/apache/kafka/connect/storage/KafkaConfigBackingStoreMockitoTest.java:
##########
@@ -1184,6 +1185,141 @@ public void
testRestoreRestartRequestInconsistentState() {
verify(configLog).stop();
}
+ @Test
+ public void testPutTaskConfigsZeroTasks() throws Exception {
+ when(configLog.partitionCount()).thenReturn(1);
+
+ configStorage.setupAndCreateKafkaBasedLog(TOPIC, config);
+ verifyConfigure();
+ configStorage.start();
+
+ // Bootstrap as if we had already added the connector, but no tasks
had been added yet
+ whiteBoxAddConnector(CONNECTOR_IDS.get(0), SAMPLE_CONFIGS.get(0),
Collections.emptyList());
+
+ // Null before writing
+ ClusterConfigState configState = configStorage.snapshot();
+ assertEquals(-1, configState.offset());
+
+ // Task configs should read to end, write to the log, read to end,
write root.
+
doAnswer(expectReadToEnd(Collections.emptyMap())).when(configLog).readToEnd();
+
+ expectConvertWriteRead(
+ COMMIT_TASKS_CONFIG_KEYS.get(0),
KafkaConfigBackingStore.CONNECTOR_TASKS_COMMIT_V0, CONFIGS_SERIALIZED.get(0),
+ "tasks", 0); // We have 0 tasks
+
+ configStorage.putTaskConfigs("connector1", Collections.emptyList());
+
+ // As soon as root is rewritten, we should see a callback notifying us
that we reconfigured some tasks
+ configUpdateListener.onTaskConfigUpdate(Collections.emptyList());
Review Comment:
Thanks for the example, would apply.
--
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: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
