test_class.declarator: intSerializer = new IntegerSerializer()
test_class.modifier: private final
test_class.original_string: private final Serializer<Integer> intSerializer = new IntegerSerializer();
test_class.type: Serializer<Integer>
test_class.var_name: intSerializer
test_class.declarator: intDeserializer = new IntegerDeserializer()
test_class.modifier: private final
test_class.original_string: private final Deserializer<Integer> intDeserializer = new IntegerDeserializer();
test_class.type: Deserializer<Integer>
test_class.var_name: intDeserializer
test_class.declarator: bytesSerializer = new ByteArraySerializer()
test_class.modifier: private final
test_class.original_string: private final Serializer<byte[]> bytesSerializer = new ByteArraySerializer();
test_class.type: Serializer<byte[]>
test_class.var_name: bytesSerializer
test_class.declarator: partition1 = new TopicPartition("topic1", 1)
test_class.modifier: private final
test_class.original_string: private final TopicPartition partition1 = new TopicPartition("topic1", 1);
test_class.type: TopicPartition
test_class.var_name: partition1
test_class.declarator: partition2 = new TopicPartition("topic2", 1)
test_class.modifier: private final
test_class.original_string: private final TopicPartition partition2 = new TopicPartition("topic2", 1);
test_class.type: TopicPartition
test_class.var_name: partition2
test_class.declarator: partitions = Utils.mkSet(partition1, partition2)
test_class.modifier: private final
test_class.original_string: private final Set<TopicPartition> partitions = Utils.mkSet(partition1, partition2);
test_class.type: Set<TopicPartition>
test_class.var_name: partitions
test_class.declarator: source1 = new MockSourceNode<>(intDeserializer, intDeserializer)
test_class.modifier: private final
test_class.original_string: private final MockSourceNode<Integer, Integer> source1 = new MockSourceNode<>(intDeserializer, intDeserializer);
test_class.type: MockSourceNode<Integer, Integer>
test_class.var_name: source1
test_class.declarator: source2 = new MockSourceNode<>(intDeserializer, intDeserializer)
test_class.modifier: private final
test_class.original_string: private final MockSourceNode<Integer, Integer> source2 = new MockSourceNode<>(intDeserializer, intDeserializer);
test_class.type: MockSourceNode<Integer, Integer>
test_class.var_name: source2
test_class.declarator: processor = new MockProcessorNode<>(10L)
test_class.modifier: private final
test_class.original_string: private final MockProcessorNode<Integer, Integer>  processor = new MockProcessorNode<>(10L);
test_class.type: MockProcessorNode<Integer, Integer>
test_class.var_name: processor
test_class.declarator: topology = new ProcessorTopology(
              Arrays.asList((ProcessorNode) source1, (ProcessorNode) source2, (ProcessorNode) processor),
              new HashMap<String, SourceNode>() {
                  {
                      put("topic1", source1);
                      put("topic2", source2);
                  }
              },
              Collections.<StateStoreSupplier>emptyList()
      )
test_class.modifier: private final
test_class.original_string: private final ProcessorTopology topology = new ProcessorTopology(
              Arrays.asList((ProcessorNode) source1, (ProcessorNode) source2, (ProcessorNode) processor),
              new HashMap<String, SourceNode>() {
                  {
                      put("topic1", source1);
                      put("topic2", source2);
                  }
              },
              Collections.<StateStoreSupplier>emptyList()
      );
test_class.type: ProcessorTopology
test_class.var_name: topology
test_class.declarator: consumer = new MockConsumer<>(OffsetResetStrategy.EARLIEST)
test_class.modifier: private final
test_class.original_string: private final MockConsumer<byte[], byte[]> consumer = new MockConsumer<>(OffsetResetStrategy.EARLIEST);
test_class.type: MockConsumer<byte[], byte[]>
test_class.var_name: consumer
test_class.declarator: producer = new MockProducer<>(false, bytesSerializer, bytesSerializer)
test_class.modifier: private final
test_class.original_string: private final MockProducer<byte[], byte[]> producer = new MockProducer<>(false, bytesSerializer, bytesSerializer);
test_class.type: MockProducer<byte[], byte[]>
test_class.var_name: producer
test_class.declarator: restoreStateConsumer = new MockConsumer<>(OffsetResetStrategy.EARLIEST)
test_class.modifier: private final
test_class.original_string: private final MockConsumer<byte[], byte[]> restoreStateConsumer = new MockConsumer<>(OffsetResetStrategy.EARLIEST);
test_class.type: MockConsumer<byte[], byte[]>
test_class.var_name: restoreStateConsumer
test_class.declarator: recordValue = intSerializer.serialize(null, 10)
test_class.modifier: private final
test_class.original_string: private final byte[] recordValue = intSerializer.serialize(null, 10);
test_class.type: byte[]
test_class.var_name: recordValue
test_class.declarator: recordKey = intSerializer.serialize(null, 1)
test_class.modifier: private final
test_class.original_string: private final byte[] recordKey = intSerializer.serialize(null, 1);
test_class.type: byte[]
test_class.var_name: recordKey
test_class.file: streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java
test_class.identifier: StreamTaskTest
test_case.body: @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);
          }
      }
test_case.class_method_signature: StreamTaskTest.testMaybePunctuate()
test_case.full_signature: @SuppressWarnings("unchecked") @Test public void testMaybePunctuate()
test_case.identifier: testMaybePunctuate
test_case.invocations: toFile createTempDirectory createConfig addRecords records topic partition topic partition topic partition addRecords records topic partition topic partition topic partition assertTrue maybePunctuate assertEquals process assertEquals assertEquals assertFalse maybePunctuate assertEquals process assertEquals assertEquals assertTrue maybePunctuate assertEquals process assertEquals assertEquals assertFalse maybePunctuate assertEquals process assertEquals assertEquals assertTrue maybePunctuate assertEquals process assertEquals assertEquals assertFalse maybePunctuate assertEquals process assertEquals assertEquals assertFalse maybePunctuate checkAndClearPunctuateResult close delete
test_case.modifiers: @SuppressWarnings("unchecked") @Test public
test_case.parameters: ()
test_case.return: void
test_case.signature: void testMaybePunctuate()
focal_class.declarator: log = LoggerFactory.getLogger(StreamTask.class)
focal_class.modifier: private static final
focal_class.original_string: private static final Logger log = LoggerFactory.getLogger(StreamTask.class);
focal_class.type: Logger
focal_class.var_name: log
focal_class.declarator: DUMMY_RECORD = new ConsumerRecord<>(ProcessorContextImpl.NONEXIST_TOPIC, -1, -1L, null, null)
focal_class.modifier: private static final
focal_class.original_string: private static final ConsumerRecord<Object, Object> DUMMY_RECORD = new ConsumerRecord<>(ProcessorContextImpl.NONEXIST_TOPIC, -1, -1L, null, null);
focal_class.type: ConsumerRecord<Object, Object>
focal_class.var_name: DUMMY_RECORD
focal_class.declarator: maxBufferedSize
focal_class.modifier: private final
focal_class.original_string: private final int maxBufferedSize;
focal_class.type: int
focal_class.var_name: maxBufferedSize
focal_class.declarator: partitionGroup
focal_class.modifier: private final
focal_class.original_string: private final PartitionGroup partitionGroup;
focal_class.type: PartitionGroup
focal_class.var_name: partitionGroup
focal_class.declarator: recordInfo = new PartitionGroup.RecordInfo()
focal_class.modifier: private final
focal_class.original_string: private final PartitionGroup.RecordInfo recordInfo = new PartitionGroup.RecordInfo();
focal_class.type: PartitionGroup.RecordInfo
focal_class.var_name: recordInfo
focal_class.declarator: punctuationQueue
focal_class.modifier: private final
focal_class.original_string: private final PunctuationQueue punctuationQueue;
focal_class.type: PunctuationQueue
focal_class.var_name: punctuationQueue
focal_class.declarator: consumedOffsets
focal_class.modifier: private final
focal_class.original_string: private final Map<TopicPartition, Long> consumedOffsets;
focal_class.type: Map<TopicPartition, Long>
focal_class.var_name: consumedOffsets
focal_class.declarator: recordCollector
focal_class.modifier: private final
focal_class.original_string: private final RecordCollector recordCollector;
focal_class.type: RecordCollector
focal_class.var_name: recordCollector
focal_class.declarator: commitRequested = false
focal_class.modifier: private
focal_class.original_string: private boolean commitRequested = false;
focal_class.type: boolean
focal_class.var_name: commitRequested
focal_class.declarator: commitOffsetNeeded = false
focal_class.modifier: private
focal_class.original_string: private boolean commitOffsetNeeded = false;
focal_class.type: boolean
focal_class.var_name: commitOffsetNeeded
focal_class.declarator: currRecord = null
focal_class.modifier: private
focal_class.original_string: private StampedRecord currRecord = null;
focal_class.type: StampedRecord
focal_class.var_name: currRecord
focal_class.declarator: currNode = null
focal_class.modifier: private
focal_class.original_string: private ProcessorNode currNode = null;
focal_class.type: ProcessorNode
focal_class.var_name: currNode
focal_class.declarator: requiresPoll = true
focal_class.modifier: private
focal_class.original_string: private boolean requiresPoll = true;
focal_class.type: boolean
focal_class.var_name: requiresPoll
focal_class.file: streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java
focal_class.identifier: StreamTask
focal_class.interfaces: implements Punctuator
focal_class.class_method_signature: StreamTask.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)
focal_class.full_signature: public  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)
focal_class.identifier: StreamTask
focal_class.modifiers: public
focal_class.parameters: (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)
focal_class.signature:  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)
focal_class.class_method_signature: StreamTask.addRecords(TopicPartition partition, Iterable<ConsumerRecord<byte[], byte[]>> records)
focal_class.full_signature: @SuppressWarnings("unchecked") public void addRecords(TopicPartition partition, Iterable<ConsumerRecord<byte[], byte[]>> records)
focal_class.identifier: addRecords
focal_class.modifiers: @SuppressWarnings("unchecked") public
focal_class.parameters: (TopicPartition partition, Iterable<ConsumerRecord<byte[], byte[]>> records)
focal_class.return: void
focal_class.signature: void addRecords(TopicPartition partition, Iterable<ConsumerRecord<byte[], byte[]>> records)
focal_class.class_method_signature: StreamTask.process()
focal_class.full_signature: @SuppressWarnings("unchecked") public int process()
focal_class.identifier: process
focal_class.modifiers: @SuppressWarnings("unchecked") public
focal_class.parameters: ()
focal_class.return: int
focal_class.signature: int process()
focal_class.class_method_signature: StreamTask.requiresPoll()
focal_class.full_signature: public boolean requiresPoll()
focal_class.identifier: requiresPoll
focal_class.modifiers: public
focal_class.parameters: ()
focal_class.return: boolean
focal_class.signature: boolean requiresPoll()
focal_class.class_method_signature: StreamTask.maybePunctuate()
focal_class.full_signature: public boolean maybePunctuate()
focal_class.identifier: maybePunctuate
focal_class.modifiers: public
focal_class.parameters: ()
focal_class.return: boolean
focal_class.signature: boolean maybePunctuate()
focal_class.class_method_signature: StreamTask.punctuate(ProcessorNode node, long timestamp)
focal_class.full_signature: @Override public void punctuate(ProcessorNode node, long timestamp)
focal_class.identifier: punctuate
focal_class.modifiers: @Override public
focal_class.parameters: (ProcessorNode node, long timestamp)
focal_class.return: void
focal_class.signature: void punctuate(ProcessorNode node, long timestamp)
focal_class.class_method_signature: StreamTask.record()
focal_class.full_signature: public StampedRecord record()
focal_class.identifier: record
focal_class.modifiers: public
focal_class.parameters: ()
focal_class.return: StampedRecord
focal_class.signature: StampedRecord record()
focal_class.class_method_signature: StreamTask.node()
focal_class.full_signature: public ProcessorNode node()
focal_class.identifier: node
focal_class.modifiers: public
focal_class.parameters: ()
focal_class.return: ProcessorNode
focal_class.signature: ProcessorNode node()
focal_class.class_method_signature: StreamTask.commit()
focal_class.full_signature: public void commit()
focal_class.identifier: commit
focal_class.modifiers: public
focal_class.parameters: ()
focal_class.return: void
focal_class.signature: void commit()
focal_class.class_method_signature: StreamTask.commitNeeded()
focal_class.full_signature: public boolean commitNeeded()
focal_class.identifier: commitNeeded
focal_class.modifiers: public
focal_class.parameters: ()
focal_class.return: boolean
focal_class.signature: boolean commitNeeded()
focal_class.class_method_signature: StreamTask.needCommit()
focal_class.full_signature: public void needCommit()
focal_class.identifier: needCommit
focal_class.modifiers: public
focal_class.parameters: ()
focal_class.return: void
focal_class.signature: void needCommit()
focal_class.class_method_signature: StreamTask.schedule(long interval)
focal_class.full_signature: public void schedule(long interval)
focal_class.identifier: schedule
focal_class.modifiers: public
focal_class.parameters: (long interval)
focal_class.return: void
focal_class.signature: void schedule(long interval)
focal_class.class_method_signature: StreamTask.close()
focal_class.full_signature: @Override public void close()
focal_class.identifier: close
focal_class.modifiers: @Override public
focal_class.parameters: ()
focal_class.return: void
focal_class.signature: void close()
focal_class.class_method_signature: StreamTask.recordCollectorOffsets()
focal_class.full_signature: @Override protected Map<TopicPartition, Long> recordCollectorOffsets()
focal_class.identifier: recordCollectorOffsets
focal_class.modifiers: @Override protected
focal_class.parameters: ()
focal_class.return: Map<TopicPartition, Long>
focal_class.signature: Map<TopicPartition, Long> recordCollectorOffsets()
focal_class.class_method_signature: StreamTask.createRecordQueue(TopicPartition partition, SourceNode source)
focal_class.full_signature: private RecordQueue createRecordQueue(TopicPartition partition, SourceNode source)
focal_class.identifier: createRecordQueue
focal_class.modifiers: private
focal_class.parameters: (TopicPartition partition, SourceNode source)
focal_class.return: RecordQueue
focal_class.signature: RecordQueue createRecordQueue(TopicPartition partition, SourceNode source)
focal_class.class_method_signature: StreamTask.forward(K key, V value)
focal_class.full_signature: @SuppressWarnings("unchecked") public void forward(K key, V value)
focal_class.identifier: forward
focal_class.modifiers: @SuppressWarnings("unchecked") public
focal_class.parameters: (K key, V value)
focal_class.return: void
focal_class.signature: void forward(K key, V value)
focal_class.class_method_signature: StreamTask.forward(K key, V value, int childIndex)
focal_class.full_signature: @SuppressWarnings("unchecked") public void forward(K key, V value, int childIndex)
focal_class.identifier: forward
focal_class.modifiers: @SuppressWarnings("unchecked") public
focal_class.parameters: (K key, V value, int childIndex)
focal_class.return: void
focal_class.signature: void forward(K key, V value, int childIndex)
focal_class.class_method_signature: StreamTask.forward(K key, V value, String childName)
focal_class.full_signature: @SuppressWarnings("unchecked") public void forward(K key, V value, String childName)
focal_class.identifier: forward
focal_class.modifiers: @SuppressWarnings("unchecked") public
focal_class.parameters: (K key, V value, String childName)
focal_class.return: void
focal_class.signature: void forward(K key, V value, String childName)
focal_class.superclass: extends AbstractTask
focal_method.body: public boolean maybePunctuate() {
          long timestamp = partitionGroup.timestamp();
  
          // if the timestamp is not known yet, meaning there is not enough data accumulated
          // to reason stream partition time, then skip.
          if (timestamp == TimestampTracker.NOT_KNOWN)
              return false;
          else
              return punctuationQueue.mayPunctuate(timestamp, this);
      }
focal_method.class_method_signature: StreamTask.maybePunctuate()
focal_method.full_signature: public boolean maybePunctuate()
focal_method.identifier: maybePunctuate
focal_method.invocations: timestamp mayPunctuate
focal_method.modifiers: public
focal_method.parameters: ()
focal_method.return: boolean
focal_method.signature: boolean maybePunctuate()