@Test public void testStorePartitions() throws Exception { File baseDir = Files.createTempDirectory("test").toFile(); try { StreamsConfig config = createConfig(baseDir); StandbyTask task = new StandbyTask(taskId, applicationId, topicPartitions, topology, consumer, restoreStateConsumer, config, null); assertEquals(Utils.mkSet(partition2), new HashSet<>(task.changeLogPartitions())); } finally { Utils.delete(baseDir); } }
public Collection<TopicPartition> changeLogPartitions() { return checkpointedOffsets.keySet(); }
StandbyTask extends AbstractTask { public Collection<TopicPartition> changeLogPartitions() { return checkpointedOffsets.keySet(); } }
StandbyTask extends AbstractTask { public Collection<TopicPartition> changeLogPartitions() { return checkpointedOffsets.keySet(); } StandbyTask(TaskId id,
                       String applicationId,
                       Collection<TopicPartition> partitions,
                       ProcessorTopology topology,
                       Consumer<byte[], byte[]> consumer,
                       Consumer<byte[], byte[]> restoreConsumer,
                       StreamsConfig config,
                       StreamsMetrics metrics); }
StandbyTask extends AbstractTask { public Collection<TopicPartition> changeLogPartitions() { return checkpointedOffsets.keySet(); } StandbyTask(TaskId id,
                       String applicationId,
                       Collection<TopicPartition> partitions,
                       ProcessorTopology topology,
                       Consumer<byte[], byte[]> consumer,
                       Consumer<byte[], byte[]> restoreConsumer,
                       StreamsConfig config,
                       StreamsMetrics metrics); Map<TopicPartition, Long> checkpointedOffsets(); Collection<TopicPartition> changeLogPartitions(); List<ConsumerRecord<byte[], byte[]>> update(TopicPartition partition, List<ConsumerRecord<byte[], byte[]>> records); void commit(); }
StandbyTask extends AbstractTask { public Collection<TopicPartition> changeLogPartitions() { return checkpointedOffsets.keySet(); } StandbyTask(TaskId id,
                       String applicationId,
                       Collection<TopicPartition> partitions,
                       ProcessorTopology topology,
                       Consumer<byte[], byte[]> consumer,
                       Consumer<byte[], byte[]> restoreConsumer,
                       StreamsConfig config,
                       StreamsMetrics metrics); Map<TopicPartition, Long> checkpointedOffsets(); Collection<TopicPartition> changeLogPartitions(); List<ConsumerRecord<byte[], byte[]>> update(TopicPartition partition, List<ConsumerRecord<byte[], byte[]>> records); void commit();  }