You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "lucasbru (via GitHub)" <gi...@apache.org> on 2024/11/08 12:56:06 UTC
[PR] Topic validation [kafka]
lucasbru opened a new pull request, #17721:
URL: https://github.com/apache/kafka/pull/17721
Finishing topic validation
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
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
Re: [PR] Topology epochs and topic validation [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru merged PR #17721:
URL: https://github.com/apache/kafka/pull/17721
--
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
Re: [PR] Topology epochs and topic validation [kafka]
Posted by "cadonna (via GitHub)" <gi...@apache.org>.
cadonna commented on code in PR #17721:
URL: https://github.com/apache/kafka/pull/17721#discussion_r1866067778
##########
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/InternalTopicIntegrationTest.java:
##########
@@ -73,32 +77,39 @@
@Timeout(600)
@Tag("integration")
public class InternalTopicIntegrationTest {
- public static final EmbeddedKafkaCluster CLUSTER = new EmbeddedKafkaCluster(1);
+
+ public static EmbeddedKafkaCluster cluster = null;
@BeforeAll
public static void startCluster() throws IOException, InterruptedException {
- CLUSTER.start();
- CLUSTER.createTopics(DEFAULT_INPUT_TOPIC, DEFAULT_INPUT_TABLE_TOPIC);
+
+ final Properties props = new Properties();
+ props.setProperty(GroupCoordinatorConfig.GROUP_COORDINATOR_REBALANCE_PROTOCOLS_CONFIG, "classic,consumer,streams");
+ props.setProperty(ServerConfigs.UNSTABLE_API_VERSIONS_ENABLE_CONFIG, "true");
+
+ cluster = new EmbeddedKafkaCluster(1, props);
Review Comment:
Good points!
--
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
Re: [PR] Topology epochs and topic validation [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #17721:
URL: https://github.com/apache/kafka/pull/17721#discussion_r1863351962
##########
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/InternalTopicIntegrationTest.java:
##########
@@ -73,32 +77,39 @@
@Timeout(600)
@Tag("integration")
public class InternalTopicIntegrationTest {
- public static final EmbeddedKafkaCluster CLUSTER = new EmbeddedKafkaCluster(1);
+
+ public static EmbeddedKafkaCluster cluster = null;
@BeforeAll
public static void startCluster() throws IOException, InterruptedException {
- CLUSTER.start();
- CLUSTER.createTopics(DEFAULT_INPUT_TOPIC, DEFAULT_INPUT_TABLE_TOPIC);
+
+ final Properties props = new Properties();
+ props.setProperty(GroupCoordinatorConfig.GROUP_COORDINATOR_REBALANCE_PROTOCOLS_CONFIG, "classic,consumer,streams");
+ props.setProperty(ServerConfigs.UNSTABLE_API_VERSIONS_ENABLE_CONFIG, "true");
+
+ cluster = new EmbeddedKafkaCluster(1, props);
Review Comment:
We need to pass those props into it. I guess we could do it with a `static {` block, but I like this more. We should also not make it final to do a `cluster = null` in @AfterAll, otherwise the cluster will stay in memory for the rest of the test suite, IIUC. Is there any magic in JUnit that unloads test classes after they have been run? As far as I know, there is not. So if we have to use static variables in test classes, we should clear them out in @AfterAll.
--
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
Re: [PR] Topology epochs and topic validation [kafka]
Posted by "cadonna (via GitHub)" <gi...@apache.org>.
cadonna commented on code in PR #17721:
URL: https://github.com/apache/kafka/pull/17721#discussion_r1863242293
##########
clients/src/main/resources/common/message/StreamsGroupHeartbeatResponse.json:
##########
@@ -30,11 +30,11 @@
// - FENCED_MEMBER_EPOCH (version 0+)
// - UNRELEASED_INSTANCE_ID (version 0+)
// - GROUP_MAX_SIZE_REACHED (version 0+)
- // - TOPIC_AUTHORIZATION_FAILED (version 0+)
+ // - TOPIC_AUTHORIZATION_FAILED (version 0+)
// - CLUSTER_AUTHORIZATION_FAILED (version 0+)
// - STREAMS_INVALID_TOPOLOGY (version 0+)
- // - STREAMS_MISSING_SOURCE_TOPICS (version 0+)
- // - STREAMS_INCONSISTENT_INTERNAL_TOPICS (version 0+)
+ // - STREAMS_INVALID_TOPOLOGY_EPOCH (version 0+)
+ // - STREAMS_TOPOLOGY_FENCED (version 0+)
"fields": [
// Same as consumer group heart beat
Review Comment:
That comment lies since `AcceptableRecoveryLag` and `TaskOffsetIntervalMs` are not part of the consumer group heartbeat.
##########
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/InternalTopicIntegrationTest.java:
##########
@@ -73,32 +77,39 @@
@Timeout(600)
@Tag("integration")
public class InternalTopicIntegrationTest {
- public static final EmbeddedKafkaCluster CLUSTER = new EmbeddedKafkaCluster(1);
+
+ public static EmbeddedKafkaCluster cluster = null;
@BeforeAll
public static void startCluster() throws IOException, InterruptedException {
- CLUSTER.start();
- CLUSTER.createTopics(DEFAULT_INPUT_TOPIC, DEFAULT_INPUT_TABLE_TOPIC);
+
+ final Properties props = new Properties();
+ props.setProperty(GroupCoordinatorConfig.GROUP_COORDINATOR_REBALANCE_PROTOCOLS_CONFIG, "classic,consumer,streams");
+ props.setProperty(ServerConfigs.UNSTABLE_API_VERSIONS_ENABLE_CONFIG, "true");
+
+ cluster = new EmbeddedKafkaCluster(1, props);
Review Comment:
What is the difference between creating the embedded cluster here or on line 81?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -2367,6 +2391,14 @@ private CoordinatorResult<StreamsGroupHeartbeatResult, CoordinatorRecord> stream
group.setMetadataRefreshDeadline(currentTimeMs + streamsGroupMetadataRefreshIntervalMs, groupEpoch);
}
+ // If we updated the groupEpoch, we may need to reconfigure the topology
+ ConfiguredTopology configuredTopology = group.configuredTopology();
+ if (bumpGroupEpoch) {
+ log.info("[GroupId {}] Configuring the topology {}", groupId, topology);
+ configuredTopology =
+ InternalTopicManager.configureTopics(logContext, topology, partitionMetadata);
+ }
+
// 2. Update the target assignment if the group epoch is larger than the target assignment epoch or a static member
Review Comment:
The number doesn't fit.
--
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