@Test public void shouldFlushRecordCollectorOnFlushState() throws Exception { final AtomicBoolean flushed = new AtomicBoolean(false); final StreamsMetrics streamsMetrics = new MockStreamsMetrics(new Metrics()); 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 void flush() { flushed.set(true); } }; } }; streamTask.flushState(); assertTrue(flushed.get()); }
@Override protected void flushState() { log.trace("{} Flushing state and producer", logPrefix); super.flushState(); recordCollector.flush(); }
StreamTask extends AbstractTask implements Punctuator { @Override protected void flushState() { log.trace("{} Flushing state and producer", logPrefix); super.flushState(); recordCollector.flush(); } }
StreamTask extends AbstractTask implements Punctuator { @Override protected void flushState() { log.trace("{} Flushing state and producer", logPrefix); super.flushState(); recordCollector.flush(); } 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 protected void flushState() { log.trace("{} Flushing state and producer", logPrefix); super.flushState(); recordCollector.flush(); } 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 protected void flushState() { log.trace("{} Flushing state and producer", logPrefix); super.flushState(); recordCollector.flush(); } 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);  }