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