@SuppressWarnings("unchecked") @Test public void shouldCheckpointOffsetsOnCommit() throws Exception { final String storeName = "test"; final String changelogTopic = ProcessorStateManager.storeChangelogTopic("appId", storeName); final InMemoryKeyValueStore inMemoryStore = new InMemoryKeyValueStore(storeName, null, null) { @Override public void init(final ProcessorContext context, final StateStore root) { context.register(root, true, null); } @Override public boolean persistent() { return true; } }; Map<String, SourceNode> sourceByTopics = new HashMap() { { put(partition1.topic(), source1); put(partition2.topic(), source2); } }; final ProcessorTopology topology = new ProcessorTopology(Collections.<ProcessorNode>emptyList(), sourceByTopics, Collections.<String, SinkNode>emptyMap(), Collections.<StateStore>singletonList(inMemoryStore), Collections.singletonMap(storeName, changelogTopic), Collections.<StateStore>emptyList()); final TopicPartition partition = new TopicPartition(changelogTopic, 0); restoreStateConsumer.updatePartitions(changelogTopic, Collections.singletonList( new PartitionInfo(changelogTopic, 0, null, new Node[0], new Node[0]))); restoreStateConsumer.updateEndOffsets(Collections.singletonMap(partition, 0L)); restoreStateConsumer.updateBeginningOffsets(Collections.singletonMap(partition, 0L)); final long offset = 543L; final StreamTask streamTask = new StreamTask(taskId00, "appId", partitions, topology, consumer, changelogReader, config, streamsMetrics, stateDirectory, null, time, producer) { @Override RecordCollector createRecordCollector() { return new NoOpRecordCollector() { @Override public Map<TopicPartition, Long> offsets() { return Collections.singletonMap(partition, offset); } }; } }; time.sleep(config.getLong(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG)); streamTask.commit(); final OffsetCheckpoint checkpoint = new OffsetCheckpoint(new File(stateDirectory.directoryForTask(taskId00), ProcessorStateManager.CHECKPOINT_FILE_NAME)); assertThat(checkpoint.read(), equalTo(Collections.singletonMap(partition, offset + 1))); }
@Override public void commit() { commitImpl(true); }
StreamTask extends AbstractTask implements Punctuator { @Override public void commit() { commitImpl(true); } }
StreamTask extends AbstractTask implements Punctuator { @Override public void commit() { commitImpl(true); } StreamTask(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,
                      final ThreadCache cache,
                      final Time time,
                      final Producer<byte[], byte[]> producer); }
StreamTask extends AbstractTask implements Punctuator { @Override public void commit() { commitImpl(true); } StreamTask(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,
                      final ThreadCache cache,
                      final Time time,
                      final Producer<byte[], byte[]> producer); @Override void resume(); @SuppressWarnings("unchecked") boolean process(); @Override void punctuate(final ProcessorNode node, final long timestamp); @Override void commit(); @Override void suspend(); @Override void close(boolean clean); @SuppressWarnings("unchecked") int addRecords(final TopicPartition partition, final Iterable<ConsumerRecord<byte[], byte[]>> records); void schedule(final long interval); }
StreamTask extends AbstractTask implements Punctuator { @Override public void commit() { commitImpl(true); } StreamTask(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,
                      final ThreadCache cache,
                      final Time time,
                      final Producer<byte[], byte[]> producer); @Override void resume(); @SuppressWarnings("unchecked") boolean process(); @Override void punctuate(final ProcessorNode node, final long timestamp); @Override void commit(); @Override void suspend(); @Override void close(boolean clean); @SuppressWarnings("unchecked") int addRecords(final TopicPartition partition, final Iterable<ConsumerRecord<byte[], byte[]>> records); void schedule(final long interval);  }