test_class.file: streams/src/test/java/org/apache/kafka/streams/processor/internals/assignment/TaskAssignorTest.java
test_class.identifier: TaskAssignorTest
test_case.body: @Test
      public void testAssignWithStandby() {
          HashMap<Integer, ClientState<Integer>> states = new HashMap<>();
          for (int i = 0; i < 6; i++) {
              states.put(i, new ClientState<Integer>(1d));
          }
          Set<Integer> tasks;
          Map<Integer, ClientState<Integer>> assignments;
          int numActiveTasks;
          int numAssignedTasks;
  
          // # of clients and # of tasks are equal.
          tasks = mkSet(0, 1, 2, 3, 4, 5);
  
          // 1 standby replicas.
          numActiveTasks = 0;
          numAssignedTasks = 0;
          assignments = TaskAssignor.assign(states, tasks, 1);
          for (ClientState<Integer> assignment : assignments.values()) {
              numActiveTasks += assignment.activeTasks.size();
              numAssignedTasks += assignment.assignedTasks.size();
              assertEquals(1, assignment.activeTasks.size());
              assertEquals(2, assignment.assignedTasks.size());
          }
          assertEquals(tasks.size(), numActiveTasks);
          assertEquals(tasks.size() * 2, numAssignedTasks);
  
          // # of clients < # of tasks
          tasks = mkSet(0, 1, 2, 3, 4, 5, 6, 7);
  
          // 1 standby replicas.
          assignments = TaskAssignor.assign(states, tasks, 1);
          numActiveTasks = 0;
          numAssignedTasks = 0;
          for (ClientState<Integer> assignment : assignments.values()) {
              numActiveTasks += assignment.activeTasks.size();
              numAssignedTasks += assignment.assignedTasks.size();
              assertTrue(1 <= assignment.activeTasks.size());
              assertTrue(2 >= assignment.activeTasks.size());
              assertTrue(2 <= assignment.assignedTasks.size());
              assertTrue(3 >= assignment.assignedTasks.size());
          }
          assertEquals(tasks.size(), numActiveTasks);
          assertEquals(tasks.size() * 2, numAssignedTasks);
  
          // # of clients > # of tasks
          tasks = mkSet(0, 1, 2, 3);
  
          // 1 standby replicas.
          assignments = TaskAssignor.assign(states, tasks, 1);
          numActiveTasks = 0;
          numAssignedTasks = 0;
          for (ClientState<Integer> assignment : assignments.values()) {
              numActiveTasks += assignment.activeTasks.size();
              numAssignedTasks += assignment.assignedTasks.size();
              assertTrue(0 <= assignment.activeTasks.size());
              assertTrue(1 >= assignment.activeTasks.size());
              assertTrue(1 <= assignment.assignedTasks.size());
              assertTrue(2 >= assignment.assignedTasks.size());
          }
          assertEquals(tasks.size(), numActiveTasks);
          assertEquals(tasks.size() * 2, numAssignedTasks);
  
          // # of clients >> # of tasks
          tasks = mkSet(0, 1);
  
          // 1 standby replicas.
          assignments = TaskAssignor.assign(states, tasks, 1);
          numActiveTasks = 0;
          numAssignedTasks = 0;
          for (ClientState<Integer> assignment : assignments.values()) {
              numActiveTasks += assignment.activeTasks.size();
              numAssignedTasks += assignment.assignedTasks.size();
              assertTrue(0 <= assignment.activeTasks.size());
              assertTrue(1 >= assignment.activeTasks.size());
              assertTrue(0 <= assignment.assignedTasks.size());
              assertTrue(1 >= assignment.assignedTasks.size());
          }
          assertEquals(tasks.size(), numActiveTasks);
          assertEquals(tasks.size() * 2, numAssignedTasks);
  
          // 2 standby replicas.
          assignments = TaskAssignor.assign(states, tasks, 2);
          numActiveTasks = 0;
          numAssignedTasks = 0;
          for (ClientState<Integer> assignment : assignments.values()) {
              numActiveTasks += assignment.activeTasks.size();
              numAssignedTasks += assignment.assignedTasks.size();
              assertTrue(0 <= assignment.activeTasks.size());
              assertTrue(1 >= assignment.activeTasks.size());
              assertTrue(1 == assignment.assignedTasks.size());
          }
          assertEquals(tasks.size(), numActiveTasks);
          assertEquals(tasks.size() * 3, numAssignedTasks);
  
          // 3 standby replicas.
          assignments = TaskAssignor.assign(states, tasks, 3);
          numActiveTasks = 0;
          numAssignedTasks = 0;
          for (ClientState<Integer> assignment : assignments.values()) {
              numActiveTasks += assignment.activeTasks.size();
              numAssignedTasks += assignment.assignedTasks.size();
              assertTrue(0 <= assignment.activeTasks.size());
              assertTrue(1 >= assignment.activeTasks.size());
              assertTrue(1 <= assignment.assignedTasks.size());
              assertTrue(2 >= assignment.assignedTasks.size());
          }
          assertEquals(tasks.size(), numActiveTasks);
          assertEquals(tasks.size() * 4, numAssignedTasks);
      }
test_case.class_method_signature: TaskAssignorTest.testAssignWithStandby()
test_case.full_signature: @Test public void testAssignWithStandby()
test_case.identifier: testAssignWithStandby
test_case.invocations: put mkSet assign values size size assertEquals size assertEquals size assertEquals size assertEquals size mkSet assign values size size assertTrue size assertTrue size assertTrue size assertTrue size assertEquals size assertEquals size mkSet assign values size size assertTrue size assertTrue size assertTrue size assertTrue size assertEquals size assertEquals size mkSet assign values size size assertTrue size assertTrue size assertTrue size assertTrue size assertEquals size assertEquals size assign values size size assertTrue size assertTrue size assertTrue size assertEquals size assertEquals size assign values size size assertTrue size assertTrue size assertTrue size assertTrue size assertEquals size assertEquals size
test_case.modifiers: @Test public
test_case.parameters: ()
test_case.return: void
test_case.signature: void testAssignWithStandby()
focal_class.declarator: log = LoggerFactory.getLogger(TaskAssignor.class)
focal_class.modifier: private static final
focal_class.original_string: private static final Logger log = LoggerFactory.getLogger(TaskAssignor.class);
focal_class.type: Logger
focal_class.var_name: log
focal_class.declarator: rand
focal_class.modifier: private final
focal_class.original_string: private final Random rand;
focal_class.type: Random
focal_class.var_name: rand
focal_class.declarator: states
focal_class.modifier: private final
focal_class.original_string: private final Map<C, ClientState<T>> states;
focal_class.type: Map<C, ClientState<T>>
focal_class.var_name: states
focal_class.declarator: taskPairs
focal_class.modifier: private final
focal_class.original_string: private final Set<TaskPair<T>> taskPairs;
focal_class.type: Set<TaskPair<T>>
focal_class.var_name: taskPairs
focal_class.declarator: maxNumTaskPairs
focal_class.modifier: private final
focal_class.original_string: private final int maxNumTaskPairs;
focal_class.type: int
focal_class.var_name: maxNumTaskPairs
focal_class.declarator: tasks
focal_class.modifier: private final
focal_class.original_string: private final ArrayList<T> tasks;
focal_class.type: ArrayList<T>
focal_class.var_name: tasks
focal_class.declarator: prevAssignmentBalanced = true
focal_class.modifier: private
focal_class.original_string: private boolean prevAssignmentBalanced = true;
focal_class.type: boolean
focal_class.var_name: prevAssignmentBalanced
focal_class.declarator: prevClientsUnchanged = true
focal_class.modifier: private
focal_class.original_string: private boolean prevClientsUnchanged = true;
focal_class.type: boolean
focal_class.var_name: prevClientsUnchanged
focal_class.file: streams/src/main/java/org/apache/kafka/streams/processor/internals/assignment/TaskAssignor.java
focal_class.identifier: TaskAssignor
focal_class.class_method_signature: TaskAssignor.assign(Map<C, ClientState<T>> states, Set<T> tasks, int numStandbyReplicas)
focal_class.full_signature: public static Map<C, ClientState<T>> assign(Map<C, ClientState<T>> states, Set<T> tasks, int numStandbyReplicas)
focal_class.identifier: assign
focal_class.modifiers: public static
focal_class.parameters: (Map<C, ClientState<T>> states, Set<T> tasks, int numStandbyReplicas)
focal_class.return: Map<C, ClientState<T>>
focal_class.signature: Map<C, ClientState<T>> assign(Map<C, ClientState<T>> states, Set<T> tasks, int numStandbyReplicas)
focal_class.class_method_signature: TaskAssignor.TaskAssignor(Map<C, ClientState<T>> states, Set<T> tasks, long randomSeed)
focal_class.full_signature: private  TaskAssignor(Map<C, ClientState<T>> states, Set<T> tasks, long randomSeed)
focal_class.identifier: TaskAssignor
focal_class.modifiers: private
focal_class.parameters: (Map<C, ClientState<T>> states, Set<T> tasks, long randomSeed)
focal_class.signature:  TaskAssignor(Map<C, ClientState<T>> states, Set<T> tasks, long randomSeed)
focal_class.class_method_signature: TaskAssignor.assignTasks()
focal_class.full_signature: public void assignTasks()
focal_class.identifier: assignTasks
focal_class.modifiers: public
focal_class.parameters: ()
focal_class.return: void
focal_class.signature: void assignTasks()
focal_class.class_method_signature: TaskAssignor.assignStandbyTasks(int numStandbyReplicas)
focal_class.full_signature: public void assignStandbyTasks(int numStandbyReplicas)
focal_class.identifier: assignStandbyTasks
focal_class.modifiers: public
focal_class.parameters: (int numStandbyReplicas)
focal_class.return: void
focal_class.signature: void assignStandbyTasks(int numStandbyReplicas)
focal_class.class_method_signature: TaskAssignor.assignTasks(boolean active)
focal_class.full_signature: private void assignTasks(boolean active)
focal_class.identifier: assignTasks
focal_class.modifiers: private
focal_class.parameters: (boolean active)
focal_class.return: void
focal_class.signature: void assignTasks(boolean active)
focal_class.class_method_signature: TaskAssignor.findClientFor(T task)
focal_class.full_signature: private ClientState<T> findClientFor(T task)
focal_class.identifier: findClientFor
focal_class.modifiers: private
focal_class.parameters: (T task)
focal_class.return: ClientState<T>
focal_class.signature: ClientState<T> findClientFor(T task)
focal_class.class_method_signature: TaskAssignor.findClientByAdditionCost(T task, boolean checkTaskPairs)
focal_class.full_signature: private ClientState<T> findClientByAdditionCost(T task, boolean checkTaskPairs)
focal_class.identifier: findClientByAdditionCost
focal_class.modifiers: private
focal_class.parameters: (T task, boolean checkTaskPairs)
focal_class.return: ClientState<T>
focal_class.signature: ClientState<T> findClientByAdditionCost(T task, boolean checkTaskPairs)
focal_class.class_method_signature: TaskAssignor.addTaskPairs(T task, ClientState<T> state)
focal_class.full_signature: private void addTaskPairs(T task, ClientState<T> state)
focal_class.identifier: addTaskPairs
focal_class.modifiers: private
focal_class.parameters: (T task, ClientState<T> state)
focal_class.return: void
focal_class.signature: void addTaskPairs(T task, ClientState<T> state)
focal_class.class_method_signature: TaskAssignor.hasNewTaskPair(T task, ClientState<T> state)
focal_class.full_signature: private boolean hasNewTaskPair(T task, ClientState<T> state)
focal_class.identifier: hasNewTaskPair
focal_class.modifiers: private
focal_class.parameters: (T task, ClientState<T> state)
focal_class.return: boolean
focal_class.signature: boolean hasNewTaskPair(T task, ClientState<T> state)
focal_class.class_method_signature: TaskAssignor.computeAdditionCost(T task, ClientState<T> state)
focal_class.full_signature: private double computeAdditionCost(T task, ClientState<T> state)
focal_class.identifier: computeAdditionCost
focal_class.modifiers: private
focal_class.parameters: (T task, ClientState<T> state)
focal_class.return: double
focal_class.signature: double computeAdditionCost(T task, ClientState<T> state)
focal_class.class_method_signature: TaskAssignor.pair(T task1, T task2)
focal_class.full_signature: private TaskPair<T> pair(T task1, T task2)
focal_class.identifier: pair
focal_class.modifiers: private
focal_class.parameters: (T task1, T task2)
focal_class.return: TaskPair<T>
focal_class.signature: TaskPair<T> pair(T task1, T task2)
focal_method.body: public static <C, T extends Comparable<T>> Map<C, ClientState<T>> assign(Map<C, ClientState<T>> states, Set<T> tasks, int numStandbyReplicas) {
          long seed = 0L;
          for (C client : states.keySet()) {
              seed += client.hashCode();
          }
  
          TaskAssignor<C, T> assignor = new TaskAssignor<>(states, tasks, seed);
          log.info("Assigning tasks to clients: {}, prevAssignmentBalanced: {}, " +
              "prevClientsUnchangeed: {}, tasks: {}, replicas: {}",
              states, assignor.prevAssignmentBalanced, assignor.prevClientsUnchanged,
              tasks, numStandbyReplicas);
  
          assignor.assignTasks();
          if (numStandbyReplicas > 0)
              assignor.assignStandbyTasks(numStandbyReplicas);
  
          log.info("Assigned with: " + assignor.states);
          return assignor.states;
      }
focal_method.class_method_signature: TaskAssignor.assign(Map<C, ClientState<T>> states, Set<T> tasks, int numStandbyReplicas)
focal_method.full_signature: public static Map<C, ClientState<T>> assign(Map<C, ClientState<T>> states, Set<T> tasks, int numStandbyReplicas)
focal_method.identifier: assign
focal_method.invocations: keySet hashCode info assignTasks assignStandbyTasks info
focal_method.modifiers: public static
focal_method.parameters: (Map<C, ClientState<T>> states, Set<T> tasks, int numStandbyReplicas)
focal_method.return: Map<C, ClientState<T>>
focal_method.signature: Map<C, ClientState<T>> assign(Map<C, ClientState<T>> states, Set<T> tasks, int numStandbyReplicas)