You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by GitBox <gi...@apache.org> on 2023/01/12 01:42:21 UTC
[GitHub] [kafka] andymg3 commented on a diff in pull request #13104: KAFKA-14612: Make sure to write a new topics ConfigRecords to metadata log iff the topic is created
andymg3 commented on code in PR #13104:
URL: https://github.com/apache/kafka/pull/13104#discussion_r1067622704
##########
metadata/src/main/java/org/apache/kafka/controller/ConfigurationControlManager.java:
##########
@@ -172,20 +172,30 @@ ControllerResult<Map<ConfigResource, ApiError>> incrementalAlterConfigs(
Map<ConfigResource, ApiError> outputResults = new HashMap<>();
for (Entry<ConfigResource, Map<String, Entry<OpType, String>>> resourceEntry :
configChanges.entrySet()) {
- incrementalAlterConfigResource(resourceEntry.getKey(),
+ ApiError apiError = incrementalAlterConfigResource(resourceEntry.getKey(),
resourceEntry.getValue(),
newlyCreatedResource,
- outputRecords,
- outputResults);
+ outputRecords);
+ outputResults.put(resourceEntry.getKey(), apiError);
}
return ControllerResult.atomicOf(outputRecords, outputResults);
}
- private void incrementalAlterConfigResource(ConfigResource configResource,
+ ControllerResult<ApiError> incrementalAlterConfig(
+ ConfigResource configResource, Map<String, Entry<OpType, String>> keyToOps,
+ boolean newlyCreatedResource) {
+ List<ApiMessageAndVersion> outputRecords = new ArrayList<>();
+ ApiError apiError = incrementalAlterConfigResource(configResource,
+ keyToOps,
+ newlyCreatedResource,
+ outputRecords);
+ return ControllerResult.atomicOf(outputRecords, apiError);
+ }
+
+ private ApiError incrementalAlterConfigResource(ConfigResource configResource,
Map<String, Entry<OpType, String>> keysToOps,
boolean newlyCreatedResource,
- List<ApiMessageAndVersion> outputRecords,
- Map<ConfigResource, ApiError> outputResults) {
+ List<ApiMessageAndVersion> outputRecords) {
Review Comment:
Yup. Fixed.
##########
metadata/src/test/java/org/apache/kafka/controller/ReplicationControlManagerTest.java:
##########
@@ -605,6 +617,45 @@ public void testCreateTopicsWithConfigs() throws Exception {
"Null value not supported for topic configs: foo",
result2.response().topics().find("bar").errorMessage()
);
+
+ CreateTopicsRequestData request3 = new CreateTopicsRequestData();
+ request3.topics().add(new CreatableTopic().setName("baz")
+ .setNumPartitions(-1).setReplicationFactor((short) -2)
+ .setConfigs(validConfigs));
+
+ ControllerResult<CreateTopicsResponseData> result3 =
+ replicationControl.createTopics(request3, Collections.singleton("baz"));
+ assertEquals(INVALID_REPLICATION_FACTOR.code(), result3.response().topics().find("baz").errorCode());
+ assertEquals(Collections.emptyList(), result3.records());
+
+ // Test request with multiple topics together.
+ CreateTopicsRequestData request4 = new CreateTopicsRequestData();
+ String batchedTopic1 = "batched-topic-1";
+ request4.topics().add(new CreatableTopic().setName(batchedTopic1)
+ .setNumPartitions(-1).setReplicationFactor((short) -1)
+ .setConfigs(validConfigs));
+ String batchedTopic2 = "batched-topic2";
+ request4.topics().add(new CreatableTopic().setName(batchedTopic2)
+ .setNumPartitions(-1).setReplicationFactor((short) -2)
+ .setConfigs(validConfigs));
+
+ Set<String> request4Topics = new HashSet<>();
+ request4Topics.add(batchedTopic1);
+ request4Topics.add(batchedTopic2);
+ ControllerResult<CreateTopicsResponseData> result4 =
+ replicationControl.createTopics(request4, request4Topics);
+
+ assertEquals(Errors.NONE.code(), result4.response().topics().find(batchedTopic1).errorCode());
+ assertEquals(INVALID_REPLICATION_FACTOR.code(), result4.response().topics().find(batchedTopic2).errorCode());
+
+ assertEquals(3, result4.records().size());
+ assertEquals(TopicRecord.class, result4.records().get(0).message().getClass());
+ TopicRecord batchedTopic1Record = (TopicRecord) result4.records().get(0).message();
+ assertEquals(batchedTopic1, batchedTopic1Record.name());
+ assertEquals(ConfigRecord.class, result4.records().get(1).message().getClass());
+ assertEquals(batchedTopic1, ((ConfigRecord) result4.records().get(1).message()).resourceName());
Review Comment:
I'd argue we already test that sort of deeper validation above in the form of:
```
ctx.replay(result1.records());
assertEquals(
"notNull",
ctx.configurationControl.getConfigs(new ConfigResource(ConfigResource.Type.TOPIC, "foo")).get("foo")
);
```
The testing I'm doing here is more to test we only emit records for topic batchedTopic1. So my thought process was no need to duplicate the more deeper validation.
Thoughts?
##########
metadata/src/main/java/org/apache/kafka/controller/ConfigurationControlManager.java:
##########
@@ -172,20 +172,30 @@ ControllerResult<Map<ConfigResource, ApiError>> incrementalAlterConfigs(
Map<ConfigResource, ApiError> outputResults = new HashMap<>();
for (Entry<ConfigResource, Map<String, Entry<OpType, String>>> resourceEntry :
configChanges.entrySet()) {
- incrementalAlterConfigResource(resourceEntry.getKey(),
+ ApiError apiError = incrementalAlterConfigResource(resourceEntry.getKey(),
resourceEntry.getValue(),
newlyCreatedResource,
- outputRecords,
- outputResults);
+ outputRecords);
+ outputResults.put(resourceEntry.getKey(), apiError);
}
return ControllerResult.atomicOf(outputRecords, outputResults);
}
- private void incrementalAlterConfigResource(ConfigResource configResource,
+ ControllerResult<ApiError> incrementalAlterConfig(
+ ConfigResource configResource, Map<String, Entry<OpType, String>> keyToOps,
Review Comment:
Changed to that. I agree the closing parenthesis on the next line reads a bit better.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org