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:
us...@infra.apache.org
