Repository: apache/kafka
File: streams/src/test/java/org/apache/kafka/streams/processor/internals/InternalTopicManagerTest.java (Java)

Code before:
        final TopicPartitionInfo partitionInfo = new TopicPartitionInfo(0, broker1,
            Collections.singletonList(broker1), Collections.singletonList(broker1));

        final KafkaFutureImpl<TopicDescription> topicDescriptionSuccessFuture = new KafkaFutureImpl<>();
        final KafkaFutureImpl<TopicDescription> topicDescriptionFailFuture = new KafkaFutureImpl<>();
        topicDescriptionSuccessFuture.complete(
            new TopicDescription(topic1, false, Collections.singletonList(partitionInfo), Collections.emptySet())
        );
        topicDescriptionFailFuture.completeExceptionally(new UnknownTopicOrPartitionException("KABOOM!"));

        final KafkaFutureImpl<CreateTopicsResult.TopicMetadataAndConfig> topicCreationFuture = new KafkaFutureImpl<>();
        topicCreationFuture.completeExceptionally(new TopicExistsException("KABOOM!"));

        // let the first describe succeed on topic, and fail on topic2, and then let creation throws topics-existed;
        // it should retry with just topic2 and then let it succeed
        when(admin.describeTopics(Set.of(topic1, topic2)))
                .thenAnswer(answer -> new MockDescribeTopicsResult(mkMap(
                        mkEntry(topic1, topicDescriptionSuccessFuture),
                        mkEntry(topic2, topicDescriptionFailFuture) // first call: missing
                )))
                .thenAnswer(answer -> new MockDescribeTopicsResult(mkMap(
                        mkEntry(topic1, topicDescriptionSuccessFuture),
                        mkEntry(topic2, topicDescriptionSuccessFuture) // second call: now exists
                )));
        when(admin.createTopics(Collections.singleton(new NewTopic(topic2, Optional.of(1), Optional.of((short) 1))
            .configs(mkMap(mkEntry(TopicConfig.CLEANUP_POLICY_CONFIG, TopicConfig.CLEANUP_POLICY_COMPACT),
                                 mkEntry(TopicConfig.MESSAGE_TIMESTAMP_TYPE_CONFIG, "CreateTime"))))))
            .thenAnswer(answer -> new MockCreateTopicsResult(Collections.singletonMap(topic2, topicCreationFuture)));

        final InternalTopicConfig topicConfig = new UnwindowedUnversionedChangelogTopicConfig(topic1, Collections.emptyMap());
        topicConfig.setNumberOfPartitions(1);
        final InternalTopicConfig topic2Config = new UnwindowedUnversionedChangelogTopicConfig(topic2, Collections.emptyMap());
        topic2Config.setNumberOfPartitions(1);
        topicManager.makeReady(mkMap(
            mkEntry(topic1, topicConfig),
            mkEntry(topic2, topic2Config)
        ));
    }

    @Test
    public void shouldNotCreateTopicIfExistsWithDifferentPartitions() {
        mockAdminClient.addTopic(
            false,
            topic1,
            List.of(
                new TopicPartitionInfo(0, broker1, singleReplica, Collections.emptyList()),
                new TopicPartitionInfo(1, broker1, singleReplica, Collections.emptyList())
            ),
            null
        );


Review comment (suggestion):
We should remove the code change from above, and keep this one, because on retry, we should only query for `topic2`.

Code after:
        final TopicPartitionInfo partitionInfo = new TopicPartitionInfo(0, broker1,
            Collections.singletonList(broker1), Collections.singletonList(broker1));

        final KafkaFutureImpl<TopicDescription> topicDescriptionSuccessFuture = new KafkaFutureImpl<>();
        final KafkaFutureImpl<TopicDescription> topicDescriptionFailFuture = new KafkaFutureImpl<>();
        topicDescriptionSuccessFuture.complete(
            new TopicDescription(topic1, false, Collections.singletonList(partitionInfo), Collections.emptySet())
        );
        topicDescriptionFailFuture.completeExceptionally(new UnknownTopicOrPartitionException("KABOOM!"));

        final KafkaFutureImpl<CreateTopicsResult.TopicMetadataAndConfig> topicCreationFuture = new KafkaFutureImpl<>();
        topicCreationFuture.completeExceptionally(new TopicExistsException("KABOOM!"));

        // let the first describe succeed on topic, and fail on topic2, and then let creation throws topics-existed;
        // it should retry with just topic2 and then let it succeed
        when(admin.describeTopics(Set.of(topic1, topic2)))
                .thenAnswer(answer -> new MockDescribeTopicsResult(mkMap(
                        mkEntry(topic1, topicDescriptionSuccessFuture),
                        mkEntry(topic2, topicDescriptionFailFuture) // first call: missing
                )));

        when(admin.createTopics(Collections.singleton(new NewTopic(topic2, Optional.of(1), Optional.of((short) 1))
            .configs(mkMap(mkEntry(TopicConfig.CLEANUP_POLICY_CONFIG, TopicConfig.CLEANUP_POLICY_COMPACT),
                                 mkEntry(TopicConfig.MESSAGE_TIMESTAMP_TYPE_CONFIG, "CreateTime"))))))
            .thenAnswer(answer -> new MockCreateTopicsResult(Collections.singletonMap(topic2, topicCreationFuture)));
        when(admin.describeTopics(Set.of(topic2)))
                .thenAnswer(answer -> new MockDescribeTopicsResult(Collections.singletonMap(topic2, topicDescriptionSuccessFuture)));

        final InternalTopicConfig topicConfig = new UnwindowedUnversionedChangelogTopicConfig(topic1, Collections.emptyMap());
        topicConfig.setNumberOfPartitions(1);
        final InternalTopicConfig topic2Config = new UnwindowedUnversionedChangelogTopicConfig(topic2, Collections.emptyMap());
        topic2Config.setNumberOfPartitions(1);
        topicManager.makeReady(mkMap(
            mkEntry(topic1, topicConfig),
            mkEntry(topic2, topic2Config)
        ));
    }

    @Test
    public void shouldNotCreateTopicIfExistsWithDifferentPartitions() {
        mockAdminClient.addTopic(
            false,
            topic1,
            List.of(
                new TopicPartitionInfo(0, broker1, singleReplica, Collections.emptyList()),
                new TopicPartitionInfo(1, broker1, singleReplica, Collections.emptyList())
            ),
            null
        );

        try {