test_class.declarator: taskId = new TaskId(0, 1)
test_class.modifier: private final
test_class.original_string: private final TaskId taskId = new TaskId(0, 1);
test_class.type: TaskId
test_class.var_name: taskId
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: applicationId = "test-application"
test_class.modifier: private final
test_class.original_string: private final String applicationId = "test-application";
test_class.type: String
test_class.var_name: applicationId
test_class.declarator: storeName1 = "store1"
test_class.modifier: private final
test_class.original_string: private final String storeName1 = "store1";
test_class.type: String
test_class.var_name: storeName1
test_class.declarator: storeName2 = "store2"
test_class.modifier: private final
test_class.original_string: private final String storeName2 = "store2";
test_class.type: String
test_class.var_name: storeName2
test_class.declarator: storeChangelogTopicName1 = ProcessorStateManager.storeChangelogTopic(applicationId, storeName1)
test_class.modifier: private final
test_class.original_string: private final String storeChangelogTopicName1 = ProcessorStateManager.storeChangelogTopic(applicationId, storeName1);
test_class.type: String
test_class.var_name: storeChangelogTopicName1
test_class.declarator: storeChangelogTopicName2 = ProcessorStateManager.storeChangelogTopic(applicationId, storeName2)
test_class.modifier: private final
test_class.original_string: private final String storeChangelogTopicName2 = ProcessorStateManager.storeChangelogTopic(applicationId, storeName2);
test_class.type: String
test_class.var_name: storeChangelogTopicName2
test_class.declarator: partition1 = new TopicPartition(storeChangelogTopicName1, 1)
test_class.modifier: private final
test_class.original_string: private final TopicPartition partition1 = new TopicPartition(storeChangelogTopicName1, 1);
test_class.type: TopicPartition
test_class.var_name: partition1
test_class.declarator: partition2 = new TopicPartition(storeChangelogTopicName2, 1)
test_class.modifier: private final
test_class.original_string: private final TopicPartition partition2 = new TopicPartition(storeChangelogTopicName2, 1);
test_class.type: TopicPartition
test_class.var_name: partition2
test_class.declarator: topicPartitions = Collections.emptySet()
test_class.modifier: private final
test_class.original_string: private final Set<TopicPartition> topicPartitions = Collections.emptySet();
test_class.type: Set<TopicPartition>
test_class.var_name: topicPartitions
test_class.declarator: topology = new ProcessorTopology(
              Collections.<ProcessorNode>emptyList(),
              Collections.<String, SourceNode>emptyMap(),
              Utils.<StateStoreSupplier>mkList(
                      new MockStateStoreSupplier(storeName1, false),
                      new MockStateStoreSupplier(storeName2, true)
              )
      )
test_class.modifier: private final
test_class.original_string: private final ProcessorTopology topology = new ProcessorTopology(
              Collections.<ProcessorNode>emptyList(),
              Collections.<String, SourceNode>emptyMap(),
              Utils.<StateStoreSupplier>mkList(
                      new MockStateStoreSupplier(storeName1, false),
                      new MockStateStoreSupplier(storeName2, true)
              )
      );
test_class.type: ProcessorTopology
test_class.var_name: topology
test_class.declarator: ktable = new TopicPartition("ktable1", 0)
test_class.modifier: private final
test_class.original_string: private final TopicPartition ktable = new TopicPartition("ktable1", 0);
test_class.type: TopicPartition
test_class.var_name: ktable
test_class.declarator: ktablePartitions = Utils.mkSet(ktable)
test_class.modifier: private final
test_class.original_string: private final Set<TopicPartition> ktablePartitions = Utils.mkSet(ktable);
test_class.type: Set<TopicPartition>
test_class.var_name: ktablePartitions
test_class.declarator: ktableTopology = new ProcessorTopology(
              Collections.<ProcessorNode>emptyList(),
              Collections.<String, SourceNode>emptyMap(),
              Utils.<StateStoreSupplier>mkList(
                      new MockStateStoreSupplier(ktable.topic(), true, false)
              )
      )
test_class.modifier: private final
test_class.original_string: private final ProcessorTopology ktableTopology = new ProcessorTopology(
              Collections.<ProcessorNode>emptyList(),
              Collections.<String, SourceNode>emptyMap(),
              Utils.<StateStoreSupplier>mkList(
                      new MockStateStoreSupplier(ktable.topic(), true, false)
              )
      );
test_class.type: ProcessorTopology
test_class.var_name: ktableTopology
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: restoreStateConsumer = new ProcessorStateManagerTest.MockRestoreConsumer()
test_class.modifier: private final
test_class.original_string: private final ProcessorStateManagerTest.MockRestoreConsumer restoreStateConsumer = new ProcessorStateManagerTest.MockRestoreConsumer();
test_class.type: ProcessorStateManagerTest.MockRestoreConsumer
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/StandbyTaskTest.java
test_class.identifier: StandbyTaskTest
test_case.body: @SuppressWarnings("unchecked")
      @Test
      public void testUpdate() throws Exception {
          File baseDir = Files.createTempDirectory("test").toFile();
          try {
              StreamsConfig config = createConfig(baseDir);
              StandbyTask task = new StandbyTask(taskId, applicationId, topicPartitions, topology, consumer, restoreStateConsumer, config, null);
  
              restoreStateConsumer.assign(new ArrayList<>(task.changeLogPartitions()));
  
              for (ConsumerRecord<Integer, Integer> record : Arrays.asList(
                      new ConsumerRecord<>(partition2.topic(), partition2.partition(), 10, 0L, TimestampType.CREATE_TIME, 0L, 0, 0, 1, 100),
                      new ConsumerRecord<>(partition2.topic(), partition2.partition(), 20, 0L, TimestampType.CREATE_TIME, 0L, 0, 0, 2, 100),
                      new ConsumerRecord<>(partition2.topic(), partition2.partition(), 30, 0L, TimestampType.CREATE_TIME, 0L, 0, 0, 3, 100))) {
                  restoreStateConsumer.bufferRecord(record);
              }
  
              for (Map.Entry<TopicPartition, Long> entry : task.checkpointedOffsets().entrySet()) {
                  TopicPartition partition = entry.getKey();
                  long offset = entry.getValue();
                  if (offset >= 0) {
                      restoreStateConsumer.seek(partition, offset);
                  } else {
                      restoreStateConsumer.seekToBeginning(singleton(partition));
                  }
              }
  
              task.update(partition2, restoreStateConsumer.poll(100).records(partition2));
  
              StandbyContextImpl context = (StandbyContextImpl) task.context();
              MockStateStoreSupplier.MockStateStore store1 =
                      (MockStateStoreSupplier.MockStateStore) context.getStateMgr().getStore(storeName1);
              MockStateStoreSupplier.MockStateStore store2 =
                      (MockStateStoreSupplier.MockStateStore) context.getStateMgr().getStore(storeName2);
  
              assertEquals(Collections.emptyList(), store1.keys);
              assertEquals(Utils.mkList(1, 2, 3), store2.keys);
  
              task.close();
  
              File taskDir = new File(StreamThread.makeStateDir(applicationId, baseDir.getCanonicalPath()), taskId.toString());
              OffsetCheckpoint checkpoint = new OffsetCheckpoint(new File(taskDir, ProcessorStateManager.CHECKPOINT_FILE_NAME));
              Map<TopicPartition, Long> offsets = checkpoint.read();
  
              assertEquals(1, offsets.size());
              assertEquals(new Long(30L + 1L), offsets.get(partition2));
  
          } finally {
              Utils.delete(baseDir);
          }
      }
test_case.class_method_signature: StandbyTaskTest.testUpdate()
test_case.full_signature: @SuppressWarnings("unchecked") @Test public void testUpdate()
test_case.identifier: testUpdate
test_case.invocations: toFile createTempDirectory createConfig assign changeLogPartitions asList topic partition topic partition topic partition bufferRecord entrySet checkpointedOffsets getKey getValue seek seekToBeginning singleton update records poll context getStore getStateMgr getStore getStateMgr assertEquals emptyList assertEquals mkList close makeStateDir getCanonicalPath toString read assertEquals size assertEquals get delete
test_case.modifiers: @SuppressWarnings("unchecked") @Test public
test_case.parameters: ()
test_case.return: void
test_case.signature: void testUpdate()
focal_class.declarator: checkpointedOffsets
focal_class.modifier: private final
focal_class.original_string: private final Map<TopicPartition, Long> checkpointedOffsets;
focal_class.type: Map<TopicPartition, Long>
focal_class.var_name: checkpointedOffsets
focal_class.file: streams/src/main/java/org/apache/kafka/streams/processor/internals/StandbyTask.java
focal_class.identifier: StandbyTask
focal_class.class_method_signature: StandbyTask.StandbyTask(TaskId id,
                         String applicationId,
                         Collection<TopicPartition> partitions,
                         ProcessorTopology topology,
                         Consumer<byte[], byte[]> consumer,
                         Consumer<byte[], byte[]> restoreConsumer,
                         StreamsConfig config,
                         StreamsMetrics metrics)
focal_class.full_signature: public  StandbyTask(TaskId id,
                         String applicationId,
                         Collection<TopicPartition> partitions,
                         ProcessorTopology topology,
                         Consumer<byte[], byte[]> consumer,
                         Consumer<byte[], byte[]> restoreConsumer,
                         StreamsConfig config,
                         StreamsMetrics metrics)
focal_class.identifier: StandbyTask
focal_class.modifiers: public
focal_class.parameters: (TaskId id,
                         String applicationId,
                         Collection<TopicPartition> partitions,
                         ProcessorTopology topology,
                         Consumer<byte[], byte[]> consumer,
                         Consumer<byte[], byte[]> restoreConsumer,
                         StreamsConfig config,
                         StreamsMetrics metrics)
focal_class.signature:  StandbyTask(TaskId id,
                         String applicationId,
                         Collection<TopicPartition> partitions,
                         ProcessorTopology topology,
                         Consumer<byte[], byte[]> consumer,
                         Consumer<byte[], byte[]> restoreConsumer,
                         StreamsConfig config,
                         StreamsMetrics metrics)
focal_class.class_method_signature: StandbyTask.checkpointedOffsets()
focal_class.full_signature: public Map<TopicPartition, Long> checkpointedOffsets()
focal_class.identifier: checkpointedOffsets
focal_class.modifiers: public
focal_class.parameters: ()
focal_class.return: Map<TopicPartition, Long>
focal_class.signature: Map<TopicPartition, Long> checkpointedOffsets()
focal_class.class_method_signature: StandbyTask.changeLogPartitions()
focal_class.full_signature: public Collection<TopicPartition> changeLogPartitions()
focal_class.identifier: changeLogPartitions
focal_class.modifiers: public
focal_class.parameters: ()
focal_class.return: Collection<TopicPartition>
focal_class.signature: Collection<TopicPartition> changeLogPartitions()
focal_class.class_method_signature: StandbyTask.update(TopicPartition partition, List<ConsumerRecord<byte[], byte[]>> records)
focal_class.full_signature: public List<ConsumerRecord<byte[], byte[]>> update(TopicPartition partition, List<ConsumerRecord<byte[], byte[]>> records)
focal_class.identifier: update
focal_class.modifiers: public
focal_class.parameters: (TopicPartition partition, List<ConsumerRecord<byte[], byte[]>> records)
focal_class.return: List<ConsumerRecord<byte[], byte[]>>
focal_class.signature: List<ConsumerRecord<byte[], byte[]>> update(TopicPartition partition, List<ConsumerRecord<byte[], byte[]>> records)
focal_class.class_method_signature: StandbyTask.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.superclass: extends AbstractTask
focal_method.body: public List<ConsumerRecord<byte[], byte[]>> update(TopicPartition partition, List<ConsumerRecord<byte[], byte[]>> records) {
          return stateMgr.updateStandbyStates(partition, records);
      }
focal_method.class_method_signature: StandbyTask.update(TopicPartition partition, List<ConsumerRecord<byte[], byte[]>> records)
focal_method.full_signature: public List<ConsumerRecord<byte[], byte[]>> update(TopicPartition partition, List<ConsumerRecord<byte[], byte[]>> records)
focal_method.identifier: update
focal_method.invocations: updateStandbyStates
focal_method.modifiers: public
focal_method.parameters: (TopicPartition partition, List<ConsumerRecord<byte[], byte[]>> records)
focal_method.return: List<ConsumerRecord<byte[], byte[]>>
focal_method.signature: List<ConsumerRecord<byte[], byte[]>> update(TopicPartition partition, List<ConsumerRecord<byte[], byte[]>> records)