chia7712 commented on code in PR #15989:
URL: https://github.com/apache/kafka/pull/15989#discussion_r1623140561
##########
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:
Please see following example:
```java
@Test
public void testPutTaskConfigsZeroTasks() throws Exception {
configStorage.setupAndCreateKafkaBasedLog(TOPIC, config);
verifyConfigure();
configStorage.start();
verify(configLog).start();
// Records to be read by consumer as it reads to the end of the log
doAnswer(expectReadToEnd(new LinkedHashMap<>())).
doAnswer(expectReadToEnd(Collections.singletonMap(COMMIT_TASKS_CONFIG_KEYS.get(0),
CONFIGS_SERIALIZED.get(0))))
.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
// Bootstrap as if we had already added the connector, but no tasks
had been added yet
addConnector(CONNECTOR_IDS.get(0), SAMPLE_CONFIGS.get(0),
Collections.emptyList());
// Null before writing
ClusterConfigState configState = configStorage.snapshot();
assertEquals(-1, configState.offset());
// Writing task configs should block until all the writes have been
performed and the root record update
// has completed
List<Map<String, String>> taskConfigs = Collections.emptyList();
configStorage.putTaskConfigs("connector1", taskConfigs);
// Validate root config by listing all connectors and tasks
configState = configStorage.snapshot();
assertEquals(1, configState.offset());
String connectorName = CONNECTOR_IDS.get(0);
assertEquals(Collections.singletonList(connectorName), new
ArrayList<>(configState.connectors()));
assertEquals(Collections.emptyList(),
configState.tasks(connectorName));
assertEquals(Collections.EMPTY_SET,
configState.inconsistentConnectors());
// As soon as root is rewritten, we should see a callback notifying
us that we reconfigured some tasks
verify(configUpdateListener).onTaskConfigUpdate(Collections.emptyList());
configStorage.stop();
verify(configLog).stop();
}
```
--
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
