@SuppressWarnings("unchecked") @Test public void testMaybePunctuate() throws Exception { 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()); assertTrue(task.process()); assertEquals(5, task.numBuffered()); assertEquals(1, source1.numReceived); assertEquals(0, source2.numReceived); assertFalse(task.maybePunctuate()); assertTrue(task.process()); assertEquals(4, task.numBuffered()); assertEquals(1, source1.numReceived); assertEquals(1, source2.numReceived); assertTrue(task.maybePunctuate()); assertTrue(task.process()); assertEquals(3, task.numBuffered()); assertEquals(2, source1.numReceived); assertEquals(1, source2.numReceived); assertFalse(task.maybePunctuate()); assertTrue(task.process()); assertEquals(2, task.numBuffered()); assertEquals(2, source1.numReceived); assertEquals(2, source2.numReceived); assertTrue(task.maybePunctuate()); assertTrue(task.process()); assertEquals(1, task.numBuffered()); assertEquals(3, source1.numReceived); assertEquals(2, source2.numReceived); assertFalse(task.maybePunctuate()); assertTrue(task.process()); assertEquals(0, task.numBuffered()); assertEquals(3, source1.numReceived); assertEquals(3, source2.numReceived); assertFalse(task.process()); assertFalse(task.maybePunctuate()); processor.supplier.checkAndClearPunctuateResult(20L, 30L, 40L); }
boolean maybePunctuate() { final long timestamp = partitionGroup.timestamp(); if (timestamp == TimestampTracker.NOT_KNOWN) { return false; } else { return punctuationQueue.mayPunctuate(timestamp, this); } }
StreamTask extends AbstractTask implements Punctuator { boolean maybePunctuate() { final long timestamp = partitionGroup.timestamp(); if (timestamp == TimestampTracker.NOT_KNOWN) { return false; } else { return punctuationQueue.mayPunctuate(timestamp, this); } } }
StreamTask extends AbstractTask implements Punctuator { boolean maybePunctuate() { final long timestamp = partitionGroup.timestamp(); if (timestamp == TimestampTracker.NOT_KNOWN) { return false; } else { return punctuationQueue.mayPunctuate(timestamp, this); } } 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 { boolean maybePunctuate() { final long timestamp = partitionGroup.timestamp(); if (timestamp == TimestampTracker.NOT_KNOWN) { return false; } else { return punctuationQueue.mayPunctuate(timestamp, this); } } 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 { boolean maybePunctuate() { final long timestamp = partitionGroup.timestamp(); if (timestamp == TimestampTracker.NOT_KNOWN) { return false; } else { return punctuationQueue.mayPunctuate(timestamp, this); } } 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);  }