@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());
}