@Test public void testStorePartitions() throws Exception { StreamsConfig config = createConfig(baseDir); StandbyTask task = new StandbyTask(taskId, applicationId, topicPartitions, topology, consumer, changelogReader, config, null, stateDirectory); assertEquals(Utils.mkSet(partition2), new HashSet<>(task.checkpointedOffsets().keySet())); }
Map<TopicPartition, Long> checkpointedOffsets() { return checkpointedOffsets; }
StandbyTask extends AbstractTask { Map<TopicPartition, Long> checkpointedOffsets() { return checkpointedOffsets; } }
StandbyTask extends AbstractTask { Map<TopicPartition, Long> checkpointedOffsets() { return checkpointedOffsets; } StandbyTask(final TaskId id,
                final String applicationId,
                final Collection<TopicPartition> partitions,
                final ProcessorTopology topology,
                final Consumer<byte[], byte[]> consumer,
                final ChangelogReader changelogReader,
                final StreamsConfig config,
                final StreamsMetrics metrics,
                final StateDirectory stateDirectory); }
StandbyTask extends AbstractTask { Map<TopicPartition, Long> checkpointedOffsets() { return checkpointedOffsets; } StandbyTask(final TaskId id,
                final String applicationId,
                final Collection<TopicPartition> partitions,
                final ProcessorTopology topology,
                final Consumer<byte[], byte[]> consumer,
                final ChangelogReader changelogReader,
                final StreamsConfig config,
                final StreamsMetrics metrics,
                final StateDirectory stateDirectory); @Override void resume(); @Override void commit(); @Override void suspend(); @Override void close(final boolean clean); List<ConsumerRecord<byte[], byte[]>> update(final TopicPartition partition,
                                                       final List<ConsumerRecord<byte[], byte[]>> records); }
StandbyTask extends AbstractTask { Map<TopicPartition, Long> checkpointedOffsets() { return checkpointedOffsets; } StandbyTask(final TaskId id,
                final String applicationId,
                final Collection<TopicPartition> partitions,
                final ProcessorTopology topology,
                final Consumer<byte[], byte[]> consumer,
                final ChangelogReader changelogReader,
                final StreamsConfig config,
                final StreamsMetrics metrics,
                final StateDirectory stateDirectory); @Override void resume(); @Override void commit(); @Override void suspend(); @Override void close(final boolean clean); List<ConsumerRecord<byte[], byte[]>> update(final TopicPartition partition,
                                                       final List<ConsumerRecord<byte[], byte[]>> records);  }