@SuppressWarnings("unchecked") @Test public void testSubscription() throws Exception { 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(), Time.SYSTEM, new StreamsMetadataState(builder, StreamsMetadataState.UNKNOWN_HOST), 0) { @Override public Set<TaskId> prevActiveTasks() { return prevTasks; } @Override public Set<TaskId> cachedTasks() { return cachedTasks; } }; 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, null); assertEquals(info.encode(), subscription.userData()); }
@Override public Subscription subscription(Set<String> topics) { final Set<TaskId> previousActiveTasks = streamThread.prevActiveTasks(); Set<TaskId> standbyTasks = streamThread.cachedTasks(); standbyTasks.removeAll(previousActiveTasks); SubscriptionInfo data = new SubscriptionInfo(streamThread.processId, previousActiveTasks, standbyTasks, this.userEndPoint); if (streamThread.builder.sourceTopicPattern() != null && !streamThread.builder.subscriptionUpdates().getUpdates().equals(topics)) { updateSubscribedTopics(topics); } return new Subscription(new ArrayList<>(topics), data.encode()); }
StreamPartitionAssignor implements PartitionAssignor, Configurable { @Override public Subscription subscription(Set<String> topics) { final Set<TaskId> previousActiveTasks = streamThread.prevActiveTasks(); Set<TaskId> standbyTasks = streamThread.cachedTasks(); standbyTasks.removeAll(previousActiveTasks); SubscriptionInfo data = new SubscriptionInfo(streamThread.processId, previousActiveTasks, standbyTasks, this.userEndPoint); if (streamThread.builder.sourceTopicPattern() != null && !streamThread.builder.subscriptionUpdates().getUpdates().equals(topics)) { updateSubscribedTopics(topics); } return new Subscription(new ArrayList<>(topics), data.encode()); } }
StreamPartitionAssignor implements PartitionAssignor, Configurable { @Override public Subscription subscription(Set<String> topics) { final Set<TaskId> previousActiveTasks = streamThread.prevActiveTasks(); Set<TaskId> standbyTasks = streamThread.cachedTasks(); standbyTasks.removeAll(previousActiveTasks); SubscriptionInfo data = new SubscriptionInfo(streamThread.processId, previousActiveTasks, standbyTasks, this.userEndPoint); if (streamThread.builder.sourceTopicPattern() != null && !streamThread.builder.subscriptionUpdates().getUpdates().equals(topics)) { updateSubscribedTopics(topics); } return new Subscription(new ArrayList<>(topics), data.encode()); }  }
StreamPartitionAssignor implements PartitionAssignor, Configurable { @Override public Subscription subscription(Set<String> topics) { final Set<TaskId> previousActiveTasks = streamThread.prevActiveTasks(); Set<TaskId> standbyTasks = streamThread.cachedTasks(); standbyTasks.removeAll(previousActiveTasks); SubscriptionInfo data = new SubscriptionInfo(streamThread.processId, previousActiveTasks, standbyTasks, this.userEndPoint); if (streamThread.builder.sourceTopicPattern() != null && !streamThread.builder.subscriptionUpdates().getUpdates().equals(topics)) { updateSubscribedTopics(topics); } return new Subscription(new ArrayList<>(topics), data.encode()); }  @Override void configure(Map<String, ?> configs); @Override String name(); @Override Subscription subscription(Set<String> topics); @Override Map<String, Assignment> assign(Cluster metadata, Map<String, Subscription> subscriptions); @Override void onAssignment(Assignment assignment); void close(); }
StreamPartitionAssignor implements PartitionAssignor, Configurable { @Override public Subscription subscription(Set<String> topics) { final Set<TaskId> previousActiveTasks = streamThread.prevActiveTasks(); Set<TaskId> standbyTasks = streamThread.cachedTasks(); standbyTasks.removeAll(previousActiveTasks); SubscriptionInfo data = new SubscriptionInfo(streamThread.processId, previousActiveTasks, standbyTasks, this.userEndPoint); if (streamThread.builder.sourceTopicPattern() != null && !streamThread.builder.subscriptionUpdates().getUpdates().equals(topics)) { updateSubscribedTopics(topics); } return new Subscription(new ArrayList<>(topics), data.encode()); }  @Override void configure(Map<String, ?> configs); @Override String name(); @Override Subscription subscription(Set<String> topics); @Override Map<String, Assignment> assign(Cluster metadata, Map<String, Subscription> subscriptions); @Override void onAssignment(Assignment assignment); void close(); final static int NOT_AVAILABLE; }