@SuppressWarnings("unchecked") @Test public void testMaybePunctuate() throws Exception { File baseDir = Files.createTempDirectory("test").toFile(); try { StreamsConfig config = createConfig(baseDir); StreamTask task = new StreamTask(new TaskId(0, 0), "applicationId", partitions, topology, consumer, producer, restoreStateConsumer, config, null); task.addRecords(partition1, records( new ConsumerRecord<>(partition1.topic(), partition1.partition(), 20, 0L, TimestampType.CREATE_TIME, 0L, 0, 0, recordKey, recordValue), new ConsumerRecord<>(partition1.topic(), partition1.partition(), 30, 0L, TimestampType.CREATE_TIME, 0L, 0, 0, recordKey, recordValue), new ConsumerRecord<>(partition1.topic(), partition1.partition(), 40, 0L, TimestampType.CREATE_TIME, 0L, 0, 0, recordKey, recordValue) )); task.addRecords(partition2, records( new ConsumerRecord<>(partition2.topic(), partition2.partition(), 25, 0L, TimestampType.CREATE_TIME, 0L, 0, 0, recordKey, recordValue), new ConsumerRecord<>(partition2.topic(), partition2.partition(), 35, 0L, TimestampType.CREATE_TIME, 0L, 0, 0, recordKey, recordValue), new ConsumerRecord<>(partition2.topic(), partition2.partition(), 45, 0L, TimestampType.CREATE_TIME, 0L, 0, 0, recordKey, recordValue) )); assertTrue(task.maybePunctuate()); assertEquals(5, task.process()); assertEquals(1, source1.numReceived); assertEquals(0, source2.numReceived); assertFalse(task.maybePunctuate()); assertEquals(4, task.process()); assertEquals(1, source1.numReceived); assertEquals(1, source2.numReceived); assertTrue(task.maybePunctuate()); assertEquals(3, task.process()); assertEquals(2, source1.numReceived); assertEquals(1, source2.numReceived); assertFalse(task.maybePunctuate()); assertEquals(2, task.process()); assertEquals(2, source1.numReceived); assertEquals(2, source2.numReceived); assertTrue(task.maybePunctuate()); assertEquals(1, task.process()); assertEquals(3, source1.numReceived); assertEquals(2, source2.numReceived); assertFalse(task.maybePunctuate()); assertEquals(0, task.process()); assertEquals(3, source1.numReceived); assertEquals(3, source2.numReceived); assertFalse(task.maybePunctuate()); processor.supplier.checkAndClearPunctuateResult(20L, 30L, 40L); task.close(); } finally { Utils.delete(baseDir); } }
public boolean maybePunctuate() { long timestamp = partitionGroup.timestamp(); if (timestamp == TimestampTracker.NOT_KNOWN) return false; else return punctuationQueue.mayPunctuate(timestamp, this); }
StreamTask extends AbstractTask implements Punctuator { public boolean maybePunctuate() { long timestamp = partitionGroup.timestamp(); if (timestamp == TimestampTracker.NOT_KNOWN) return false; else return punctuationQueue.mayPunctuate(timestamp, this); } }
StreamTask extends AbstractTask implements Punctuator { public boolean maybePunctuate() { long timestamp = partitionGroup.timestamp(); if (timestamp == TimestampTracker.NOT_KNOWN) return false; else return punctuationQueue.mayPunctuate(timestamp, this); } StreamTask(TaskId id,
                      String applicationId,
                      Collection<TopicPartition> partitions,
                      ProcessorTopology topology,
                      Consumer<byte[], byte[]> consumer,
                      Producer<byte[], byte[]> producer,
                      Consumer<byte[], byte[]> restoreConsumer,
                      StreamsConfig config,
                      StreamsMetrics metrics); }
StreamTask extends AbstractTask implements Punctuator { public boolean maybePunctuate() { long timestamp = partitionGroup.timestamp(); if (timestamp == TimestampTracker.NOT_KNOWN) return false; else return punctuationQueue.mayPunctuate(timestamp, this); } StreamTask(TaskId id,
                      String applicationId,
                      Collection<TopicPartition> partitions,
                      ProcessorTopology topology,
                      Consumer<byte[], byte[]> consumer,
                      Producer<byte[], byte[]> producer,
                      Consumer<byte[], byte[]> restoreConsumer,
                      StreamsConfig config,
                      StreamsMetrics metrics); @SuppressWarnings("unchecked") void addRecords(TopicPartition partition, Iterable<ConsumerRecord<byte[], byte[]>> records); @SuppressWarnings("unchecked") int process(); boolean requiresPoll(); boolean maybePunctuate(); @Override void punctuate(ProcessorNode node, long timestamp); StampedRecord record(); ProcessorNode node(); void commit(); boolean commitNeeded(); void needCommit(); void schedule(long interval); @Override void close(); @SuppressWarnings("unchecked") void forward(K key, V value); @SuppressWarnings("unchecked") void forward(K key, V value, int childIndex); @SuppressWarnings("unchecked") void forward(K key, V value, String childName); }
StreamTask extends AbstractTask implements Punctuator { public boolean maybePunctuate() { long timestamp = partitionGroup.timestamp(); if (timestamp == TimestampTracker.NOT_KNOWN) return false; else return punctuationQueue.mayPunctuate(timestamp, this); } StreamTask(TaskId id,
                      String applicationId,
                      Collection<TopicPartition> partitions,
                      ProcessorTopology topology,
                      Consumer<byte[], byte[]> consumer,
                      Producer<byte[], byte[]> producer,
                      Consumer<byte[], byte[]> restoreConsumer,
                      StreamsConfig config,
                      StreamsMetrics metrics); @SuppressWarnings("unchecked") void addRecords(TopicPartition partition, Iterable<ConsumerRecord<byte[], byte[]>> records); @SuppressWarnings("unchecked") int process(); boolean requiresPoll(); boolean maybePunctuate(); @Override void punctuate(ProcessorNode node, long timestamp); StampedRecord record(); ProcessorNode node(); void commit(); boolean commitNeeded(); void needCommit(); void schedule(long interval); @Override void close(); @SuppressWarnings("unchecked") void forward(K key, V value); @SuppressWarnings("unchecked") void forward(K key, V value, int childIndex); @SuppressWarnings("unchecked") void forward(K key, V value, String childName);  }