@SuppressWarnings("unchecked")
@Test
public void testSubscription() throws Exception {

    StreamsConfig config = new StreamsConfig(configProps());
    TopologyBuilder builder = new TopologyBuilder();
    builder.addSource("source1", "topic1");
    builder.addSource("source2", "topic2");
    builder.addProcessor("processor", new MockProcessorSupplier(), "source1", "source2");
    final Set<TaskId> prevTasks = Utils.mkSet(
            new TaskId(0, 1), new TaskId(1, 1), new TaskId(2, 1));
    final Set<TaskId> cachedTasks = Utils.mkSet(
            new TaskId(0, 1), new TaskId(1, 1), new TaskId(2, 1),
            new TaskId(0, 2), new TaskId(1, 2), new TaskId(2, 2));
    String clientId = "client-id";
    UUID processId = UUID.randomUUID();
    StreamThread thread = new StreamThread(builder, config, new MockClientSupplier(), "test", clientId, processId, new Metrics(), new SystemTime()) {
        @Override
        public Set<TaskId> prevTasks() {
            return prevTasks;
        }
        @Override
        public Set<TaskId> cachedTasks() {
            return cachedTasks;
        }
    };
    StreamPartitionAssignor partitionAssignor = new StreamPartitionAssignor();
    partitionAssignor.configure(config.getConsumerConfigs(thread, "test", clientId));
    PartitionAssignor.Subscription subscription = partitionAssignor.subscription(Utils.mkSet("topic1", "topic2"));
    Collections.sort(subscription.topics());
    assertEquals(Utils.mkList("topic1", "topic2"), subscription.topics());
    Set<TaskId> standbyTasks = new HashSet<>(cachedTasks);
    standbyTasks.removeAll(prevTasks);
    SubscriptionInfo info = new SubscriptionInfo(processId, prevTasks, standbyTasks);
    assertEquals(info.encode(), subscription.userData());
}