diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/CreateTopicsResult.java b/clients/src/main/java/org/apache/kafka/clients/admin/CreateTopicsResult.java index 0b75959cd8093..100e996b5d8d0 100644 --- a/clients/src/main/java/org/apache/kafka/clients/admin/CreateTopicsResult.java +++ b/clients/src/main/java/org/apache/kafka/clients/admin/CreateTopicsResult.java @@ -115,7 +115,7 @@ public static class TopicMetadataAndConfig { private final int replicationFactor; private final Config config; - TopicMetadataAndConfig(Uuid topicId, int numPartitions, int replicationFactor, Config config) { + public TopicMetadataAndConfig(Uuid topicId, int numPartitions, int replicationFactor, Config config) { this.exception = null; this.topicId = topicId; this.numPartitions = numPartitions; @@ -123,7 +123,7 @@ public static class TopicMetadataAndConfig { this.config = config; } - TopicMetadataAndConfig(ApiException exception) { + public TopicMetadataAndConfig(ApiException exception) { this.exception = exception; this.topicId = Uuid.ZERO_UUID; this.numPartitions = UNKNOWN; diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/DeleteTopicsResult.java b/clients/src/main/java/org/apache/kafka/clients/admin/DeleteTopicsResult.java index d48c7e00ed4a4..17eba2c2ec421 100644 --- a/clients/src/main/java/org/apache/kafka/clients/admin/DeleteTopicsResult.java +++ b/clients/src/main/java/org/apache/kafka/clients/admin/DeleteTopicsResult.java @@ -32,7 +32,7 @@ public class DeleteTopicsResult { final Map> futures; - DeleteTopicsResult(Map> futures) { + protected DeleteTopicsResult(Map> futures) { this.futures = futures; } diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopicManager.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopicManager.java index 4e2d52fe16de2..2ca130a9dd2f2 100644 --- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopicManager.java +++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopicManager.java @@ -20,6 +20,7 @@ import org.apache.kafka.clients.admin.Config; import org.apache.kafka.clients.admin.ConfigEntry; import org.apache.kafka.clients.admin.CreateTopicsResult; +import org.apache.kafka.clients.admin.DeleteTopicsResult; import org.apache.kafka.clients.admin.DescribeConfigsResult; import org.apache.kafka.clients.admin.DescribeTopicsResult; import org.apache.kafka.clients.admin.NewTopic; @@ -44,6 +45,8 @@ import org.slf4j.Logger; import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collection; import java.util.Collections; import java.util.HashMap; import java.util.HashSet; @@ -186,14 +189,23 @@ public ValidationResult validate(final Map topicCon ); } - maybeThrowTimeoutException(topicDescriptionsStillToValidate, topicConfigsStillToValidate, deadline); + maybeThrowTimeoutException( + Arrays.asList(topicDescriptionsStillToValidate, topicConfigsStillToValidate), + deadline, + String.format("Could not validate internal topics within %d milliseconds. " + + "This can happen if the Kafka cluster is temporarily not available.", retryTimeoutMs) + ); if (!descriptionsForTopic.isEmpty() || !configsForTopic.isEmpty()) { Utils.sleep(100); } } - maybeSleep(topicDescriptionsStillToValidate, topicConfigsStillToValidate, deadline); + maybeSleep( + Arrays.asList(topicDescriptionsStillToValidate, topicConfigsStillToValidate), + deadline, + "validated" + ); } log.info("Completed validation of internal topics {}.", topicConfigs.keySet()); @@ -205,8 +217,7 @@ private void doValidateTopic(final ValidationResult validationResult, final Map topicsConfigs, final Set topicsStillToValidate, final BiConsumer validator) { - for (final InternalTopicConfig topicConfig : topicsConfigs.values()) { - final String topicName = topicConfig.name(); + for (final String topicName : new HashSet<>(topicsStillToValidate)) { if (!futuresForTopic.containsKey(topicName)) { throw new IllegalStateException("Description results do not contain topics to validate. " + BUG_ERROR_MESSAGE); } @@ -214,7 +225,8 @@ private void doValidateTopic(final ValidationResult validationResult, if (future.isDone()) { try { final V brokerSideTopicConfig = future.get(); - validator.accept(topicConfig, brokerSideTopicConfig); + final InternalTopicConfig streamsSideTopicConfig = topicsConfigs.get(topicName); + validator.accept(streamsSideTopicConfig, brokerSideTopicConfig); topicsStillToValidate.remove(topicName); } catch (final ExecutionException executionException) { final Throwable cause = executionException.getCause(); @@ -242,35 +254,6 @@ private void doValidateTopic(final ValidationResult validationResult, } } - private void maybeThrowTimeoutException(final Set topicDescriptionsStillToValidate, - final Set topicConfigsStillToValidate, - final long deadline) { - if (!topicDescriptionsStillToValidate.isEmpty() || !topicConfigsStillToValidate.isEmpty()) { - final long now = time.milliseconds(); - if (now >= deadline) { - final String timeoutError = String.format("Could not validate internal topics within %d milliseconds. " + - "This can happen if the Kafka cluster is temporarily not available.", retryTimeoutMs); - log.error(timeoutError); - throw new TimeoutException(timeoutError); - } - } - } - - private void maybeSleep(final Set topicDescriptionsStillToValidate, - final Set topicConfigsStillToValidate, - final long deadline) { - if (!topicDescriptionsStillToValidate.isEmpty() || !topicConfigsStillToValidate.isEmpty()) { - final long now = time.milliseconds(); - log.info( - "Internal topics {} could not be validated. Will retry in {} milliseconds. Remaining time in milliseconds: {}", - new HashSet<>(topicDescriptionsStillToValidate).addAll(topicConfigsStillToValidate), - retryBackOffMs, - deadline - now - ); - Utils.sleep(retryBackOffMs); - } - } - private void validatePartitionCount(final ValidationResult validationResult, final InternalTopicConfig topicConfig, final TopicDescription topicDescription) { @@ -580,4 +563,219 @@ private Set validateTopics(final Set topicsToValidate, return topicsToCreate; } + + /** + * Sets up internal topics. + * + * Either the given topic are all created or the method fails with an exception. + * + * @param topicConfigs internal topics to setup + */ + public void setup(final Map topicConfigs) { + log.info("Starting to setup internal topics {}.", topicConfigs.keySet()); + + final long now = time.milliseconds(); + final long deadline = now + retryTimeoutMs; + + final Map> streamsSideTopicConfigs = topicConfigs.values().stream() + .collect(Collectors.toMap( + InternalTopicConfig::name, + topicConfig -> topicConfig.getProperties(defaultTopicConfigs, windowChangeLogAdditionalRetention) + )); + final Set createdTopics = new HashSet<>(); + final Set topicStillToCreate = new HashSet<>(topicConfigs.keySet()); + while (!topicStillToCreate.isEmpty()) { + final Set newTopics = topicStillToCreate.stream() + .map(topicName -> new NewTopic( + topicName, + topicConfigs.get(topicName).numberOfPartitions(), + Optional.of(replicationFactor) + ).configs(streamsSideTopicConfigs.get(topicName)) + ).collect(Collectors.toSet()); + + log.info("Going to create internal topics: " + newTopics); + final CreateTopicsResult createTopicsResult = adminClient.createTopics(newTopics); + + processCreateTopicResults(createTopicsResult, topicStillToCreate, createdTopics, deadline); + + maybeSleep(Collections.singletonList(topicStillToCreate), deadline, "created"); + } + + log.info("Completed setup of internal topics {}.", topicConfigs.keySet()); + } + + private void processCreateTopicResults(final CreateTopicsResult createTopicsResult, + final Set topicStillToCreate, + final Set createdTopics, + final long deadline) { + final Map lastErrorsSeenForTopic = new HashMap<>(); + final Map> createResultForTopic = createTopicsResult.values(); + while (!createResultForTopic.isEmpty()) { + for (final String topicName : new HashSet<>(topicStillToCreate)) { + if (!createResultForTopic.containsKey(topicName)) { + cleanUpCreatedTopics(createdTopics); + throw new IllegalStateException("Create topic results do not contain internal topic " + topicName + + " to setup. " + BUG_ERROR_MESSAGE); + } + final KafkaFuture createResult = createResultForTopic.get(topicName); + if (createResult.isDone()) { + try { + createResult.get(); + createdTopics.add(topicName); + topicStillToCreate.remove(topicName); + } catch (final ExecutionException executionException) { + final Throwable cause = executionException.getCause(); + if (cause instanceof TopicExistsException) { + lastErrorsSeenForTopic.put(topicName, cause); + log.info("Internal topic {} already exists. Topic is probably marked for deletion. " + + "Will retry to create this topic later (to let broker complete async delete operation first)", + topicName); + } else if (cause instanceof TimeoutException) { + lastErrorsSeenForTopic.put(topicName, cause); + log.info("Creating internal topic {} timed out.", topicName); + } else { + cleanUpCreatedTopics(createdTopics); + log.error("Unexpected error during creation of internal topic: ", cause); + throw new StreamsException( + String.format("Could not create internal topic %s for the following reason: ", topicName), + cause + ); + } + } catch (final InterruptedException interruptedException) { + throw new InterruptException(interruptedException); + } finally { + createResultForTopic.remove(topicName); + } + } + } + + maybeThrowTimeoutExceptionDuringSetup( + topicStillToCreate, + createdTopics, + lastErrorsSeenForTopic, + deadline + ); + + if (!createResultForTopic.isEmpty()) { + Utils.sleep(100); + } + } + } + + private void cleanUpCreatedTopics(final Set topicsToCleanUp) { + log.info("Starting to clean up internal topics {}.", topicsToCleanUp); + + final long now = time.milliseconds(); + final long deadline = now + retryTimeoutMs; + + final Set topicsStillToCleanup = new HashSet<>(topicsToCleanUp); + while (!topicsStillToCleanup.isEmpty()) { + log.info("Going to cleanup internal topics: " + topicsStillToCleanup); + final DeleteTopicsResult deleteTopicsResult = adminClient.deleteTopics(topicsStillToCleanup); + final Map> deleteResultForTopic = deleteTopicsResult.values(); + while (!deleteResultForTopic.isEmpty()) { + for (final String topicName : new HashSet<>(topicsStillToCleanup)) { + if (!deleteResultForTopic.containsKey(topicName)) { + throw new IllegalStateException("Delete topic results do not contain internal topic " + topicName + + " to clean up. " + BUG_ERROR_MESSAGE); + } + final KafkaFuture deleteResult = deleteResultForTopic.get(topicName); + if (deleteResult.isDone()) { + try { + deleteResult.get(); + topicsStillToCleanup.remove(topicName); + } catch (final ExecutionException executionException) { + final Throwable cause = executionException.getCause(); + if (cause instanceof UnknownTopicOrPartitionException) { + log.info("Internal topic {} to clean up is missing", topicName); + } else if (cause instanceof LeaderNotAvailableException) { + log.info("The leader of internal topic {} to clean up is not available.", topicName); + } else if (cause instanceof TimeoutException) { + log.info("Cleaning up internal topic {} timed out.", topicName); + } else { + log.error("Unexpected error during cleanup of internal topics: ", cause); + throw new StreamsException( + String.format("Could not clean up internal topics %s, because during the cleanup " + + "of topic %s the following error occurred: ", + topicsStillToCleanup, topicName), + cause + ); + } + } catch (final InterruptedException interruptedException) { + throw new InterruptException(interruptedException); + } finally { + deleteResultForTopic.remove(topicName); + } + } + } + + maybeThrowTimeoutException( + Collections.singletonList(topicsStillToCleanup), + deadline, + String.format("Could not cleanup internal topics within %d milliseconds. This can happen if the " + + "Kafka cluster is temporarily not available or the broker did not complete topic creation " + + "before the cleanup. The following internal topics could not be cleaned up: %s", + retryTimeoutMs, topicsStillToCleanup) + ); + + if (!deleteResultForTopic.isEmpty()) { + Utils.sleep(100); + } + } + + maybeSleep( + Collections.singletonList(topicsStillToCleanup), + deadline, + "validated" + ); + } + + log.info("Completed cleanup of internal topics {}.", topicsToCleanUp); + } + + private void maybeThrowTimeoutException(final List> topicStillToProcess, + final long deadline, + final String errorMessage) { + if (topicStillToProcess.stream().anyMatch(resultSet -> !resultSet.isEmpty())) { + final long now = time.milliseconds(); + if (now >= deadline) { + log.error(errorMessage); + throw new TimeoutException(errorMessage); + } + } + } + + private void maybeThrowTimeoutExceptionDuringSetup(final Set topicStillToProcess, + final Set createdTopics, + final Map lastErrorsSeenForTopic, + final long deadline) { + if (topicStillToProcess.stream().anyMatch(resultSet -> !resultSet.isEmpty())) { + final long now = time.milliseconds(); + if (now >= deadline) { + cleanUpCreatedTopics(createdTopics); + final String errorMessage = String.format("Could not create internal topics within %d milliseconds. This can happen if the " + + "Kafka cluster is temporarily not available or a topic is marked for deletion and the broker " + + "did not complete its deletion within the timeout. The last errors seen per topic are: %s", + retryTimeoutMs, lastErrorsSeenForTopic); + log.error(errorMessage); + throw new TimeoutException(errorMessage); + } + } + } + + private void maybeSleep(final List> resultSetsStillToValidate, + final long deadline, + final String action) { + if (resultSetsStillToValidate.stream().anyMatch(resultSet -> !resultSet.isEmpty())) { + final long now = time.milliseconds(); + log.info( + "Internal topics {} could not be {}. Will retry in {} milliseconds. Remaining time in milliseconds: {}", + resultSetsStillToValidate.stream().flatMap(Collection::stream).collect(Collectors.toSet()), + action, + retryBackOffMs, + deadline - now + ); + Utils.sleep(retryBackOffMs); + } + } } diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/InternalTopicManagerTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/InternalTopicManagerTest.java index 16f5a0bb7f628..ead5a28c89f1e 100644 --- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/InternalTopicManagerTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/InternalTopicManagerTest.java @@ -20,6 +20,8 @@ import org.apache.kafka.clients.admin.Config; import org.apache.kafka.clients.admin.ConfigEntry; import org.apache.kafka.clients.admin.CreateTopicsResult; +import org.apache.kafka.clients.admin.CreateTopicsResult.TopicMetadataAndConfig; +import org.apache.kafka.clients.admin.DeleteTopicsResult; import org.apache.kafka.clients.admin.DescribeConfigsResult; import org.apache.kafka.clients.admin.DescribeTopicsResult; import org.apache.kafka.clients.admin.MockAdminClient; @@ -30,6 +32,7 @@ import org.apache.kafka.common.KafkaFuture; import org.apache.kafka.common.Node; import org.apache.kafka.common.TopicPartitionInfo; +import org.apache.kafka.common.Uuid; import org.apache.kafka.common.config.ConfigResource; import org.apache.kafka.common.config.ConfigResource.Type; import org.apache.kafka.common.config.TopicConfig; @@ -123,6 +126,426 @@ public void shutdown() { mockAdminClient.close(); } + @Test + public void shouldCreateTopics() throws Exception { + final InternalTopicConfig internalTopicConfig1 = setupRepartitionTopicConfig(topic1, 1); + final InternalTopicConfig internalTopicConfig2 = setupRepartitionTopicConfig(topic2, 1); + + internalTopicManager.setup(mkMap( + mkEntry(topic1, internalTopicConfig1), + mkEntry(topic2, internalTopicConfig2) + )); + + final Set newlyCreatedTopics = mockAdminClient.listTopics().names().get(); + assertThat(newlyCreatedTopics.size(), is(2)); + assertThat(newlyCreatedTopics, hasItem(topic1)); + assertThat(newlyCreatedTopics, hasItem(topic2)); + } + + @Test + public void shouldNotCreateTopicsWithEmptyInput() throws Exception { + + internalTopicManager.setup(Collections.emptyMap()); + + final Set newlyCreatedTopics = mockAdminClient.listTopics().names().get(); + assertThat(newlyCreatedTopics, empty()); + } + + @Test + public void shouldOnlyRetryNotSuccessfulFuturesDuringSetup() { + final AdminClient admin = EasyMock.createNiceMock(AdminClient.class); + final StreamsConfig streamsConfig = new StreamsConfig(config); + final InternalTopicManager topicManager = new InternalTopicManager(Time.SYSTEM, admin, streamsConfig); + final KafkaFutureImpl createTopicFailFuture = new KafkaFutureImpl<>(); + createTopicFailFuture.completeExceptionally(new TopicExistsException("exists")); + final KafkaFutureImpl createTopicSuccessfulFuture = new KafkaFutureImpl<>(); + createTopicSuccessfulFuture.complete( + new TopicMetadataAndConfig(Uuid.randomUuid(), 1, 1, new Config(Collections.emptyList())) + ); + final InternalTopicConfig internalTopicConfig1 = setupRepartitionTopicConfig(topic1, 1); + final InternalTopicConfig internalTopicConfig2 = setupRepartitionTopicConfig(topic2, 1); + final NewTopic newTopic1 = newTopic(topic1, internalTopicConfig1, streamsConfig); + final NewTopic newTopic2 = newTopic(topic2, internalTopicConfig2, streamsConfig); + EasyMock.expect(admin.createTopics(mkSet(newTopic1, newTopic2))) + .andAnswer(() -> new MockCreateTopicsResult(mkMap( + mkEntry(topic1, createTopicSuccessfulFuture), + mkEntry(topic2, createTopicFailFuture) + ))); + EasyMock.expect(admin.createTopics(mkSet(newTopic2))) + .andAnswer(() -> new MockCreateTopicsResult(mkMap( + mkEntry(topic2, createTopicSuccessfulFuture) + ))); + EasyMock.replay(admin); + + topicManager.setup(mkMap( + mkEntry(topic1, internalTopicConfig1), + mkEntry(topic2, internalTopicConfig2) + )); + + EasyMock.verify(admin); + } + + @Test + public void shouldRetryCreateTopicWhenCreationTimesOut() { + shouldRetryCreateTopicWhenRetriableExceptionIsThrown(new TimeoutException("timed out")); + } + + @Test + public void shouldRetryCreateTopicWhenTopicNotYetDeleted() { + shouldRetryCreateTopicWhenRetriableExceptionIsThrown(new TopicExistsException("exists")); + } + + private void shouldRetryCreateTopicWhenRetriableExceptionIsThrown(final Exception retriableException) { + final AdminClient admin = EasyMock.createNiceMock(AdminClient.class); + final StreamsConfig streamsConfig = new StreamsConfig(config); + final InternalTopicManager topicManager = new InternalTopicManager(Time.SYSTEM, admin, streamsConfig); + final KafkaFutureImpl createTopicFailFuture = new KafkaFutureImpl<>(); + createTopicFailFuture.completeExceptionally(retriableException); + final KafkaFutureImpl createTopicSuccessfulFuture = new KafkaFutureImpl<>(); + createTopicSuccessfulFuture.complete( + new TopicMetadataAndConfig(Uuid.randomUuid(), 1, 1, new Config(Collections.emptyList())) + ); + final InternalTopicConfig internalTopicConfig = setupRepartitionTopicConfig(topic1, 1); + final NewTopic newTopic = newTopic(topic1, internalTopicConfig, streamsConfig); + EasyMock.expect(admin.createTopics(mkSet(newTopic))) + .andAnswer(() -> new MockCreateTopicsResult(mkMap( + mkEntry(topic1, createTopicSuccessfulFuture) + ))); + EasyMock.expect(admin.createTopics(mkSet(newTopic))) + .andAnswer(() -> new MockCreateTopicsResult(mkMap( + mkEntry(topic2, createTopicSuccessfulFuture) + ))); + EasyMock.replay(admin); + + topicManager.setup(mkMap( + mkEntry(topic1, internalTopicConfig) + )); + } + + @Test + public void shouldThrowTimeoutExceptionIfTopicExistsDuringSetup() { + setupTopicInMockAdminClient(topic1, Collections.emptyMap()); + final InternalTopicConfig internalTopicConfig = setupRepartitionTopicConfig(topic1, 1); + + final TimeoutException exception = assertThrows( + TimeoutException.class, + () -> internalTopicManager.setup(Collections.singletonMap(topic1, internalTopicConfig)) + ); + + assertThat( + exception.getMessage(), + is("Could not create internal topics within " + + (Integer) config.get(StreamsConfig.consumerPrefix(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG)) / 2 + + " milliseconds. This can happen if the Kafka cluster is temporarily not available or a topic is marked" + + " for deletion and the broker did not complete its deletion within the timeout." + + " The last errors seen per topic are:" + + " {" + topic1 + "=org.apache.kafka.common.errors.TopicExistsException: Topic test_topic exists already.}") + ); + } + + @Test + public void shouldThrowWhenCreateTopicsThrowsUnexpectedException() { + final AdminClient admin = EasyMock.createNiceMock(AdminClient.class); + final StreamsConfig streamsConfig = new StreamsConfig(config); + final InternalTopicManager topicManager = new InternalTopicManager(Time.SYSTEM, admin, streamsConfig); + final InternalTopicConfig internalTopicConfig = setupRepartitionTopicConfig(topic1, 1); + final KafkaFutureImpl createTopicFailFuture = new KafkaFutureImpl<>(); + createTopicFailFuture.completeExceptionally(new IllegalStateException("Nobody expects the Spanish inquisition")); + final NewTopic newTopic = newTopic(topic1, internalTopicConfig, streamsConfig); + EasyMock.expect(admin.createTopics(mkSet(newTopic))) + .andStubAnswer(() -> new MockCreateTopicsResult(mkMap( + mkEntry(topic1, createTopicFailFuture) + ))); + EasyMock.replay(admin); + + assertThrows(StreamsException.class, () -> topicManager.setup(mkMap( + mkEntry(topic1, internalTopicConfig) + ))); + } + + @Test + public void shouldThrowWhenCreateTopicsResultsDoNotContainTopic() { + final AdminClient admin = EasyMock.createNiceMock(AdminClient.class); + final StreamsConfig streamsConfig = new StreamsConfig(config); + final InternalTopicManager topicManager = new InternalTopicManager(Time.SYSTEM, admin, streamsConfig); + final InternalTopicConfig internalTopicConfig = setupRepartitionTopicConfig(topic1, 1); + final NewTopic newTopic = newTopic(topic1, internalTopicConfig, streamsConfig); + EasyMock.expect(admin.createTopics(mkSet(newTopic))) + .andStubAnswer(() -> new MockCreateTopicsResult(Collections.singletonMap(topic2, new KafkaFutureImpl<>()))); + EasyMock.replay(admin); + + assertThrows( + IllegalStateException.class, + () -> topicManager.setup(Collections.singletonMap(topic1, internalTopicConfig)) + ); + } + + @Test + public void shouldThrowTimeoutExceptionWhenCreateTopicExceedsTimeout() { + final AdminClient admin = EasyMock.createNiceMock(AdminClient.class); + final MockTime time = new MockTime( + (Integer) config.get(StreamsConfig.consumerPrefix(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG)) / 3 + ); + final StreamsConfig streamsConfig = new StreamsConfig(config); + final InternalTopicManager topicManager = new InternalTopicManager(time, admin, streamsConfig); + final KafkaFutureImpl createTopicFailFuture = new KafkaFutureImpl<>(); + createTopicFailFuture.completeExceptionally(new TimeoutException()); + final InternalTopicConfig internalTopicConfig = setupRepartitionTopicConfig(topic1, 1); + final NewTopic newTopic = newTopic(topic1, internalTopicConfig, streamsConfig); + EasyMock.expect(admin.createTopics(mkSet(newTopic))) + .andStubAnswer(() -> new MockCreateTopicsResult(mkMap(mkEntry(topic1, createTopicFailFuture)))); + EasyMock.replay(admin); + + assertThrows( + TimeoutException.class, + () -> topicManager.setup(Collections.singletonMap(topic1, internalTopicConfig)) + ); + } + + @Test + public void shouldThrowTimeoutExceptionWhenFuturesNeverCompleteDuringSetup() { + final AdminClient admin = EasyMock.createNiceMock(AdminClient.class); + final MockTime time = new MockTime( + (Integer) config.get(StreamsConfig.consumerPrefix(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG)) / 3 + ); + final StreamsConfig streamsConfig = new StreamsConfig(config); + final InternalTopicManager topicManager = new InternalTopicManager(time, admin, streamsConfig); + final KafkaFutureImpl createTopicFutureThatNeverCompletes = new KafkaFutureImpl<>(); + final InternalTopicConfig internalTopicConfig = setupRepartitionTopicConfig(topic1, 1); + final NewTopic newTopic = newTopic(topic1, internalTopicConfig, streamsConfig); + EasyMock.expect(admin.createTopics(mkSet(newTopic))) + .andStubAnswer(() -> new MockCreateTopicsResult(mkMap(mkEntry(topic1, createTopicFutureThatNeverCompletes)))); + EasyMock.replay(admin); + + assertThrows( + TimeoutException.class, + () -> topicManager.setup(Collections.singletonMap(topic1, internalTopicConfig)) + ); + } + + @Test + public void shouldCleanUpWhenUnexpectedExceptionIsThrownDuringSetup() { + final AdminClient admin = EasyMock.createNiceMock(AdminClient.class); + final StreamsConfig streamsConfig = new StreamsConfig(config); + final MockTime time = new MockTime( + (Integer) config.get(StreamsConfig.consumerPrefix(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG)) / 3 + ); + final InternalTopicManager topicManager = new InternalTopicManager(time, admin, streamsConfig); + final InternalTopicConfig internalTopicConfig1 = setupRepartitionTopicConfig(topic1, 1); + final InternalTopicConfig internalTopicConfig2 = setupRepartitionTopicConfig(topic2, 1); + setupCleanUpScenario(admin, streamsConfig, internalTopicConfig1, internalTopicConfig2); + final KafkaFutureImpl deleteTopicSuccessfulFuture = new KafkaFutureImpl<>(); + deleteTopicSuccessfulFuture.complete(null); + EasyMock.expect(admin.deleteTopics(mkSet(topic1))) + .andAnswer(() -> new MockDeleteTopicsResult(mkMap(mkEntry(topic1, deleteTopicSuccessfulFuture)))); + EasyMock.replay(admin); + + assertThrows( + StreamsException.class, + () -> topicManager.setup(mkMap( + mkEntry(topic1, internalTopicConfig1), + mkEntry(topic2, internalTopicConfig2) + )) + ); + + EasyMock.verify(admin); + } + + @Test + public void shouldCleanUpWhenCreateTopicsResultsDoNotContainTopic() { + final AdminClient admin = EasyMock.createNiceMock(AdminClient.class); + final StreamsConfig streamsConfig = new StreamsConfig(config); + final InternalTopicManager topicManager = new InternalTopicManager(Time.SYSTEM, admin, streamsConfig); + final InternalTopicConfig internalTopicConfig1 = setupRepartitionTopicConfig(topic1, 1); + final InternalTopicConfig internalTopicConfig2 = setupRepartitionTopicConfig(topic2, 1); + final KafkaFutureImpl createTopicFailFuture1 = new KafkaFutureImpl<>(); + createTopicFailFuture1.completeExceptionally(new TopicExistsException("exists")); + final KafkaFutureImpl createTopicSuccessfulFuture = new KafkaFutureImpl<>(); + createTopicSuccessfulFuture.complete( + new TopicMetadataAndConfig(Uuid.randomUuid(), 1, 1, new Config(Collections.emptyList())) + ); + final NewTopic newTopic1 = newTopic(topic1, internalTopicConfig1, streamsConfig); + final NewTopic newTopic2 = newTopic(topic2, internalTopicConfig2, streamsConfig); + EasyMock.expect(admin.createTopics(mkSet(newTopic1, newTopic2))) + .andAnswer(() -> new MockCreateTopicsResult(mkMap( + mkEntry(topic1, createTopicSuccessfulFuture), + mkEntry(topic2, createTopicFailFuture1) + ))); + EasyMock.expect(admin.createTopics(mkSet(newTopic2))) + .andAnswer(() -> new MockCreateTopicsResult(mkMap( + mkEntry(topic3, createTopicSuccessfulFuture) + ))); + final KafkaFutureImpl deleteTopicSuccessfulFuture = new KafkaFutureImpl<>(); + deleteTopicSuccessfulFuture.complete(null); + EasyMock.expect(admin.deleteTopics(mkSet(topic1))) + .andAnswer(() -> new MockDeleteTopicsResult(mkMap(mkEntry(topic1, deleteTopicSuccessfulFuture)))); + EasyMock.replay(admin); + + assertThrows( + IllegalStateException.class, + () -> topicManager.setup(mkMap( + mkEntry(topic1, internalTopicConfig1), + mkEntry(topic2, internalTopicConfig2) + )) + ); + + EasyMock.verify(admin); + } + + @Test + public void shouldCleanUpWhenCreateTopicsTimesOut() { + final AdminClient admin = EasyMock.createNiceMock(AdminClient.class); + final StreamsConfig streamsConfig = new StreamsConfig(config); + final MockTime time = new MockTime( + (Integer) config.get(StreamsConfig.consumerPrefix(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG)) / 3 + ); + final InternalTopicManager topicManager = new InternalTopicManager(time, admin, streamsConfig); + final InternalTopicConfig internalTopicConfig1 = setupRepartitionTopicConfig(topic1, 1); + final InternalTopicConfig internalTopicConfig2 = setupRepartitionTopicConfig(topic2, 1); + final KafkaFutureImpl createTopicFailFuture1 = new KafkaFutureImpl<>(); + createTopicFailFuture1.completeExceptionally(new TopicExistsException("exists")); + final KafkaFutureImpl createTopicSuccessfulFuture = new KafkaFutureImpl<>(); + createTopicSuccessfulFuture.complete( + new TopicMetadataAndConfig(Uuid.randomUuid(), 1, 1, new Config(Collections.emptyList())) + ); + final NewTopic newTopic1 = newTopic(topic1, internalTopicConfig1, streamsConfig); + final NewTopic newTopic2 = newTopic(topic2, internalTopicConfig2, streamsConfig); + EasyMock.expect(admin.createTopics(mkSet(newTopic1, newTopic2))) + .andAnswer(() -> new MockCreateTopicsResult(mkMap( + mkEntry(topic1, createTopicSuccessfulFuture), + mkEntry(topic2, createTopicFailFuture1) + ))); + final KafkaFutureImpl createTopicFutureThatNeverCompletes = new KafkaFutureImpl<>(); + EasyMock.expect(admin.createTopics(mkSet(newTopic2))) + .andStubAnswer(() -> new MockCreateTopicsResult(mkMap(mkEntry(topic2, createTopicFutureThatNeverCompletes)))); + final KafkaFutureImpl deleteTopicSuccessfulFuture = new KafkaFutureImpl<>(); + deleteTopicSuccessfulFuture.complete(null); + EasyMock.expect(admin.deleteTopics(mkSet(topic1))) + .andAnswer(() -> new MockDeleteTopicsResult(mkMap(mkEntry(topic1, deleteTopicSuccessfulFuture)))); + EasyMock.replay(admin); + + assertThrows( + TimeoutException.class, + () -> topicManager.setup(mkMap( + mkEntry(topic1, internalTopicConfig1), + mkEntry(topic2, internalTopicConfig2) + )) + ); + + EasyMock.verify(admin); + } + + @Test + public void shouldRetryDeleteTopicWhenTopicUnknown() { + shouldRetryDeleteTopicWhenRetriableException(new UnknownTopicOrPartitionException()); + } + + @Test + public void shouldRetryDeleteTopicWhenLeaderNotAvailable() { + shouldRetryDeleteTopicWhenRetriableException(new LeaderNotAvailableException("leader not available")); + } + + @Test + public void shouldRetryDeleteTopicWhenFutureTimesOut() { + shouldRetryDeleteTopicWhenRetriableException(new TimeoutException("timed out")); + } + + private void shouldRetryDeleteTopicWhenRetriableException(final Exception retriableException) { + final AdminClient admin = EasyMock.createNiceMock(AdminClient.class); + final StreamsConfig streamsConfig = new StreamsConfig(config); + final InternalTopicManager topicManager = new InternalTopicManager(Time.SYSTEM, admin, streamsConfig); + final InternalTopicConfig internalTopicConfig1 = setupRepartitionTopicConfig(topic1, 1); + final InternalTopicConfig internalTopicConfig2 = setupRepartitionTopicConfig(topic2, 1); + setupCleanUpScenario(admin, streamsConfig, internalTopicConfig1, internalTopicConfig2); + final KafkaFutureImpl deleteTopicFailFuture = new KafkaFutureImpl<>(); + deleteTopicFailFuture.completeExceptionally(retriableException); + final KafkaFutureImpl deleteTopicSuccessfulFuture = new KafkaFutureImpl<>(); + deleteTopicSuccessfulFuture.complete(null); + EasyMock.expect(admin.deleteTopics(mkSet(topic1))) + .andAnswer(() -> new MockDeleteTopicsResult(mkMap(mkEntry(topic1, deleteTopicFailFuture)))) + .andAnswer(() -> new MockDeleteTopicsResult(mkMap(mkEntry(topic1, deleteTopicSuccessfulFuture)))); + EasyMock.replay(admin); + + assertThrows( + StreamsException.class, + () -> topicManager.setup(mkMap( + mkEntry(topic1, internalTopicConfig1), + mkEntry(topic2, internalTopicConfig2) + )) + ); + EasyMock.verify(); + } + + @Test + public void shouldThrowTimeoutExceptionWhenFuturesNeverCompleteDuringCleanUp() { + final AdminClient admin = EasyMock.createNiceMock(AdminClient.class); + final StreamsConfig streamsConfig = new StreamsConfig(config); + final MockTime time = new MockTime( + (Integer) config.get(StreamsConfig.consumerPrefix(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG)) / 3 + ); + final InternalTopicManager topicManager = new InternalTopicManager(time, admin, streamsConfig); + final InternalTopicConfig internalTopicConfig1 = setupRepartitionTopicConfig(topic1, 1); + final InternalTopicConfig internalTopicConfig2 = setupRepartitionTopicConfig(topic2, 1); + setupCleanUpScenario(admin, streamsConfig, internalTopicConfig1, internalTopicConfig2); + final KafkaFutureImpl deleteTopicFutureThatNeverCompletes = new KafkaFutureImpl<>(); + EasyMock.expect(admin.deleteTopics(mkSet(topic1))) + .andStubAnswer(() -> new MockDeleteTopicsResult(mkMap(mkEntry(topic1, deleteTopicFutureThatNeverCompletes)))); + EasyMock.replay(admin); + + assertThrows( + TimeoutException.class, + () -> topicManager.setup(mkMap( + mkEntry(topic1, internalTopicConfig1), + mkEntry(topic2, internalTopicConfig2) + )) + ); + } + + @Test + public void shouldThrowWhenDeleteTopicsThrowsUnexpectedException() { + final AdminClient admin = EasyMock.createNiceMock(AdminClient.class); + final StreamsConfig streamsConfig = new StreamsConfig(config); + final InternalTopicManager topicManager = new InternalTopicManager(Time.SYSTEM, admin, streamsConfig); + final InternalTopicConfig internalTopicConfig1 = setupRepartitionTopicConfig(topic1, 1); + final InternalTopicConfig internalTopicConfig2 = setupRepartitionTopicConfig(topic2, 1); + setupCleanUpScenario(admin, streamsConfig, internalTopicConfig1, internalTopicConfig2); + final KafkaFutureImpl deleteTopicFailFuture = new KafkaFutureImpl<>(); + deleteTopicFailFuture.completeExceptionally(new IllegalStateException("Nobody expects the Spanish inquisition")); + EasyMock.expect(admin.deleteTopics(mkSet(topic1))) + .andStubAnswer(() -> new MockDeleteTopicsResult(mkMap(mkEntry(topic1, deleteTopicFailFuture)))); + EasyMock.replay(admin); + + assertThrows( + StreamsException.class, + () -> topicManager.setup(mkMap( + mkEntry(topic1, internalTopicConfig1), + mkEntry(topic2, internalTopicConfig2) + )) + ); + } + + private void setupCleanUpScenario(final AdminClient admin, final StreamsConfig streamsConfig, final InternalTopicConfig internalTopicConfig1, final InternalTopicConfig internalTopicConfig2) { + final KafkaFutureImpl createTopicFailFuture1 = new KafkaFutureImpl<>(); + createTopicFailFuture1.completeExceptionally(new TopicExistsException("exists")); + final KafkaFutureImpl createTopicFailFuture2 = new KafkaFutureImpl<>(); + createTopicFailFuture2.completeExceptionally(new IllegalStateException("Nobody expects the Spanish inquisition")); + final KafkaFutureImpl createTopicSuccessfulFuture = new KafkaFutureImpl<>(); + createTopicSuccessfulFuture.complete( + new TopicMetadataAndConfig(Uuid.randomUuid(), 1, 1, new Config(Collections.emptyList())) + ); + final NewTopic newTopic1 = newTopic(topic1, internalTopicConfig1, streamsConfig); + final NewTopic newTopic2 = newTopic(topic2, internalTopicConfig2, streamsConfig); + EasyMock.expect(admin.createTopics(mkSet(newTopic1, newTopic2))) + .andAnswer(() -> new MockCreateTopicsResult(mkMap( + mkEntry(topic1, createTopicSuccessfulFuture), + mkEntry(topic2, createTopicFailFuture1) + ))); + EasyMock.expect(admin.createTopics(mkSet(newTopic2))) + .andAnswer(() -> new MockCreateTopicsResult(mkMap( + mkEntry(topic2, createTopicFailFuture2) + ))); + } + @Test public void shouldReturnCorrectPartitionCounts() { mockAdminClient.addTopic( @@ -653,56 +1076,22 @@ public void shouldReportMisconfigurationsOfCleanupPolicyForWindowedChangelogTopi @Test public void shouldReportMisconfigurationsOfCleanupPolicyForRepartitionTopics() { final long retentionMs = 1000; - mockAdminClient.addTopic( - false, - topic1, - Collections.singletonList(new TopicPartitionInfo(0, broker1, cluster, Collections.emptyList())), - mkMap( - mkEntry(TopicConfig.CLEANUP_POLICY_CONFIG, TopicConfig.CLEANUP_POLICY_DELETE), - mkEntry(TopicConfig.RETENTION_MS_CONFIG, String.valueOf(-1)), - mkEntry(TopicConfig.RETENTION_BYTES_CONFIG, null) - ) - ); - mockAdminClient.addTopic( - false, - topic2, - Collections.singletonList(new TopicPartitionInfo(0, broker1, cluster, Collections.emptyList())), - mkMap( - mkEntry(TopicConfig.CLEANUP_POLICY_CONFIG, TopicConfig.CLEANUP_POLICY_COMPACT), - mkEntry(TopicConfig.RETENTION_MS_CONFIG, String.valueOf(-1)), - mkEntry(TopicConfig.RETENTION_BYTES_CONFIG, null) - ) - ); - mockAdminClient.addTopic( - false, - topic3, - Collections.singletonList(new TopicPartitionInfo(0, broker1, cluster, Collections.emptyList())), - mkMap( - mkEntry(TopicConfig.CLEANUP_POLICY_CONFIG, TopicConfig.CLEANUP_POLICY_COMPACT + "," + TopicConfig.CLEANUP_POLICY_DELETE), - mkEntry(TopicConfig.RETENTION_MS_CONFIG, String.valueOf(-1)), - mkEntry(TopicConfig.RETENTION_BYTES_CONFIG, null) - ) - ); - mockAdminClient.addTopic( - false, - topic4, - Collections.singletonList(new TopicPartitionInfo(0, broker1, cluster, Collections.emptyList())), - mkMap( - mkEntry(TopicConfig.CLEANUP_POLICY_CONFIG, TopicConfig.CLEANUP_POLICY_DELETE), - mkEntry(TopicConfig.RETENTION_MS_CONFIG, String.valueOf(retentionMs)), - mkEntry(TopicConfig.RETENTION_BYTES_CONFIG, null) - ) - ); - mockAdminClient.addTopic( - false, - topic5, - Collections.singletonList(new TopicPartitionInfo(0, broker1, cluster, Collections.emptyList())), - mkMap( - mkEntry(TopicConfig.CLEANUP_POLICY_CONFIG, TopicConfig.CLEANUP_POLICY_DELETE), - mkEntry(TopicConfig.RETENTION_MS_CONFIG, String.valueOf(-1)), - mkEntry(TopicConfig.RETENTION_BYTES_CONFIG, "1024") - ) + setupTopicInMockAdminClient(topic1, repartitionTopicConfig()); + final Map repartitionTopicConfigCleanupPolicyCompact = repartitionTopicConfig(); + repartitionTopicConfigCleanupPolicyCompact.put(TopicConfig.CLEANUP_POLICY_CONFIG, TopicConfig.CLEANUP_POLICY_COMPACT); + setupTopicInMockAdminClient(topic2, repartitionTopicConfigCleanupPolicyCompact); + final Map repartitionTopicConfigCleanupPolicyCompactAndDelete = repartitionTopicConfig(); + repartitionTopicConfigCleanupPolicyCompactAndDelete.put( + TopicConfig.CLEANUP_POLICY_CONFIG, + TopicConfig.CLEANUP_POLICY_COMPACT + "," + TopicConfig.CLEANUP_POLICY_DELETE ); + setupTopicInMockAdminClient(topic3, repartitionTopicConfigCleanupPolicyCompactAndDelete); + final Map repartitionTopicConfigWithFiniteRetentionMs = repartitionTopicConfig(); + repartitionTopicConfigWithFiniteRetentionMs.put(TopicConfig.RETENTION_MS_CONFIG, String.valueOf(retentionMs)); + setupTopicInMockAdminClient(topic4, repartitionTopicConfigWithFiniteRetentionMs); + final Map repartitionTopicConfigWithRetentionBytesSet = repartitionTopicConfig(); + repartitionTopicConfigWithRetentionBytesSet.put(TopicConfig.RETENTION_BYTES_CONFIG, "1024"); + setupTopicInMockAdminClient(topic5, repartitionTopicConfigWithRetentionBytesSet); final InternalTopicConfig internalTopicConfig1 = setupRepartitionTopicConfig(topic1, 1); final InternalTopicConfig internalTopicConfig2 = setupRepartitionTopicConfig(topic2, 1); final InternalTopicConfig internalTopicConfig3 = setupRepartitionTopicConfig(topic3, 1); @@ -754,16 +1143,9 @@ public void shouldReportMisconfigurationsOfCleanupPolicyForRepartitionTopics() { public void shouldReportMultipleMisconfigurationsForSameTopic() { final long retentionMs = 1000; final long shorterRetentionMs = 900; - mockAdminClient.addTopic( - false, - topic1, - Collections.singletonList(new TopicPartitionInfo(0, broker1, cluster, Collections.emptyList())), - mkMap( - mkEntry(TopicConfig.CLEANUP_POLICY_CONFIG, TopicConfig.CLEANUP_POLICY_COMPACT + "," + TopicConfig.CLEANUP_POLICY_DELETE), - mkEntry(TopicConfig.RETENTION_MS_CONFIG, String.valueOf(shorterRetentionMs)), - mkEntry(TopicConfig.RETENTION_BYTES_CONFIG, "1024") - ) - ); + final Map windowedChangelogConfig = windowedChangelogConfig(shorterRetentionMs); + windowedChangelogConfig.put(TopicConfig.RETENTION_BYTES_CONFIG, "1024"); + setupTopicInMockAdminClient(topic1, windowedChangelogConfig); final InternalTopicConfig internalTopicConfig1 = setupWindowedChangelogTopicConfig(topic1, 1, retentionMs); final ValidationResult validationResult = internalTopicManager.validate(mkMap( @@ -789,12 +1171,7 @@ public void shouldReportMultipleMisconfigurationsForSameTopic() { @Test public void shouldThrowWhenPartitionCountUnknown() { - mockAdminClient.addTopic( - false, - topic1, - Collections.singletonList(new TopicPartitionInfo(0, broker1, cluster, Collections.emptyList())), - mkMap(mkEntry(TopicConfig.CLEANUP_POLICY_CONFIG, TopicConfig.CLEANUP_POLICY_DELETE)) - ); + setupTopicInMockAdminClient(topic1, repartitionTopicConfig()); final InternalTopicConfig internalTopicConfig = new RepartitionTopicConfig(topic1, Collections.emptyMap()); assertThrows( @@ -804,17 +1181,26 @@ public void shouldThrowWhenPartitionCountUnknown() { } @Test - public void shouldRetryWhenCallsThrowTimeoutExceptionDuringValidation() { - mockAdminClient.addTopic( - false, - topic1, - Collections.singletonList(new TopicPartitionInfo(0, broker1, cluster, Collections.emptyList())), - mkMap( - mkEntry(TopicConfig.CLEANUP_POLICY_CONFIG, TopicConfig.CLEANUP_POLICY_DELETE), - mkEntry(TopicConfig.RETENTION_MS_CONFIG, "-1"), - mkEntry(TopicConfig.RETENTION_BYTES_CONFIG, null) - ) + public void shouldNotThrowExceptionIfTopicExistsWithDifferentReplication() { + setupTopicInMockAdminClient(topic1, repartitionTopicConfig()); + // attempt to create it again with replication 1 + final InternalTopicManager internalTopicManager2 = new InternalTopicManager( + Time.SYSTEM, + mockAdminClient, + new StreamsConfig(config) ); + + final InternalTopicConfig internalTopicConfig = setupRepartitionTopicConfig(topic1, 1); + final ValidationResult validationResult = + internalTopicManager2.validate(Collections.singletonMap(topic1, internalTopicConfig)); + + assertThat(validationResult.missingTopics(), empty()); + assertThat(validationResult.misconfigurationsForTopics(), anEmptyMap()); + } + + @Test + public void shouldRetryWhenCallsThrowTimeoutExceptionDuringValidation() { + setupTopicInMockAdminClient(topic1, repartitionTopicConfig()); mockAdminClient.timeoutNextRequest(2); final InternalTopicConfig internalTopicConfig = setupRepartitionTopicConfig(topic1, 1); @@ -898,6 +1284,63 @@ public void shouldOnlyRetryDescribeConfigsWhenDescribeConfigsThrowsLeaderNotAvai EasyMock.verify(admin); } + @Test + public void shouldOnlyRetryNotSuccessfulFuturesDuringValidation() { + final AdminClient admin = EasyMock.createNiceMock(AdminClient.class); + final InternalTopicManager topicManager = new InternalTopicManager( + Time.SYSTEM, + admin, + new StreamsConfig(config) + ); + final KafkaFutureImpl topicDescriptionFailFuture = new KafkaFutureImpl<>(); + topicDescriptionFailFuture.completeExceptionally(new LeaderNotAvailableException("Leader Not Available!")); + final KafkaFutureImpl topicDescriptionSuccessfulFuture1 = new KafkaFutureImpl<>(); + topicDescriptionSuccessfulFuture1.complete(new TopicDescription( + topic1, + false, + Collections.singletonList(new TopicPartitionInfo(0, broker1, cluster, Collections.emptyList())) + )); + final KafkaFutureImpl topicDescriptionSuccessfulFuture2 = new KafkaFutureImpl<>(); + topicDescriptionSuccessfulFuture2.complete(new TopicDescription( + topic2, + false, + Collections.singletonList(new TopicPartitionInfo(0, broker1, cluster, Collections.emptyList())) + )); + EasyMock.expect(admin.describeTopics(mkSet(topic1, topic2))) + .andAnswer(() -> new MockDescribeTopicsResult(mkMap( + mkEntry(topic1, topicDescriptionSuccessfulFuture1), + mkEntry(topic2, topicDescriptionFailFuture) + ))); + EasyMock.expect(admin.describeTopics(mkSet(topic2))) + .andAnswer(() -> new MockDescribeTopicsResult(mkMap( + mkEntry(topic2, topicDescriptionSuccessfulFuture2) + ))); + final KafkaFutureImpl topicConfigSuccessfulFuture = new KafkaFutureImpl<>(); + topicConfigSuccessfulFuture.complete( + new Config(repartitionTopicConfig().entrySet().stream() + .map(entry -> new ConfigEntry(entry.getKey(), entry.getValue())).collect(Collectors.toSet())) + ); + final ConfigResource topicResource1 = new ConfigResource(Type.TOPIC, topic1); + final ConfigResource topicResource2 = new ConfigResource(Type.TOPIC, topic2); + EasyMock.expect(admin.describeConfigs(mkSet(topicResource1, topicResource2))) + .andAnswer(() -> new MockDescribeConfigsResult(mkMap( + mkEntry(topicResource1, topicConfigSuccessfulFuture), + mkEntry(topicResource2, topicConfigSuccessfulFuture) + ))); + EasyMock.replay(admin); + final InternalTopicConfig internalTopicConfig1 = setupRepartitionTopicConfig(topic1, 1); + final InternalTopicConfig internalTopicConfig2 = setupRepartitionTopicConfig(topic2, 1); + + final ValidationResult validationResult = topicManager.validate(mkMap( + mkEntry(topic1, internalTopicConfig1), + mkEntry(topic2, internalTopicConfig2) + )); + + assertThat(validationResult.missingTopics(), empty()); + assertThat(validationResult.misconfigurationsForTopics(), anEmptyMap()); + EasyMock.verify(admin); + } + @Test public void shouldThrowWhenDescribeTopicsThrowsUnexpectedExceptionDuringValidation() { final AdminClient admin = EasyMock.createNiceMock(AdminClient.class); @@ -1154,6 +1597,19 @@ public void shouldThrowTimeoutExceptionWhenFuturesNeverCompleteDuringValidation( ); } + private NewTopic newTopic(final String topicName, + final InternalTopicConfig topicConfig, + final StreamsConfig streamsConfig) { + return new NewTopic( + topicName, + topicConfig.numberOfPartitions(), + Optional.of(streamsConfig.getInt(StreamsConfig.REPLICATION_FACTOR_CONFIG).shortValue()) + ).configs(topicConfig.getProperties( + Collections.emptyMap(), + streamsConfig.getLong(StreamsConfig.WINDOW_STORE_CHANGE_LOG_ADDITIONAL_RETENTION_MS_CONFIG)) + ); + } + private Map repartitionTopicConfig() { return mkMap( mkEntry(TopicConfig.CLEANUP_POLICY_CONFIG, TopicConfig.CLEANUP_POLICY_DELETE), @@ -1212,21 +1668,24 @@ private InternalTopicConfig setupRepartitionTopicConfig(final String topicName, } private static class MockCreateTopicsResult extends CreateTopicsResult { - MockCreateTopicsResult(final Map> futures) { super(futures); } } - private static class MockDescribeTopicsResult extends DescribeTopicsResult { + private static class MockDeleteTopicsResult extends DeleteTopicsResult { + MockDeleteTopicsResult(final Map> futures) { + super(futures); + } + } + private static class MockDescribeTopicsResult extends DescribeTopicsResult { MockDescribeTopicsResult(final Map> futures) { super(futures); } } private static class MockDescribeConfigsResult extends DescribeConfigsResult { - MockDescribeConfigsResult(final Map> futures) { super(futures); }