@Test public void shouldReturnCorrectPartitionCounts() throws Exception { InternalTopicManager internalTopicManager = new InternalTopicManager(streamsKafkaClient, 1, WINDOW_CHANGE_LOG_ADDITIONAL_RETENTION_DEFAULT, time); Assert.assertEquals(Collections.singletonMap(topic, 1), internalTopicManager.getNumPartitions(Collections.singleton(topic))); }
public Map<String, Integer> getNumPartitions(final Set<String> topics) { for (int i = 0; i < MAX_TOPIC_READY_TRY; i++) { try { final MetadataResponse metadata = streamsKafkaClient.fetchMetadata(); final Map<String, Integer> existingTopicPartitions = fetchExistingPartitionCountByTopic(metadata); existingTopicPartitions.keySet().retainAll(topics); return existingTopicPartitions; } catch (StreamsException ex) { log.warn("Could not get number of partitions: " + ex.getMessage() + " Retry #" + i); } time.sleep(100L); } throw new StreamsException("Could not get number of partitions."); }
InternalTopicManager { public Map<String, Integer> getNumPartitions(final Set<String> topics) { for (int i = 0; i < MAX_TOPIC_READY_TRY; i++) { try { final MetadataResponse metadata = streamsKafkaClient.fetchMetadata(); final Map<String, Integer> existingTopicPartitions = fetchExistingPartitionCountByTopic(metadata); existingTopicPartitions.keySet().retainAll(topics); return existingTopicPartitions; } catch (StreamsException ex) { log.warn("Could not get number of partitions: " + ex.getMessage() + " Retry #" + i); } time.sleep(100L); } throw new StreamsException("Could not get number of partitions."); } }
InternalTopicManager { public Map<String, Integer> getNumPartitions(final Set<String> topics) { for (int i = 0; i < MAX_TOPIC_READY_TRY; i++) { try { final MetadataResponse metadata = streamsKafkaClient.fetchMetadata(); final Map<String, Integer> existingTopicPartitions = fetchExistingPartitionCountByTopic(metadata); existingTopicPartitions.keySet().retainAll(topics); return existingTopicPartitions; } catch (StreamsException ex) { log.warn("Could not get number of partitions: " + ex.getMessage() + " Retry #" + i); } time.sleep(100L); } throw new StreamsException("Could not get number of partitions."); } InternalTopicManager(final StreamsKafkaClient streamsKafkaClient, final int replicationFactor,
                                final long windowChangeLogAdditionalRetention, final Time time); }
InternalTopicManager { public Map<String, Integer> getNumPartitions(final Set<String> topics) { for (int i = 0; i < MAX_TOPIC_READY_TRY; i++) { try { final MetadataResponse metadata = streamsKafkaClient.fetchMetadata(); final Map<String, Integer> existingTopicPartitions = fetchExistingPartitionCountByTopic(metadata); existingTopicPartitions.keySet().retainAll(topics); return existingTopicPartitions; } catch (StreamsException ex) { log.warn("Could not get number of partitions: " + ex.getMessage() + " Retry #" + i); } time.sleep(100L); } throw new StreamsException("Could not get number of partitions."); } InternalTopicManager(final StreamsKafkaClient streamsKafkaClient, final int replicationFactor,
                                final long windowChangeLogAdditionalRetention, final Time time); void makeReady(final Map<InternalTopicConfig, Integer> topics); Map<String, Integer> getNumPartitions(final Set<String> topics); void close(); }
InternalTopicManager { public Map<String, Integer> getNumPartitions(final Set<String> topics) { for (int i = 0; i < MAX_TOPIC_READY_TRY; i++) { try { final MetadataResponse metadata = streamsKafkaClient.fetchMetadata(); final Map<String, Integer> existingTopicPartitions = fetchExistingPartitionCountByTopic(metadata); existingTopicPartitions.keySet().retainAll(topics); return existingTopicPartitions; } catch (StreamsException ex) { log.warn("Could not get number of partitions: " + ex.getMessage() + " Retry #" + i); } time.sleep(100L); } throw new StreamsException("Could not get number of partitions."); } InternalTopicManager(final StreamsKafkaClient streamsKafkaClient, final int replicationFactor,
                                final long windowChangeLogAdditionalRetention, final Time time); void makeReady(final Map<InternalTopicConfig, Integer> topics); Map<String, Integer> getNumPartitions(final Set<String> topics); void close(); static final String CLEANUP_POLICY_PROP; static final String RETENTION_MS; static final Long WINDOW_CHANGE_LOG_ADDITIONAL_RETENTION_DEFAULT; }