chia7712 commented on code in PR #15989: URL: https://github.com/apache/kafka/pull/15989#discussion_r1621334504
########## 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: We need to use `verify` to make sure this method is called as expected. -- 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