@Test public void shouldThrowExceptionIfConsumerHasCurrentSubscription() throws Exception { consumer.subscribe(Collections.singleton("sometopic")); try { changelogReader.restore(); fail("Should have thrown IllegalStateException"); } catch (final IllegalStateException e) { } }
public void restore() { final long start = time.milliseconds(); try { if (!consumer.subscription().isEmpty()) { throw new IllegalStateException(String.format("Restore consumer should have not subscribed to any partitions (%s) beforehand", consumer.subscription())); } final Map<TopicPartition, Long> endOffsets = consumer.endOffsets(stateRestorers.keySet()); final Map<TopicPartition, StateRestorer> needsRestoring = new HashMap<>(); for (final Map.Entry<TopicPartition, Long> entry : endOffsets.entrySet()) { TopicPartition topicPartition = entry.getKey(); Long offset = entry.getValue(); final StateRestorer restorer = stateRestorers.get(topicPartition); if (restorer.checkpoint() >= offset) { restorer.setRestoredOffset(restorer.checkpoint()); } else { needsRestoring.put(topicPartition, restorer); } } log.info("{} Starting restoring state stores from changelog topics {}", logPrefix, needsRestoring.keySet()); consumer.assign(needsRestoring.keySet()); final List<StateRestorer> needsPositionUpdate = new ArrayList<>(); for (final StateRestorer restorer : needsRestoring.values()) { if (restorer.checkpoint() != StateRestorer.NO_CHECKPOINT) { consumer.seek(restorer.partition(), restorer.checkpoint()); logRestoreOffsets(restorer.partition(), restorer.checkpoint(), endOffsets.get(restorer.partition())); restorer.setStartingOffset(consumer.position(restorer.partition())); } else { consumer.seekToBeginning(Collections.singletonList(restorer.partition())); needsPositionUpdate.add(restorer); } } for (final StateRestorer restorer : needsPositionUpdate) { final long position = consumer.position(restorer.partition()); restorer.setStartingOffset(position); logRestoreOffsets(restorer.partition(), position, endOffsets.get(restorer.partition())); } final Set<TopicPartition> partitions = new HashSet<>(needsRestoring.keySet()); while (!partitions.isEmpty()) { final ConsumerRecords<byte[], byte[]> allRecords = consumer.poll(10); final Iterator<TopicPartition> partitionIterator = partitions.iterator(); while (partitionIterator.hasNext()) { restorePartition(endOffsets, allRecords, partitionIterator); } } } finally { consumer.assign(Collections.<TopicPartition>emptyList()); log.debug("{} Took {} ms to restore all active states", logPrefix, time.milliseconds() - start); } }
StoreChangelogReader implements ChangelogReader { public void restore() { final long start = time.milliseconds(); try { if (!consumer.subscription().isEmpty()) { throw new IllegalStateException(String.format("Restore consumer should have not subscribed to any partitions (%s) beforehand", consumer.subscription())); } final Map<TopicPartition, Long> endOffsets = consumer.endOffsets(stateRestorers.keySet()); final Map<TopicPartition, StateRestorer> needsRestoring = new HashMap<>(); for (final Map.Entry<TopicPartition, Long> entry : endOffsets.entrySet()) { TopicPartition topicPartition = entry.getKey(); Long offset = entry.getValue(); final StateRestorer restorer = stateRestorers.get(topicPartition); if (restorer.checkpoint() >= offset) { restorer.setRestoredOffset(restorer.checkpoint()); } else { needsRestoring.put(topicPartition, restorer); } } log.info("{} Starting restoring state stores from changelog topics {}", logPrefix, needsRestoring.keySet()); consumer.assign(needsRestoring.keySet()); final List<StateRestorer> needsPositionUpdate = new ArrayList<>(); for (final StateRestorer restorer : needsRestoring.values()) { if (restorer.checkpoint() != StateRestorer.NO_CHECKPOINT) { consumer.seek(restorer.partition(), restorer.checkpoint()); logRestoreOffsets(restorer.partition(), restorer.checkpoint(), endOffsets.get(restorer.partition())); restorer.setStartingOffset(consumer.position(restorer.partition())); } else { consumer.seekToBeginning(Collections.singletonList(restorer.partition())); needsPositionUpdate.add(restorer); } } for (final StateRestorer restorer : needsPositionUpdate) { final long position = consumer.position(restorer.partition()); restorer.setStartingOffset(position); logRestoreOffsets(restorer.partition(), position, endOffsets.get(restorer.partition())); } final Set<TopicPartition> partitions = new HashSet<>(needsRestoring.keySet()); while (!partitions.isEmpty()) { final ConsumerRecords<byte[], byte[]> allRecords = consumer.poll(10); final Iterator<TopicPartition> partitionIterator = partitions.iterator(); while (partitionIterator.hasNext()) { restorePartition(endOffsets, allRecords, partitionIterator); } } } finally { consumer.assign(Collections.<TopicPartition>emptyList()); log.debug("{} Took {} ms to restore all active states", logPrefix, time.milliseconds() - start); } } }
StoreChangelogReader implements ChangelogReader { public void restore() { final long start = time.milliseconds(); try { if (!consumer.subscription().isEmpty()) { throw new IllegalStateException(String.format("Restore consumer should have not subscribed to any partitions (%s) beforehand", consumer.subscription())); } final Map<TopicPartition, Long> endOffsets = consumer.endOffsets(stateRestorers.keySet()); final Map<TopicPartition, StateRestorer> needsRestoring = new HashMap<>(); for (final Map.Entry<TopicPartition, Long> entry : endOffsets.entrySet()) { TopicPartition topicPartition = entry.getKey(); Long offset = entry.getValue(); final StateRestorer restorer = stateRestorers.get(topicPartition); if (restorer.checkpoint() >= offset) { restorer.setRestoredOffset(restorer.checkpoint()); } else { needsRestoring.put(topicPartition, restorer); } } log.info("{} Starting restoring state stores from changelog topics {}", logPrefix, needsRestoring.keySet()); consumer.assign(needsRestoring.keySet()); final List<StateRestorer> needsPositionUpdate = new ArrayList<>(); for (final StateRestorer restorer : needsRestoring.values()) { if (restorer.checkpoint() != StateRestorer.NO_CHECKPOINT) { consumer.seek(restorer.partition(), restorer.checkpoint()); logRestoreOffsets(restorer.partition(), restorer.checkpoint(), endOffsets.get(restorer.partition())); restorer.setStartingOffset(consumer.position(restorer.partition())); } else { consumer.seekToBeginning(Collections.singletonList(restorer.partition())); needsPositionUpdate.add(restorer); } } for (final StateRestorer restorer : needsPositionUpdate) { final long position = consumer.position(restorer.partition()); restorer.setStartingOffset(position); logRestoreOffsets(restorer.partition(), position, endOffsets.get(restorer.partition())); } final Set<TopicPartition> partitions = new HashSet<>(needsRestoring.keySet()); while (!partitions.isEmpty()) { final ConsumerRecords<byte[], byte[]> allRecords = consumer.poll(10); final Iterator<TopicPartition> partitionIterator = partitions.iterator(); while (partitionIterator.hasNext()) { restorePartition(endOffsets, allRecords, partitionIterator); } } } finally { consumer.assign(Collections.<TopicPartition>emptyList()); log.debug("{} Took {} ms to restore all active states", logPrefix, time.milliseconds() - start); } } 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 { public void restore() { final long start = time.milliseconds(); try { if (!consumer.subscription().isEmpty()) { throw new IllegalStateException(String.format("Restore consumer should have not subscribed to any partitions (%s) beforehand", consumer.subscription())); } final Map<TopicPartition, Long> endOffsets = consumer.endOffsets(stateRestorers.keySet()); final Map<TopicPartition, StateRestorer> needsRestoring = new HashMap<>(); for (final Map.Entry<TopicPartition, Long> entry : endOffsets.entrySet()) { TopicPartition topicPartition = entry.getKey(); Long offset = entry.getValue(); final StateRestorer restorer = stateRestorers.get(topicPartition); if (restorer.checkpoint() >= offset) { restorer.setRestoredOffset(restorer.checkpoint()); } else { needsRestoring.put(topicPartition, restorer); } } log.info("{} Starting restoring state stores from changelog topics {}", logPrefix, needsRestoring.keySet()); consumer.assign(needsRestoring.keySet()); final List<StateRestorer> needsPositionUpdate = new ArrayList<>(); for (final StateRestorer restorer : needsRestoring.values()) { if (restorer.checkpoint() != StateRestorer.NO_CHECKPOINT) { consumer.seek(restorer.partition(), restorer.checkpoint()); logRestoreOffsets(restorer.partition(), restorer.checkpoint(), endOffsets.get(restorer.partition())); restorer.setStartingOffset(consumer.position(restorer.partition())); } else { consumer.seekToBeginning(Collections.singletonList(restorer.partition())); needsPositionUpdate.add(restorer); } } for (final StateRestorer restorer : needsPositionUpdate) { final long position = consumer.position(restorer.partition()); restorer.setStartingOffset(position); logRestoreOffsets(restorer.partition(), position, endOffsets.get(restorer.partition())); } final Set<TopicPartition> partitions = new HashSet<>(needsRestoring.keySet()); while (!partitions.isEmpty()) { final ConsumerRecords<byte[], byte[]> allRecords = consumer.poll(10); final Iterator<TopicPartition> partitionIterator = partitions.iterator(); while (partitionIterator.hasNext()) { restorePartition(endOffsets, allRecords, partitionIterator); } } } finally { consumer.assign(Collections.<TopicPartition>emptyList()); log.debug("{} Took {} ms to restore all active states", logPrefix, time.milliseconds() - start); } } 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 { public void restore() { final long start = time.milliseconds(); try { if (!consumer.subscription().isEmpty()) { throw new IllegalStateException(String.format("Restore consumer should have not subscribed to any partitions (%s) beforehand", consumer.subscription())); } final Map<TopicPartition, Long> endOffsets = consumer.endOffsets(stateRestorers.keySet()); final Map<TopicPartition, StateRestorer> needsRestoring = new HashMap<>(); for (final Map.Entry<TopicPartition, Long> entry : endOffsets.entrySet()) { TopicPartition topicPartition = entry.getKey(); Long offset = entry.getValue(); final StateRestorer restorer = stateRestorers.get(topicPartition); if (restorer.checkpoint() >= offset) { restorer.setRestoredOffset(restorer.checkpoint()); } else { needsRestoring.put(topicPartition, restorer); } } log.info("{} Starting restoring state stores from changelog topics {}", logPrefix, needsRestoring.keySet()); consumer.assign(needsRestoring.keySet()); final List<StateRestorer> needsPositionUpdate = new ArrayList<>(); for (final StateRestorer restorer : needsRestoring.values()) { if (restorer.checkpoint() != StateRestorer.NO_CHECKPOINT) { consumer.seek(restorer.partition(), restorer.checkpoint()); logRestoreOffsets(restorer.partition(), restorer.checkpoint(), endOffsets.get(restorer.partition())); restorer.setStartingOffset(consumer.position(restorer.partition())); } else { consumer.seekToBeginning(Collections.singletonList(restorer.partition())); needsPositionUpdate.add(restorer); } } for (final StateRestorer restorer : needsPositionUpdate) { final long position = consumer.position(restorer.partition()); restorer.setStartingOffset(position); logRestoreOffsets(restorer.partition(), position, endOffsets.get(restorer.partition())); } final Set<TopicPartition> partitions = new HashSet<>(needsRestoring.keySet()); while (!partitions.isEmpty()) { final ConsumerRecords<byte[], byte[]> allRecords = consumer.poll(10); final Iterator<TopicPartition> partitionIterator = partitions.iterator(); while (partitionIterator.hasNext()) { restorePartition(endOffsets, allRecords, partitionIterator); } } } finally { consumer.assign(Collections.<TopicPartition>emptyList()); log.debug("{} Took {} ms to restore all active states", logPrefix, time.milliseconds() - start); } } 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();  }