@SuppressWarnings("unchecked") @Test public void shouldFallbackToPartitionsForIfPartitionNotInAllPartitionsList() throws Exception { final MockConsumer<byte[], byte[]> consumer = new MockConsumer(OffsetResetStrategy.EARLIEST) { @Override public List<PartitionInfo> partitionsFor(final String topic) { return Collections.singletonList(partitionInfo); } }; final StoreChangelogReader changelogReader = new StoreChangelogReader(consumer, new MockTime(), 10); changelogReader.validatePartitionExists(topicPartition, "store"); }
@Override public void validatePartitionExists(final TopicPartition topicPartition, final String storeName) { final long start = time.milliseconds(); if (partitionInfo.isEmpty()) { try { partitionInfo.putAll(consumer.listTopics()); } catch (final TimeoutException e) { log.warn("{} Could not list topics so will fall back to partition by partition fetching", logPrefix); } } final long endTime = time.milliseconds() + partitionValidationTimeoutMs; while (!hasPartition(topicPartition) && time.milliseconds() < endTime) { try { final List<PartitionInfo> partitions = consumer.partitionsFor(topicPartition.topic()); if (partitions != null) { partitionInfo.put(topicPartition.topic(), partitions); } } catch (final TimeoutException e) { throw new StreamsException(String.format("Could not fetch partition info for topic: %s before expiration of the configured request timeout", topicPartition.topic())); } } if (!hasPartition(topicPartition)) { throw new StreamsException(String.format("Store %s's change log (%s) does not contain partition %s", storeName, topicPartition.topic(), topicPartition.partition())); } log.debug("{} Took {} ms to validate that partition {} exists", logPrefix, time.milliseconds() - start, topicPartition); }
StoreChangelogReader implements ChangelogReader { @Override public void validatePartitionExists(final TopicPartition topicPartition, final String storeName) { final long start = time.milliseconds(); if (partitionInfo.isEmpty()) { try { partitionInfo.putAll(consumer.listTopics()); } catch (final TimeoutException e) { log.warn("{} Could not list topics so will fall back to partition by partition fetching", logPrefix); } } final long endTime = time.milliseconds() + partitionValidationTimeoutMs; while (!hasPartition(topicPartition) && time.milliseconds() < endTime) { try { final List<PartitionInfo> partitions = consumer.partitionsFor(topicPartition.topic()); if (partitions != null) { partitionInfo.put(topicPartition.topic(), partitions); } } catch (final TimeoutException e) { throw new StreamsException(String.format("Could not fetch partition info for topic: %s before expiration of the configured request timeout", topicPartition.topic())); } } if (!hasPartition(topicPartition)) { throw new StreamsException(String.format("Store %s's change log (%s) does not contain partition %s", storeName, topicPartition.topic(), topicPartition.partition())); } log.debug("{} Took {} ms to validate that partition {} exists", logPrefix, time.milliseconds() - start, topicPartition); } }
StoreChangelogReader implements ChangelogReader { @Override public void validatePartitionExists(final TopicPartition topicPartition, final String storeName) { final long start = time.milliseconds(); if (partitionInfo.isEmpty()) { try { partitionInfo.putAll(consumer.listTopics()); } catch (final TimeoutException e) { log.warn("{} Could not list topics so will fall back to partition by partition fetching", logPrefix); } } final long endTime = time.milliseconds() + partitionValidationTimeoutMs; while (!hasPartition(topicPartition) && time.milliseconds() < endTime) { try { final List<PartitionInfo> partitions = consumer.partitionsFor(topicPartition.topic()); if (partitions != null) { partitionInfo.put(topicPartition.topic(), partitions); } } catch (final TimeoutException e) { throw new StreamsException(String.format("Could not fetch partition info for topic: %s before expiration of the configured request timeout", topicPartition.topic())); } } if (!hasPartition(topicPartition)) { throw new StreamsException(String.format("Store %s's change log (%s) does not contain partition %s", storeName, topicPartition.topic(), topicPartition.partition())); } log.debug("{} Took {} ms to validate that partition {} exists", logPrefix, time.milliseconds() - start, topicPartition); } StoreChangelogReader(final String threadId, final Consumer<byte[], byte[]> consumer, final Time time, final long partitionValidationTimeoutMs);  StoreChangelogReader(final Consumer<byte[], byte[]> consumer, final Time time, final long partitionValidationTimeoutMs); }
StoreChangelogReader implements ChangelogReader { @Override public void validatePartitionExists(final TopicPartition topicPartition, final String storeName) { final long start = time.milliseconds(); if (partitionInfo.isEmpty()) { try { partitionInfo.putAll(consumer.listTopics()); } catch (final TimeoutException e) { log.warn("{} Could not list topics so will fall back to partition by partition fetching", logPrefix); } } final long endTime = time.milliseconds() + partitionValidationTimeoutMs; while (!hasPartition(topicPartition) && time.milliseconds() < endTime) { try { final List<PartitionInfo> partitions = consumer.partitionsFor(topicPartition.topic()); if (partitions != null) { partitionInfo.put(topicPartition.topic(), partitions); } } catch (final TimeoutException e) { throw new StreamsException(String.format("Could not fetch partition info for topic: %s before expiration of the configured request timeout", topicPartition.topic())); } } if (!hasPartition(topicPartition)) { throw new StreamsException(String.format("Store %s's change log (%s) does not contain partition %s", storeName, topicPartition.topic(), topicPartition.partition())); } log.debug("{} Took {} ms to validate that partition {} exists", logPrefix, time.milliseconds() - start, topicPartition); } StoreChangelogReader(final String threadId, final Consumer<byte[], byte[]> consumer, final Time time, final long partitionValidationTimeoutMs);  StoreChangelogReader(final Consumer<byte[], byte[]> consumer, final Time time, final long partitionValidationTimeoutMs); @Override void validatePartitionExists(final TopicPartition topicPartition, final String storeName); @Override void register(final StateRestorer restorer); void restore(); @Override Map<TopicPartition, Long> restoredOffsets(); }
StoreChangelogReader implements ChangelogReader { @Override public void validatePartitionExists(final TopicPartition topicPartition, final String storeName) { final long start = time.milliseconds(); if (partitionInfo.isEmpty()) { try { partitionInfo.putAll(consumer.listTopics()); } catch (final TimeoutException e) { log.warn("{} Could not list topics so will fall back to partition by partition fetching", logPrefix); } } final long endTime = time.milliseconds() + partitionValidationTimeoutMs; while (!hasPartition(topicPartition) && time.milliseconds() < endTime) { try { final List<PartitionInfo> partitions = consumer.partitionsFor(topicPartition.topic()); if (partitions != null) { partitionInfo.put(topicPartition.topic(), partitions); } } catch (final TimeoutException e) { throw new StreamsException(String.format("Could not fetch partition info for topic: %s before expiration of the configured request timeout", topicPartition.topic())); } } if (!hasPartition(topicPartition)) { throw new StreamsException(String.format("Store %s's change log (%s) does not contain partition %s", storeName, topicPartition.topic(), topicPartition.partition())); } log.debug("{} Took {} ms to validate that partition {} exists", logPrefix, time.milliseconds() - start, topicPartition); } StoreChangelogReader(final String threadId, final Consumer<byte[], byte[]> consumer, final Time time, final long partitionValidationTimeoutMs);  StoreChangelogReader(final Consumer<byte[], byte[]> consumer, final Time time, final long partitionValidationTimeoutMs); @Override void validatePartitionExists(final TopicPartition topicPartition, final String storeName); @Override void register(final StateRestorer restorer); void restore(); @Override Map<TopicPartition, Long> restoredOffsets();  }