test_class.declarator: clientId = "clientId"
test_class.modifier: private final
test_class.original_string: private final String clientId = "clientId";
test_class.type: String
test_class.var_name: clientId
test_class.declarator: applicationId = "stream-thread-test"
test_class.modifier: private final
test_class.original_string: private final String applicationId = "stream-thread-test";
test_class.type: String
test_class.var_name: applicationId
test_class.declarator: processId = UUID.randomUUID()
test_class.modifier: private final
test_class.original_string: private final UUID processId = UUID.randomUUID();
test_class.type: UUID
test_class.var_name: processId
test_class.declarator: t1p1 = new TopicPartition("topic1", 1)
test_class.modifier: private
test_class.original_string: private TopicPartition t1p1 = new TopicPartition("topic1", 1);
test_class.type: TopicPartition
test_class.var_name: t1p1
test_class.declarator: t1p2 = new TopicPartition("topic1", 2)
test_class.modifier: private
test_class.original_string: private TopicPartition t1p2 = new TopicPartition("topic1", 2);
test_class.type: TopicPartition
test_class.var_name: t1p2
test_class.declarator: t2p1 = new TopicPartition("topic2", 1)
test_class.modifier: private
test_class.original_string: private TopicPartition t2p1 = new TopicPartition("topic2", 1);
test_class.type: TopicPartition
test_class.var_name: t2p1
test_class.declarator: t2p2 = new TopicPartition("topic2", 2)
test_class.modifier: private
test_class.original_string: private TopicPartition t2p2 = new TopicPartition("topic2", 2);
test_class.type: TopicPartition
test_class.var_name: t2p2
test_class.declarator: t3p1 = new TopicPartition("topic3", 1)
test_class.modifier: private
test_class.original_string: private TopicPartition t3p1 = new TopicPartition("topic3", 1);
test_class.type: TopicPartition
test_class.var_name: t3p1
test_class.declarator: t3p2 = new TopicPartition("topic3", 2)
test_class.modifier: private
test_class.original_string: private TopicPartition t3p2 = new TopicPartition("topic3", 2);
test_class.type: TopicPartition
test_class.var_name: t3p2
test_class.declarator: infos = Arrays.asList(
              new PartitionInfo("topic1", 0, Node.noNode(), new Node[0], new Node[0]),
              new PartitionInfo("topic1", 1, Node.noNode(), new Node[0], new Node[0]),
              new PartitionInfo("topic1", 2, Node.noNode(), new Node[0], new Node[0]),
              new PartitionInfo("topic2", 0, Node.noNode(), new Node[0], new Node[0]),
              new PartitionInfo("topic2", 1, Node.noNode(), new Node[0], new Node[0]),
              new PartitionInfo("topic2", 2, Node.noNode(), new Node[0], new Node[0]),
              new PartitionInfo("topic3", 0, Node.noNode(), new Node[0], new Node[0]),
              new PartitionInfo("topic3", 1, Node.noNode(), new Node[0], new Node[0]),
              new PartitionInfo("topic3", 2, Node.noNode(), new Node[0], new Node[0])
      )
test_class.modifier: private
test_class.original_string: private List<PartitionInfo> infos = Arrays.asList(
              new PartitionInfo("topic1", 0, Node.noNode(), new Node[0], new Node[0]),
              new PartitionInfo("topic1", 1, Node.noNode(), new Node[0], new Node[0]),
              new PartitionInfo("topic1", 2, Node.noNode(), new Node[0], new Node[0]),
              new PartitionInfo("topic2", 0, Node.noNode(), new Node[0], new Node[0]),
              new PartitionInfo("topic2", 1, Node.noNode(), new Node[0], new Node[0]),
              new PartitionInfo("topic2", 2, Node.noNode(), new Node[0], new Node[0]),
              new PartitionInfo("topic3", 0, Node.noNode(), new Node[0], new Node[0]),
              new PartitionInfo("topic3", 1, Node.noNode(), new Node[0], new Node[0]),
              new PartitionInfo("topic3", 2, Node.noNode(), new Node[0], new Node[0])
      );
test_class.type: List<PartitionInfo>
test_class.var_name: infos
test_class.declarator: metadata = new Cluster(Arrays.asList(Node.noNode()), infos, Collections.<String>emptySet())
test_class.modifier: private
test_class.original_string: private Cluster metadata = new Cluster(Arrays.asList(Node.noNode()), infos, Collections.<String>emptySet());
test_class.type: Cluster
test_class.var_name: metadata
test_class.declarator: subscription =
              new PartitionAssignor.Subscription(Arrays.asList("topic1", "topic2", "topic3"), subscriptionUserData())
test_class.modifier: private final
test_class.original_string: private final PartitionAssignor.Subscription subscription =
              new PartitionAssignor.Subscription(Arrays.asList("topic1", "topic2", "topic3"), subscriptionUserData());
test_class.type: PartitionAssignor.Subscription
test_class.var_name: subscription
test_class.declarator: task1 = new TaskId(0, 1)
test_class.modifier: private final
test_class.original_string: private final TaskId task1 = new TaskId(0, 1);
test_class.type: TaskId
test_class.var_name: task1
test_class.declarator: task2 = new TaskId(0, 2)
test_class.modifier: private final
test_class.original_string: private final TaskId task2 = new TaskId(0, 2);
test_class.type: TaskId
test_class.var_name: task2
test_class.declarator: task3 = new TaskId(0, 3)
test_class.modifier: private final
test_class.original_string: private final TaskId task3 = new TaskId(0, 3);
test_class.type: TaskId
test_class.var_name: task3
test_class.declarator: task4 = new TaskId(1, 1)
test_class.modifier: private final
test_class.original_string: private final TaskId task4 = new TaskId(1, 1);
test_class.type: TaskId
test_class.var_name: task4
test_class.declarator: task5 = new TaskId(1, 2)
test_class.modifier: private final
test_class.original_string: private final TaskId task5 = new TaskId(1, 2);
test_class.type: TaskId
test_class.var_name: task5
test_class.file: streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java
test_class.identifier: StreamThreadTest
test_case.body: @SuppressWarnings("unchecked")
      @Test
      public void testPartitionAssignmentChange() throws Exception {
          StreamsConfig config = new StreamsConfig(configProps());
  
          TopologyBuilder builder = new TopologyBuilder();
          builder.addSource("source1", "topic1");
          builder.addSource("source2", "topic2");
          builder.addSource("source3", "topic3");
          builder.addProcessor("processor", new MockProcessorSupplier(), "source2", "source3");
  
          StreamThread thread = new StreamThread(builder, config, new MockClientSupplier(), applicationId, clientId, processId, new Metrics(), new SystemTime()) {
              @Override
              protected StreamTask createStreamTask(TaskId id, Collection<TopicPartition> partitionsForTask) {
                  ProcessorTopology topology = builder.build("X", id.topicGroupId);
                  return new TestStreamTask(id, applicationId, partitionsForTask, topology, consumer, producer, restoreConsumer, config);
              }
          };
  
          initPartitionGrouper(config, thread);
  
          ConsumerRebalanceListener rebalanceListener = thread.rebalanceListener;
  
          assertTrue(thread.tasks().isEmpty());
  
          List<TopicPartition> revokedPartitions;
          List<TopicPartition> assignedPartitions;
          Set<TopicPartition> expectedGroup1;
          Set<TopicPartition> expectedGroup2;
  
          revokedPartitions = Collections.emptyList();
          assignedPartitions = Collections.singletonList(t1p1);
          expectedGroup1 = new HashSet<>(Arrays.asList(t1p1));
  
          rebalanceListener.onPartitionsRevoked(revokedPartitions);
          rebalanceListener.onPartitionsAssigned(assignedPartitions);
  
          assertTrue(thread.tasks().containsKey(task1));
          assertEquals(expectedGroup1, thread.tasks().get(task1).partitions());
          assertEquals(1, thread.tasks().size());
  
          revokedPartitions = assignedPartitions;
          assignedPartitions = Collections.singletonList(t1p2);
          expectedGroup2 = new HashSet<>(Arrays.asList(t1p2));
  
          rebalanceListener.onPartitionsRevoked(revokedPartitions);
          rebalanceListener.onPartitionsAssigned(assignedPartitions);
  
          assertTrue(thread.tasks().containsKey(task2));
          assertEquals(expectedGroup2, thread.tasks().get(task2).partitions());
          assertEquals(1, thread.tasks().size());
  
          revokedPartitions = assignedPartitions;
          assignedPartitions = Arrays.asList(t1p1, t1p2);
          expectedGroup1 = new HashSet<>(Collections.singleton(t1p1));
          expectedGroup2 = new HashSet<>(Collections.singleton(t1p2));
  
          rebalanceListener.onPartitionsRevoked(revokedPartitions);
          rebalanceListener.onPartitionsAssigned(assignedPartitions);
  
          assertTrue(thread.tasks().containsKey(task1));
          assertTrue(thread.tasks().containsKey(task2));
          assertEquals(expectedGroup1, thread.tasks().get(task1).partitions());
          assertEquals(expectedGroup2, thread.tasks().get(task2).partitions());
          assertEquals(2, thread.tasks().size());
  
          revokedPartitions = assignedPartitions;
          assignedPartitions = Arrays.asList(t2p1, t2p2, t3p1, t3p2);
          expectedGroup1 = new HashSet<>(Arrays.asList(t2p1, t3p1));
          expectedGroup2 = new HashSet<>(Arrays.asList(t2p2, t3p2));
  
          rebalanceListener.onPartitionsRevoked(revokedPartitions);
          rebalanceListener.onPartitionsAssigned(assignedPartitions);
  
          assertTrue(thread.tasks().containsKey(task4));
          assertTrue(thread.tasks().containsKey(task5));
          assertEquals(expectedGroup1, thread.tasks().get(task4).partitions());
          assertEquals(expectedGroup2, thread.tasks().get(task5).partitions());
          assertEquals(2, thread.tasks().size());
  
          revokedPartitions = assignedPartitions;
          assignedPartitions = Arrays.asList(t1p1, t2p1, t3p1);
          expectedGroup1 = new HashSet<>(Arrays.asList(t1p1));
          expectedGroup2 = new HashSet<>(Arrays.asList(t2p1, t3p1));
  
          rebalanceListener.onPartitionsRevoked(revokedPartitions);
          rebalanceListener.onPartitionsAssigned(assignedPartitions);
  
          assertTrue(thread.tasks().containsKey(task1));
          assertTrue(thread.tasks().containsKey(task4));
          assertEquals(expectedGroup1, thread.tasks().get(task1).partitions());
          assertEquals(expectedGroup2, thread.tasks().get(task4).partitions());
          assertEquals(2, thread.tasks().size());
  
          revokedPartitions = assignedPartitions;
          assignedPartitions = Collections.emptyList();
  
          rebalanceListener.onPartitionsRevoked(revokedPartitions);
          rebalanceListener.onPartitionsAssigned(assignedPartitions);
  
          assertTrue(thread.tasks().isEmpty());
      }
test_case.class_method_signature: StreamThreadTest.testPartitionAssignmentChange()
test_case.full_signature: @SuppressWarnings("unchecked") @Test public void testPartitionAssignmentChange()
test_case.identifier: testPartitionAssignmentChange
test_case.invocations: configProps addSource addSource addSource addProcessor build initPartitionGrouper assertTrue isEmpty tasks emptyList singletonList asList onPartitionsRevoked onPartitionsAssigned assertTrue containsKey tasks assertEquals partitions get tasks assertEquals size tasks singletonList asList onPartitionsRevoked onPartitionsAssigned assertTrue containsKey tasks assertEquals partitions get tasks assertEquals size tasks asList singleton singleton onPartitionsRevoked onPartitionsAssigned assertTrue containsKey tasks assertTrue containsKey tasks assertEquals partitions get tasks assertEquals partitions get tasks assertEquals size tasks asList asList asList onPartitionsRevoked onPartitionsAssigned assertTrue containsKey tasks assertTrue containsKey tasks assertEquals partitions get tasks assertEquals partitions get tasks assertEquals size tasks asList asList asList onPartitionsRevoked onPartitionsAssigned assertTrue containsKey tasks assertTrue containsKey tasks assertEquals partitions get tasks assertEquals partitions get tasks assertEquals size tasks emptyList onPartitionsRevoked onPartitionsAssigned assertTrue isEmpty tasks
test_case.modifiers: @SuppressWarnings("unchecked") @Test public
test_case.parameters: ()
test_case.return: void
test_case.signature: void testPartitionAssignmentChange()
focal_class.declarator: log = LoggerFactory.getLogger(StreamThread.class)
focal_class.modifier: private static final
focal_class.original_string: private static final Logger log = LoggerFactory.getLogger(StreamThread.class);
focal_class.type: Logger
focal_class.var_name: log
focal_class.declarator: STREAM_THREAD_ID_SEQUENCE = new AtomicInteger(1)
focal_class.modifier: private static final
focal_class.original_string: private static final AtomicInteger STREAM_THREAD_ID_SEQUENCE = new AtomicInteger(1);
focal_class.type: AtomicInteger
focal_class.var_name: STREAM_THREAD_ID_SEQUENCE
focal_class.declarator: partitionGrouper
focal_class.modifier: public final
focal_class.original_string: public final PartitionGrouper partitionGrouper;
focal_class.type: PartitionGrouper
focal_class.var_name: partitionGrouper
focal_class.declarator: applicationId
focal_class.modifier: public final
focal_class.original_string: public final String applicationId;
focal_class.type: String
focal_class.var_name: applicationId
focal_class.declarator: clientId
focal_class.modifier: public final
focal_class.original_string: public final String clientId;
focal_class.type: String
focal_class.var_name: clientId
focal_class.declarator: processId
focal_class.modifier: public final
focal_class.original_string: public final UUID processId;
focal_class.type: UUID
focal_class.var_name: processId
focal_class.declarator: config
focal_class.modifier: protected final
focal_class.original_string: protected final StreamsConfig config;
focal_class.type: StreamsConfig
focal_class.var_name: config
focal_class.declarator: builder
focal_class.modifier: protected final
focal_class.original_string: protected final TopologyBuilder builder;
focal_class.type: TopologyBuilder
focal_class.var_name: builder
focal_class.declarator: sourceTopics
focal_class.modifier: protected final
focal_class.original_string: protected final Set<String> sourceTopics;
focal_class.type: Set<String>
focal_class.var_name: sourceTopics
focal_class.declarator: producer
focal_class.modifier: protected final
focal_class.original_string: protected final Producer<byte[], byte[]> producer;
focal_class.type: Producer<byte[], byte[]>
focal_class.var_name: producer
focal_class.declarator: consumer
focal_class.modifier: protected final
focal_class.original_string: protected final Consumer<byte[], byte[]> consumer;
focal_class.type: Consumer<byte[], byte[]>
focal_class.var_name: consumer
focal_class.declarator: restoreConsumer
focal_class.modifier: protected final
focal_class.original_string: protected final Consumer<byte[], byte[]> restoreConsumer;
focal_class.type: Consumer<byte[], byte[]>
focal_class.var_name: restoreConsumer
focal_class.declarator: running
focal_class.modifier: private final
focal_class.original_string: private final AtomicBoolean running;
focal_class.type: AtomicBoolean
focal_class.var_name: running
focal_class.declarator: activeTasks
focal_class.modifier: private final
focal_class.original_string: private final Map<TaskId, StreamTask> activeTasks;
focal_class.type: Map<TaskId, StreamTask>
focal_class.var_name: activeTasks
focal_class.declarator: standbyTasks
focal_class.modifier: private final
focal_class.original_string: private final Map<TaskId, StandbyTask> standbyTasks;
focal_class.type: Map<TaskId, StandbyTask>
focal_class.var_name: standbyTasks
focal_class.declarator: activeTasksByPartition
focal_class.modifier: private final
focal_class.original_string: private final Map<TopicPartition, StreamTask> activeTasksByPartition;
focal_class.type: Map<TopicPartition, StreamTask>
focal_class.var_name: activeTasksByPartition
focal_class.declarator: standbyTasksByPartition
focal_class.modifier: private final
focal_class.original_string: private final Map<TopicPartition, StandbyTask> standbyTasksByPartition;
focal_class.type: Map<TopicPartition, StandbyTask>
focal_class.var_name: standbyTasksByPartition
focal_class.declarator: prevTasks
focal_class.modifier: private final
focal_class.original_string: private final Set<TaskId> prevTasks;
focal_class.type: Set<TaskId>
focal_class.var_name: prevTasks
focal_class.declarator: time
focal_class.modifier: private final
focal_class.original_string: private final Time time;
focal_class.type: Time
focal_class.var_name: time
focal_class.declarator: stateDir
focal_class.modifier: private final
focal_class.original_string: private final File stateDir;
focal_class.type: File
focal_class.var_name: stateDir
focal_class.declarator: pollTimeMs
focal_class.modifier: private final
focal_class.original_string: private final long pollTimeMs;
focal_class.type: long
focal_class.var_name: pollTimeMs
focal_class.declarator: cleanTimeMs
focal_class.modifier: private final
focal_class.original_string: private final long cleanTimeMs;
focal_class.type: long
focal_class.var_name: cleanTimeMs
focal_class.declarator: commitTimeMs
focal_class.modifier: private final
focal_class.original_string: private final long commitTimeMs;
focal_class.type: long
focal_class.var_name: commitTimeMs
focal_class.declarator: sensors
focal_class.modifier: private final
focal_class.original_string: private final StreamsMetricsImpl sensors;
focal_class.type: StreamsMetricsImpl
focal_class.var_name: sensors
focal_class.declarator: partitionAssignor = null
focal_class.modifier: private
focal_class.original_string: private StreamPartitionAssignor partitionAssignor = null;
focal_class.type: StreamPartitionAssignor
focal_class.var_name: partitionAssignor
focal_class.declarator: lastClean
focal_class.modifier: private
focal_class.original_string: private long lastClean;
focal_class.type: long
focal_class.var_name: lastClean
focal_class.declarator: lastCommit
focal_class.modifier: private
focal_class.original_string: private long lastCommit;
focal_class.type: long
focal_class.var_name: lastCommit
focal_class.declarator: rebalanceException = null
focal_class.modifier: private
focal_class.original_string: private Throwable rebalanceException = null;
focal_class.type: Throwable
focal_class.var_name: rebalanceException
focal_class.declarator: standbyRecords
focal_class.modifier: private
focal_class.original_string: private Map<TopicPartition, List<ConsumerRecord<byte[], byte[]>>> standbyRecords;
focal_class.type: Map<TopicPartition, List<ConsumerRecord<byte[], byte[]>>>
focal_class.var_name: standbyRecords
focal_class.declarator: processStandbyRecords = false
focal_class.modifier: private
focal_class.original_string: private boolean processStandbyRecords = false;
focal_class.type: boolean
focal_class.var_name: processStandbyRecords
focal_class.declarator: rebalanceListener = new ConsumerRebalanceListener() {
          @Override
          public void onPartitionsAssigned(Collection<TopicPartition> assignment) {
              try {
                  addStreamTasks(assignment);
                  addStandbyTasks();
                  lastClean = time.milliseconds(); // start the cleaning cycle
              } catch (Throwable t) {
                  rebalanceException = t;
                  throw t;
              }
          }
  
          @Override
          public void onPartitionsRevoked(Collection<TopicPartition> assignment) {
              try {
                  commitAll();
                  lastClean = Long.MAX_VALUE; // stop the cleaning cycle until partitions are assigned
              } catch (Throwable t) {
                  rebalanceException = t;
                  throw t;
              } finally {
                  // TODO: right now upon partition revocation, we always remove all the tasks;
                  // this behavior can be optimized to only remove affected tasks in the future
                  removeStreamTasks();
                  removeStandbyTasks();
              }
          }
      }
focal_class.modifier: final
focal_class.original_string: final ConsumerRebalanceListener rebalanceListener = new ConsumerRebalanceListener() {
          @Override
          public void onPartitionsAssigned(Collection<TopicPartition> assignment) {
              try {
                  addStreamTasks(assignment);
                  addStandbyTasks();
                  lastClean = time.milliseconds(); // start the cleaning cycle
              } catch (Throwable t) {
                  rebalanceException = t;
                  throw t;
              }
          }
  
          @Override
          public void onPartitionsRevoked(Collection<TopicPartition> assignment) {
              try {
                  commitAll();
                  lastClean = Long.MAX_VALUE; // stop the cleaning cycle until partitions are assigned
              } catch (Throwable t) {
                  rebalanceException = t;
                  throw t;
              } finally {
                  // TODO: right now upon partition revocation, we always remove all the tasks;
                  // this behavior can be optimized to only remove affected tasks in the future
                  removeStreamTasks();
                  removeStandbyTasks();
              }
          }
      };
focal_class.type: ConsumerRebalanceListener
focal_class.var_name: rebalanceListener
focal_class.file: streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java
focal_class.identifier: StreamThread
focal_class.class_method_signature: StreamThread.makeStateDir(String applicationId, String baseDirName)
focal_class.full_signature: static File makeStateDir(String applicationId, String baseDirName)
focal_class.identifier: makeStateDir
focal_class.modifiers: static
focal_class.parameters: (String applicationId, String baseDirName)
focal_class.return: File
focal_class.signature: File makeStateDir(String applicationId, String baseDirName)
focal_class.class_method_signature: StreamThread.StreamThread(TopologyBuilder builder,
                          StreamsConfig config,
                          KafkaClientSupplier clientSupplier,
                          String applicationId,
                          String clientId,
                          UUID processId,
                          Metrics metrics,
                          Time time)
focal_class.full_signature: public  StreamThread(TopologyBuilder builder,
                          StreamsConfig config,
                          KafkaClientSupplier clientSupplier,
                          String applicationId,
                          String clientId,
                          UUID processId,
                          Metrics metrics,
                          Time time)
focal_class.identifier: StreamThread
focal_class.modifiers: public
focal_class.parameters: (TopologyBuilder builder,
                          StreamsConfig config,
                          KafkaClientSupplier clientSupplier,
                          String applicationId,
                          String clientId,
                          UUID processId,
                          Metrics metrics,
                          Time time)
focal_class.signature:  StreamThread(TopologyBuilder builder,
                          StreamsConfig config,
                          KafkaClientSupplier clientSupplier,
                          String applicationId,
                          String clientId,
                          UUID processId,
                          Metrics metrics,
                          Time time)
focal_class.class_method_signature: StreamThread.partitionAssignor(StreamPartitionAssignor partitionAssignor)
focal_class.full_signature: public void partitionAssignor(StreamPartitionAssignor partitionAssignor)
focal_class.identifier: partitionAssignor
focal_class.modifiers: public
focal_class.parameters: (StreamPartitionAssignor partitionAssignor)
focal_class.return: void
focal_class.signature: void partitionAssignor(StreamPartitionAssignor partitionAssignor)
focal_class.class_method_signature: StreamThread.run()
focal_class.full_signature: @Override public void run()
focal_class.identifier: run
focal_class.modifiers: @Override public
focal_class.parameters: ()
focal_class.return: void
focal_class.signature: void run()
focal_class.class_method_signature: StreamThread.close()
focal_class.full_signature: public void close()
focal_class.identifier: close
focal_class.modifiers: public
focal_class.parameters: ()
focal_class.return: void
focal_class.signature: void close()
focal_class.class_method_signature: StreamThread.tasks()
focal_class.full_signature: public Map<TaskId, StreamTask> tasks()
focal_class.identifier: tasks
focal_class.modifiers: public
focal_class.parameters: ()
focal_class.return: Map<TaskId, StreamTask>
focal_class.signature: Map<TaskId, StreamTask> tasks()
focal_class.class_method_signature: StreamThread.shutdown()
focal_class.full_signature: private void shutdown()
focal_class.identifier: shutdown
focal_class.modifiers: private
focal_class.parameters: ()
focal_class.return: void
focal_class.signature: void shutdown()
focal_class.class_method_signature: StreamThread.runLoop()
focal_class.full_signature: private void runLoop()
focal_class.identifier: runLoop
focal_class.modifiers: private
focal_class.parameters: ()
focal_class.return: void
focal_class.signature: void runLoop()
focal_class.class_method_signature: StreamThread.maybeUpdateStandbyTasks()
focal_class.full_signature: private void maybeUpdateStandbyTasks()
focal_class.identifier: maybeUpdateStandbyTasks
focal_class.modifiers: private
focal_class.parameters: ()
focal_class.return: void
focal_class.signature: void maybeUpdateStandbyTasks()
focal_class.class_method_signature: StreamThread.stillRunning()
focal_class.full_signature: private boolean stillRunning()
focal_class.identifier: stillRunning
focal_class.modifiers: private
focal_class.parameters: ()
focal_class.return: boolean
focal_class.signature: boolean stillRunning()
focal_class.class_method_signature: StreamThread.maybePunctuate(StreamTask task)
focal_class.full_signature: private void maybePunctuate(StreamTask task)
focal_class.identifier: maybePunctuate
focal_class.modifiers: private
focal_class.parameters: (StreamTask task)
focal_class.return: void
focal_class.signature: void maybePunctuate(StreamTask task)
focal_class.class_method_signature: StreamThread.maybeCommit()
focal_class.full_signature: protected void maybeCommit()
focal_class.identifier: maybeCommit
focal_class.modifiers: protected
focal_class.parameters: ()
focal_class.return: void
focal_class.signature: void maybeCommit()
focal_class.class_method_signature: StreamThread.commitAll()
focal_class.full_signature: private void commitAll()
focal_class.identifier: commitAll
focal_class.modifiers: private
focal_class.parameters: ()
focal_class.return: void
focal_class.signature: void commitAll()
focal_class.class_method_signature: StreamThread.commitOne(AbstractTask task, long now)
focal_class.full_signature: private void commitOne(AbstractTask task, long now)
focal_class.identifier: commitOne
focal_class.modifiers: private
focal_class.parameters: (AbstractTask task, long now)
focal_class.return: void
focal_class.signature: void commitOne(AbstractTask task, long now)
focal_class.class_method_signature: StreamThread.maybeClean()
focal_class.full_signature: protected void maybeClean()
focal_class.identifier: maybeClean
focal_class.modifiers: protected
focal_class.parameters: ()
focal_class.return: void
focal_class.signature: void maybeClean()
focal_class.class_method_signature: StreamThread.prevTasks()
focal_class.full_signature: public Set<TaskId> prevTasks()
focal_class.identifier: prevTasks
focal_class.modifiers: public
focal_class.parameters: ()
focal_class.return: Set<TaskId>
focal_class.signature: Set<TaskId> prevTasks()
focal_class.class_method_signature: StreamThread.cachedTasks()
focal_class.full_signature: public Set<TaskId> cachedTasks()
focal_class.identifier: cachedTasks
focal_class.modifiers: public
focal_class.parameters: ()
focal_class.return: Set<TaskId>
focal_class.signature: Set<TaskId> cachedTasks()
focal_class.class_method_signature: StreamThread.createStreamTask(TaskId id, Collection<TopicPartition> partitions)
focal_class.full_signature: protected StreamTask createStreamTask(TaskId id, Collection<TopicPartition> partitions)
focal_class.identifier: createStreamTask
focal_class.modifiers: protected
focal_class.parameters: (TaskId id, Collection<TopicPartition> partitions)
focal_class.return: StreamTask
focal_class.signature: StreamTask createStreamTask(TaskId id, Collection<TopicPartition> partitions)
focal_class.class_method_signature: StreamThread.addStreamTasks(Collection<TopicPartition> assignment)
focal_class.full_signature: private void addStreamTasks(Collection<TopicPartition> assignment)
focal_class.identifier: addStreamTasks
focal_class.modifiers: private
focal_class.parameters: (Collection<TopicPartition> assignment)
focal_class.return: void
focal_class.signature: void addStreamTasks(Collection<TopicPartition> assignment)
focal_class.class_method_signature: StreamThread.removeStreamTasks()
focal_class.full_signature: private void removeStreamTasks()
focal_class.identifier: removeStreamTasks
focal_class.modifiers: private
focal_class.parameters: ()
focal_class.return: void
focal_class.signature: void removeStreamTasks()
focal_class.class_method_signature: StreamThread.closeOne(AbstractTask task)
focal_class.full_signature: private void closeOne(AbstractTask task)
focal_class.identifier: closeOne
focal_class.modifiers: private
focal_class.parameters: (AbstractTask task)
focal_class.return: void
focal_class.signature: void closeOne(AbstractTask task)
focal_class.class_method_signature: StreamThread.createStandbyTask(TaskId id, Collection<TopicPartition> partitions)
focal_class.full_signature: protected StandbyTask createStandbyTask(TaskId id, Collection<TopicPartition> partitions)
focal_class.identifier: createStandbyTask
focal_class.modifiers: protected
focal_class.parameters: (TaskId id, Collection<TopicPartition> partitions)
focal_class.return: StandbyTask
focal_class.signature: StandbyTask createStandbyTask(TaskId id, Collection<TopicPartition> partitions)
focal_class.class_method_signature: StreamThread.addStandbyTasks()
focal_class.full_signature: private void addStandbyTasks()
focal_class.identifier: addStandbyTasks
focal_class.modifiers: private
focal_class.parameters: ()
focal_class.return: void
focal_class.signature: void addStandbyTasks()
focal_class.class_method_signature: StreamThread.removeStandbyTasks()
focal_class.full_signature: private void removeStandbyTasks()
focal_class.identifier: removeStandbyTasks
focal_class.modifiers: private
focal_class.parameters: ()
focal_class.return: void
focal_class.signature: void removeStandbyTasks()
focal_class.superclass: extends Thread
focal_method.body: public Map<TaskId, StreamTask> tasks() {
          return Collections.unmodifiableMap(activeTasks);
      }
focal_method.class_method_signature: StreamThread.tasks()
focal_method.full_signature: public Map<TaskId, StreamTask> tasks()
focal_method.identifier: tasks
focal_method.invocations: unmodifiableMap
focal_method.modifiers: public
focal_method.parameters: ()
focal_method.return: Map<TaskId, StreamTask>
focal_method.signature: Map<TaskId, StreamTask> tasks()