@Test public void shouldCallPunctuateOnPassedInProcessorNode() throws Exception { task.punctuate(processor, 5); assertThat(processor.punctuatedAt, equalTo(5L)); task.punctuate(processor, 10); assertThat(processor.punctuatedAt, equalTo(10L)); }
@Override public void punctuate(final ProcessorNode node, final long timestamp) { if (processorContext.currentNode() != null) { throw new IllegalStateException(String.format("%s Current node is not null", logPrefix)); } updateProcessorContext(new StampedRecord(DUMMY_RECORD, timestamp), node); log.trace("{} Punctuating processor {} with timestamp {}", logPrefix, node.name(), timestamp); try { node.punctuate(timestamp); } catch (final KafkaException e) { throw new StreamsException(String.format("%s Exception caught while punctuating processor '%s'", logPrefix, node.name()), e); } finally { processorContext.setCurrentNode(null); } }
StreamTask extends AbstractTask implements Punctuator { @Override public void punctuate(final ProcessorNode node, final long timestamp) { if (processorContext.currentNode() != null) { throw new IllegalStateException(String.format("%s Current node is not null", logPrefix)); } updateProcessorContext(new StampedRecord(DUMMY_RECORD, timestamp), node); log.trace("{} Punctuating processor {} with timestamp {}", logPrefix, node.name(), timestamp); try { node.punctuate(timestamp); } catch (final KafkaException e) { throw new StreamsException(String.format("%s Exception caught while punctuating processor '%s'", logPrefix, node.name()), e); } finally { processorContext.setCurrentNode(null); } } }
StreamTask extends AbstractTask implements Punctuator { @Override public void punctuate(final ProcessorNode node, final long timestamp) { if (processorContext.currentNode() != null) { throw new IllegalStateException(String.format("%s Current node is not null", logPrefix)); } updateProcessorContext(new StampedRecord(DUMMY_RECORD, timestamp), node); log.trace("{} Punctuating processor {} with timestamp {}", logPrefix, node.name(), timestamp); try { node.punctuate(timestamp); } catch (final KafkaException e) { throw new StreamsException(String.format("%s Exception caught while punctuating processor '%s'", logPrefix, node.name()), e); } finally { processorContext.setCurrentNode(null); } } 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 punctuate(final ProcessorNode node, final long timestamp) { if (processorContext.currentNode() != null) { throw new IllegalStateException(String.format("%s Current node is not null", logPrefix)); } updateProcessorContext(new StampedRecord(DUMMY_RECORD, timestamp), node); log.trace("{} Punctuating processor {} with timestamp {}", logPrefix, node.name(), timestamp); try { node.punctuate(timestamp); } catch (final KafkaException e) { throw new StreamsException(String.format("%s Exception caught while punctuating processor '%s'", logPrefix, node.name()), e); } finally { processorContext.setCurrentNode(null); } } 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 punctuate(final ProcessorNode node, final long timestamp) { if (processorContext.currentNode() != null) { throw new IllegalStateException(String.format("%s Current node is not null", logPrefix)); } updateProcessorContext(new StampedRecord(DUMMY_RECORD, timestamp), node); log.trace("{} Punctuating processor {} with timestamp {}", logPrefix, node.name(), timestamp); try { node.punctuate(timestamp); } catch (final KafkaException e) { throw new StreamsException(String.format("%s Exception caught while punctuating processor '%s'", logPrefix, node.name()), e); } finally { processorContext.setCurrentNode(null); } } 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);  }