test_class.declarator: t1p0 = new TopicPartition("topic1", 0)
test_class.modifier: private
test_class.original_string: private TopicPartition t1p0 = new TopicPartition("topic1", 0);
test_class.type: TopicPartition
test_class.var_name: t1p0
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: t2p0 = new TopicPartition("topic2", 0)
test_class.modifier: private
test_class.original_string: private TopicPartition t2p0 = new TopicPartition("topic2", 0);
test_class.type: TopicPartition
test_class.var_name: t2p0
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: t3p0 = new TopicPartition("topic3", 0)
test_class.modifier: private
test_class.original_string: private TopicPartition t3p0 = new TopicPartition("topic3", 0);
test_class.type: TopicPartition
test_class.var_name: t3p0
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: t3p3 = new TopicPartition("topic3", 3)
test_class.modifier: private
test_class.original_string: private TopicPartition t3p3 = new TopicPartition("topic3", 3);
test_class.type: TopicPartition
test_class.var_name: t3p3
test_class.declarator: allTopics = Utils.mkSet("topic1", "topic2")
test_class.modifier: private
test_class.original_string: private Set<String> allTopics = Utils.mkSet("topic1", "topic2");
test_class.type: Set<String>
test_class.var_name: allTopics
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]),
              new PartitionInfo("topic3", 3, 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]),
              new PartitionInfo("topic3", 3, 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: task0 = new TaskId(0, 0)
test_class.modifier: private final
test_class.original_string: private final TaskId task0 = new TaskId(0, 0);
test_class.type: TaskId
test_class.var_name: task0
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.file: streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamPartitionAssignorTest.java
test_class.identifier: StreamPartitionAssignorTest
test_case.body: @Test
      public void testOnAssignment() throws Exception {
          StreamsConfig config = new StreamsConfig(configProps());
  
          TopicPartition t2p3 = new TopicPartition("topic2", 3);
  
          TopologyBuilder builder = new TopologyBuilder();
          builder.addSource("source1", "topic1");
          builder.addSource("source2", "topic2");
          builder.addProcessor("processor", new MockProcessorSupplier(), "source1", "source2");
  
          UUID uuid = UUID.randomUUID();
          String client1 = "client1";
  
          StreamThread thread = new StreamThread(builder, config, new MockClientSupplier(), "test", client1, uuid, new Metrics(), new SystemTime());
  
          StreamPartitionAssignor partitionAssignor = new StreamPartitionAssignor();
          partitionAssignor.configure(config.getConsumerConfigs(thread, "test", client1));
  
          List<TaskId> activeTaskList = Utils.mkList(task0, task3);
          Map<TaskId, Set<TopicPartition>> standbyTasks = new HashMap<>();
          standbyTasks.put(task1, Utils.mkSet(new TopicPartition("t1", 0)));
          standbyTasks.put(task2, Utils.mkSet(new TopicPartition("t2", 0)));
  
          AssignmentInfo info = new AssignmentInfo(activeTaskList, standbyTasks);
          PartitionAssignor.Assignment assignment = new PartitionAssignor.Assignment(Utils.mkList(t1p0, t2p3), info.encode());
          partitionAssignor.onAssignment(assignment);
  
          assertEquals(Utils.mkSet(task0), partitionAssignor.tasksForPartition(t1p0));
          assertEquals(Utils.mkSet(task3), partitionAssignor.tasksForPartition(t2p3));
          assertEquals(standbyTasks, partitionAssignor.standbyTasks());
      }
test_case.class_method_signature: StreamPartitionAssignorTest.testOnAssignment()
test_case.full_signature: @Test public void testOnAssignment()
test_case.identifier: testOnAssignment
test_case.invocations: configProps addSource addSource addProcessor randomUUID configure getConsumerConfigs mkList put mkSet put mkSet mkList encode onAssignment assertEquals mkSet tasksForPartition assertEquals mkSet tasksForPartition assertEquals standbyTasks
test_case.modifiers: @Test public
test_case.parameters: ()
test_case.return: void
test_case.signature: void testOnAssignment()
focal_class.declarator: log = LoggerFactory.getLogger(StreamPartitionAssignor.class)
focal_class.modifier: private static final
focal_class.original_string: private static final Logger log = LoggerFactory.getLogger(StreamPartitionAssignor.class);
focal_class.type: Logger
focal_class.var_name: log
focal_class.declarator: PARTITION_COMPARATOR = new Comparator<TopicPartition>() {
          @Override
          public int compare(TopicPartition p1, TopicPartition p2) {
              int result = p1.topic().compareTo(p2.topic());
  
              if (result != 0) {
                  return result;
              } else {
                  return p1.partition() < p2.partition() ? -1 : (p1.partition() > p2.partition() ? 1 : 0);
              }
          }
      }
focal_class.modifier: private static final
focal_class.original_string: private static final Comparator<TopicPartition> PARTITION_COMPARATOR = new Comparator<TopicPartition>() {
          @Override
          public int compare(TopicPartition p1, TopicPartition p2) {
              int result = p1.topic().compareTo(p2.topic());
  
              if (result != 0) {
                  return result;
              } else {
                  return p1.partition() < p2.partition() ? -1 : (p1.partition() > p2.partition() ? 1 : 0);
              }
          }
      };
focal_class.type: Comparator<TopicPartition>
focal_class.var_name: PARTITION_COMPARATOR
focal_class.declarator: streamThread
focal_class.modifier: private
focal_class.original_string: private StreamThread streamThread;
focal_class.type: StreamThread
focal_class.var_name: streamThread
focal_class.declarator: numStandbyReplicas
focal_class.modifier: private
focal_class.original_string: private int numStandbyReplicas;
focal_class.type: int
focal_class.var_name: numStandbyReplicas
focal_class.declarator: topicGroups
focal_class.modifier: private
focal_class.original_string: private Map<Integer, TopologyBuilder.TopicsInfo> topicGroups;
focal_class.type: Map<Integer, TopologyBuilder.TopicsInfo>
focal_class.var_name: topicGroups
focal_class.declarator: partitionToTaskIds
focal_class.modifier: private
focal_class.original_string: private Map<TopicPartition, Set<TaskId>> partitionToTaskIds;
focal_class.type: Map<TopicPartition, Set<TaskId>>
focal_class.var_name: partitionToTaskIds
focal_class.declarator: stateChangelogTopicToTaskIds
focal_class.modifier: private
focal_class.original_string: private Map<String, Set<TaskId>> stateChangelogTopicToTaskIds;
focal_class.type: Map<String, Set<TaskId>>
focal_class.var_name: stateChangelogTopicToTaskIds
focal_class.declarator: internalSourceTopicToTaskIds
focal_class.modifier: private
focal_class.original_string: private Map<String, Set<TaskId>> internalSourceTopicToTaskIds;
focal_class.type: Map<String, Set<TaskId>>
focal_class.var_name: internalSourceTopicToTaskIds
focal_class.declarator: standbyTasks
focal_class.modifier: private
focal_class.original_string: private Map<TaskId, Set<TopicPartition>> standbyTasks;
focal_class.type: Map<TaskId, Set<TopicPartition>>
focal_class.var_name: standbyTasks
focal_class.declarator: internalTopicManager
focal_class.modifier: private
focal_class.original_string: private InternalTopicManager internalTopicManager;
focal_class.type: InternalTopicManager
focal_class.var_name: internalTopicManager
focal_class.file: streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamPartitionAssignor.java
focal_class.identifier: StreamPartitionAssignor
focal_class.interfaces: implements PartitionAssignor, Configurable
focal_class.class_method_signature: StreamPartitionAssignor.configure(Map<String, ?> configs)
focal_class.full_signature: @Override public void configure(Map<String, ?> configs)
focal_class.identifier: configure
focal_class.modifiers: @Override public
focal_class.parameters: (Map<String, ?> configs)
focal_class.return: void
focal_class.signature: void configure(Map<String, ?> configs)
focal_class.class_method_signature: StreamPartitionAssignor.name()
focal_class.full_signature: @Override public String name()
focal_class.identifier: name
focal_class.modifiers: @Override public
focal_class.parameters: ()
focal_class.return: String
focal_class.signature: String name()
focal_class.class_method_signature: StreamPartitionAssignor.subscription(Set<String> topics)
focal_class.full_signature: @Override public Subscription subscription(Set<String> topics)
focal_class.identifier: subscription
focal_class.modifiers: @Override public
focal_class.parameters: (Set<String> topics)
focal_class.return: Subscription
focal_class.signature: Subscription subscription(Set<String> topics)
focal_class.class_method_signature: StreamPartitionAssignor.prepareTopic(Map<String, Set<TaskId>> topicToTaskIds,
                                                              boolean compactTopic,
                                                              boolean postPartitionPhase)
focal_class.full_signature: private Map<TopicPartition, PartitionInfo> prepareTopic(Map<String, Set<TaskId>> topicToTaskIds,
                                                              boolean compactTopic,
                                                              boolean postPartitionPhase)
focal_class.identifier: prepareTopic
focal_class.modifiers: private
focal_class.parameters: (Map<String, Set<TaskId>> topicToTaskIds,
                                                              boolean compactTopic,
                                                              boolean postPartitionPhase)
focal_class.return: Map<TopicPartition, PartitionInfo>
focal_class.signature: Map<TopicPartition, PartitionInfo> prepareTopic(Map<String, Set<TaskId>> topicToTaskIds,
                                                              boolean compactTopic,
                                                              boolean postPartitionPhase)
focal_class.class_method_signature: StreamPartitionAssignor.assign(Cluster metadata, Map<String, Subscription> subscriptions)
focal_class.full_signature: @Override public Map<String, Assignment> assign(Cluster metadata, Map<String, Subscription> subscriptions)
focal_class.identifier: assign
focal_class.modifiers: @Override public
focal_class.parameters: (Cluster metadata, Map<String, Subscription> subscriptions)
focal_class.return: Map<String, Assignment>
focal_class.signature: Map<String, Assignment> assign(Cluster metadata, Map<String, Subscription> subscriptions)
focal_class.class_method_signature: StreamPartitionAssignor.onAssignment(Assignment assignment)
focal_class.full_signature: @Override public void onAssignment(Assignment assignment)
focal_class.identifier: onAssignment
focal_class.modifiers: @Override public
focal_class.parameters: (Assignment assignment)
focal_class.return: void
focal_class.signature: void onAssignment(Assignment assignment)
focal_class.class_method_signature: StreamPartitionAssignor.ensureCopartitioning(Collection<Set<String>> copartitionGroups, Map<Integer, Set<String>> internalTopicGroups, Cluster metadata)
focal_class.full_signature: private void ensureCopartitioning(Collection<Set<String>> copartitionGroups, Map<Integer, Set<String>> internalTopicGroups, Cluster metadata)
focal_class.identifier: ensureCopartitioning
focal_class.modifiers: private
focal_class.parameters: (Collection<Set<String>> copartitionGroups, Map<Integer, Set<String>> internalTopicGroups, Cluster metadata)
focal_class.return: void
focal_class.signature: void ensureCopartitioning(Collection<Set<String>> copartitionGroups, Map<Integer, Set<String>> internalTopicGroups, Cluster metadata)
focal_class.class_method_signature: StreamPartitionAssignor.ensureCopartitioning(Set<String> copartitionGroup, Set<String> internalTopics, Cluster metadata)
focal_class.full_signature: private void ensureCopartitioning(Set<String> copartitionGroup, Set<String> internalTopics, Cluster metadata)
focal_class.identifier: ensureCopartitioning
focal_class.modifiers: private
focal_class.parameters: (Set<String> copartitionGroup, Set<String> internalTopics, Cluster metadata)
focal_class.return: void
focal_class.signature: void ensureCopartitioning(Set<String> copartitionGroup, Set<String> internalTopics, Cluster metadata)
focal_class.class_method_signature: StreamPartitionAssignor.tasksForState(String stateName)
focal_class.full_signature: public Set<TaskId> tasksForState(String stateName)
focal_class.identifier: tasksForState
focal_class.modifiers: public
focal_class.parameters: (String stateName)
focal_class.return: Set<TaskId>
focal_class.signature: Set<TaskId> tasksForState(String stateName)
focal_class.class_method_signature: StreamPartitionAssignor.tasksForPartition(TopicPartition partition)
focal_class.full_signature: public Set<TaskId> tasksForPartition(TopicPartition partition)
focal_class.identifier: tasksForPartition
focal_class.modifiers: public
focal_class.parameters: (TopicPartition partition)
focal_class.return: Set<TaskId>
focal_class.signature: Set<TaskId> tasksForPartition(TopicPartition partition)
focal_class.class_method_signature: StreamPartitionAssignor.standbyTasks()
focal_class.full_signature: public Map<TaskId, Set<TopicPartition>> standbyTasks()
focal_class.identifier: standbyTasks
focal_class.modifiers: public
focal_class.parameters: ()
focal_class.return: Map<TaskId, Set<TopicPartition>>
focal_class.signature: Map<TaskId, Set<TopicPartition>> standbyTasks()
focal_class.class_method_signature: StreamPartitionAssignor.setInternalTopicManager(InternalTopicManager internalTopicManager)
focal_class.full_signature: public void setInternalTopicManager(InternalTopicManager internalTopicManager)
focal_class.identifier: setInternalTopicManager
focal_class.modifiers: public
focal_class.parameters: (InternalTopicManager internalTopicManager)
focal_class.return: void
focal_class.signature: void setInternalTopicManager(InternalTopicManager internalTopicManager)
focal_method.body: @Override
      public void onAssignment(Assignment assignment) {
          List<TopicPartition> partitions = new ArrayList<>(assignment.partitions());
  
          Collections.sort(partitions, PARTITION_COMPARATOR);
  
          AssignmentInfo info = AssignmentInfo.decode(assignment.userData());
          this.standbyTasks = info.standbyTasks;
  
          Map<TopicPartition, Set<TaskId>> partitionToTaskIds = new HashMap<>();
          Iterator<TaskId> iter = info.activeTasks.iterator();
          for (TopicPartition partition : partitions) {
              Set<TaskId> taskIds = partitionToTaskIds.get(partition);
              if (taskIds == null) {
                  taskIds = new HashSet<>();
                  partitionToTaskIds.put(partition, taskIds);
              }
  
              if (iter.hasNext()) {
                  taskIds.add(iter.next());
              } else {
                  TaskAssignmentException ex = new TaskAssignmentException(
                          "failed to find a task id for the partition=" + partition.toString() +
                          ", partitions=" + partitions.size() + ", assignmentInfo=" + info.toString()
                  );
                  log.error(ex.getMessage(), ex);
                  throw ex;
              }
          }
          this.partitionToTaskIds = partitionToTaskIds;
      }
focal_method.class_method_signature: StreamPartitionAssignor.onAssignment(Assignment assignment)
focal_method.full_signature: @Override public void onAssignment(Assignment assignment)
focal_method.identifier: onAssignment
focal_method.invocations: partitions sort decode userData iterator get put hasNext add next toString size toString error getMessage
focal_method.modifiers: @Override public
focal_method.parameters: (Assignment assignment)
focal_method.return: void
focal_method.signature: void onAssignment(Assignment assignment)