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 2021/02/01 23:49:31 UTC
[GitHub] [kafka] mjsax opened a new pull request #10020: KAFKA-10689: fix windowed FKJ topology and put checks in assignor to avoid infinite loops
mjsax opened a new pull request #10020:
URL: https://github.com/apache/kafka/pull/10020
Backport of #9568 to `2.6`
Fix infinite loop in assignor when trying to resolve the number of partitions in a topology with a windowed FKJ. Also adds a check to this loop to break out and fail the application if we detect that we are/will be stuck in an infinite loop.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] mjsax commented on a change in pull request #10020: KAFKA-10689: fix windowed FKJ topology and put checks in assignor to avoid infinite loops
Posted by GitBox <gi...@apache.org>.
mjsax commented on a change in pull request #10020:
URL: https://github.com/apache/kafka/pull/10020#discussion_r568222503
##########
File path: streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignorTest.java
##########
@@ -1987,6 +1996,65 @@ public void testUniqueFieldOverflow() {
assertEquals(-128, partitionAssignor.uniqueField());
}
+ @Test
+ public void shouldThrowTaskAssignmentExceptionWhenUnableToResolvePartitionCount() {
+ builder = new CorruptedInternalTopologyBuilder();
+ final InternalStreamsBuilder streamsBuilder = new InternalStreamsBuilder(builder);
+
+ final KStream<String, String> inputTopic = streamsBuilder.stream(singleton("topic1"), new ConsumedInternal<>());
+ final KTable<String, String> inputTable = streamsBuilder.table("topic2", new ConsumedInternal<>(), new MaterializedInternal<>(Materialized.as("store")));
+ inputTopic
+ .groupBy(
+ (k, v) -> k,
+ Grouped.with("GroupName", Serdes.String(), Serdes.String())
+ )
+ .windowedBy(TimeWindows.of(Duration.ofMinutes(10)))
+ .aggregate(
+ () -> "",
+ (k, v, a) -> a + k)
+ .leftJoin(
+ inputTable,
+ v -> v,
+ (x, y) -> x + y
+ );
+ streamsBuilder.buildAndOptimizeTopology();
+
+ configureDefault();
+
+ subscriptions.put("consumer",
+ new Subscription(
+ singletonList("topic"),
+ defaultSubscriptionInfo.encode()
+ ));
+ final Map<String, Assignment> assignments = partitionAssignor.assign(metadata, new GroupSubscription(subscriptions)).groupAssignment();
+ assertThat(AssignmentInfo.decode(assignments.get("consumer").userData()).errCode(),
+ equalTo(AssignorError.INCOMPLETE_SOURCE_TOPIC_METADATA.code()));
Review comment:
The change to the original PR is, tat we expect a different error code here. But I guess it's still ok.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] mjsax merged pull request #10020: KAFKA-10689: fix windowed FKJ topology and put checks in assignor to avoid infinite loops
Posted by GitBox <gi...@apache.org>.
mjsax merged pull request #10020:
URL: https://github.com/apache/kafka/pull/10020
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] mjsax merged pull request #10020: KAFKA-10689: fix windowed FKJ topology and put checks in assignor to avoid infinite loops
Posted by GitBox <gi...@apache.org>.
mjsax merged pull request #10020:
URL: https://github.com/apache/kafka/pull/10020
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] mjsax commented on a change in pull request #10020: KAFKA-10689: fix windowed FKJ topology and put checks in assignor to avoid infinite loops
Posted by GitBox <gi...@apache.org>.
mjsax commented on a change in pull request #10020:
URL: https://github.com/apache/kafka/pull/10020#discussion_r568222503
##########
File path: streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignorTest.java
##########
@@ -1987,6 +1996,65 @@ public void testUniqueFieldOverflow() {
assertEquals(-128, partitionAssignor.uniqueField());
}
+ @Test
+ public void shouldThrowTaskAssignmentExceptionWhenUnableToResolvePartitionCount() {
+ builder = new CorruptedInternalTopologyBuilder();
+ final InternalStreamsBuilder streamsBuilder = new InternalStreamsBuilder(builder);
+
+ final KStream<String, String> inputTopic = streamsBuilder.stream(singleton("topic1"), new ConsumedInternal<>());
+ final KTable<String, String> inputTable = streamsBuilder.table("topic2", new ConsumedInternal<>(), new MaterializedInternal<>(Materialized.as("store")));
+ inputTopic
+ .groupBy(
+ (k, v) -> k,
+ Grouped.with("GroupName", Serdes.String(), Serdes.String())
+ )
+ .windowedBy(TimeWindows.of(Duration.ofMinutes(10)))
+ .aggregate(
+ () -> "",
+ (k, v, a) -> a + k)
+ .leftJoin(
+ inputTable,
+ v -> v,
+ (x, y) -> x + y
+ );
+ streamsBuilder.buildAndOptimizeTopology();
+
+ configureDefault();
+
+ subscriptions.put("consumer",
+ new Subscription(
+ singletonList("topic"),
+ defaultSubscriptionInfo.encode()
+ ));
+ final Map<String, Assignment> assignments = partitionAssignor.assign(metadata, new GroupSubscription(subscriptions)).groupAssignment();
+ assertThat(AssignmentInfo.decode(assignments.get("consumer").userData()).errCode(),
+ equalTo(AssignorError.INCOMPLETE_SOURCE_TOPIC_METADATA.code()));
Review comment:
The change to the original PR is, tat we expect a different error code here. But I guess it's still ok.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ableegoldman commented on a change in pull request #10020: KAFKA-10689: fix windowed FKJ topology and put checks in assignor to avoid infinite loops
Posted by GitBox <gi...@apache.org>.
ableegoldman commented on a change in pull request #10020:
URL: https://github.com/apache/kafka/pull/10020#discussion_r568227868
##########
File path: streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignorTest.java
##########
@@ -1987,6 +1996,65 @@ public void testUniqueFieldOverflow() {
assertEquals(-128, partitionAssignor.uniqueField());
}
+ @Test
+ public void shouldThrowTaskAssignmentExceptionWhenUnableToResolvePartitionCount() {
+ builder = new CorruptedInternalTopologyBuilder();
+ final InternalStreamsBuilder streamsBuilder = new InternalStreamsBuilder(builder);
+
+ final KStream<String, String> inputTopic = streamsBuilder.stream(singleton("topic1"), new ConsumedInternal<>());
+ final KTable<String, String> inputTable = streamsBuilder.table("topic2", new ConsumedInternal<>(), new MaterializedInternal<>(Materialized.as("store")));
+ inputTopic
+ .groupBy(
+ (k, v) -> k,
+ Grouped.with("GroupName", Serdes.String(), Serdes.String())
+ )
+ .windowedBy(TimeWindows.of(Duration.ofMinutes(10)))
+ .aggregate(
+ () -> "",
+ (k, v, a) -> a + k)
+ .leftJoin(
+ inputTable,
+ v -> v,
+ (x, y) -> x + y
+ );
+ streamsBuilder.buildAndOptimizeTopology();
+
+ configureDefault();
+
+ subscriptions.put("consumer",
+ new Subscription(
+ singletonList("topic"),
+ defaultSubscriptionInfo.encode()
+ ));
+ final Map<String, Assignment> assignments = partitionAssignor.assign(metadata, new GroupSubscription(subscriptions)).groupAssignment();
+ assertThat(AssignmentInfo.decode(assignments.get("consumer").userData()).errCode(),
+ equalTo(AssignorError.INCOMPLETE_SOURCE_TOPIC_METADATA.code()));
Review comment:
Yeah it's a little awkward/misleading, but ideally no one should be hitting this exception in the 1st place once we have this fix. So this seems fine for 2.6.2
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ableegoldman commented on a change in pull request #10020: KAFKA-10689: fix windowed FKJ topology and put checks in assignor to avoid infinite loops
Posted by GitBox <gi...@apache.org>.
ableegoldman commented on a change in pull request #10020:
URL: https://github.com/apache/kafka/pull/10020#discussion_r568227868
##########
File path: streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignorTest.java
##########
@@ -1987,6 +1996,65 @@ public void testUniqueFieldOverflow() {
assertEquals(-128, partitionAssignor.uniqueField());
}
+ @Test
+ public void shouldThrowTaskAssignmentExceptionWhenUnableToResolvePartitionCount() {
+ builder = new CorruptedInternalTopologyBuilder();
+ final InternalStreamsBuilder streamsBuilder = new InternalStreamsBuilder(builder);
+
+ final KStream<String, String> inputTopic = streamsBuilder.stream(singleton("topic1"), new ConsumedInternal<>());
+ final KTable<String, String> inputTable = streamsBuilder.table("topic2", new ConsumedInternal<>(), new MaterializedInternal<>(Materialized.as("store")));
+ inputTopic
+ .groupBy(
+ (k, v) -> k,
+ Grouped.with("GroupName", Serdes.String(), Serdes.String())
+ )
+ .windowedBy(TimeWindows.of(Duration.ofMinutes(10)))
+ .aggregate(
+ () -> "",
+ (k, v, a) -> a + k)
+ .leftJoin(
+ inputTable,
+ v -> v,
+ (x, y) -> x + y
+ );
+ streamsBuilder.buildAndOptimizeTopology();
+
+ configureDefault();
+
+ subscriptions.put("consumer",
+ new Subscription(
+ singletonList("topic"),
+ defaultSubscriptionInfo.encode()
+ ));
+ final Map<String, Assignment> assignments = partitionAssignor.assign(metadata, new GroupSubscription(subscriptions)).groupAssignment();
+ assertThat(AssignmentInfo.decode(assignments.get("consumer").userData()).errCode(),
+ equalTo(AssignorError.INCOMPLETE_SOURCE_TOPIC_METADATA.code()));
Review comment:
Yeah it's a little awkward/misleading, but ideally no one should be hitting this exception in the 1st place once we have this fix. So this seems fine for 2.6.2
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org