@Test public void shouldCloseSuspendedTasksThatAreNoLongerAssignedToThisStreamThreadBeforeCreatingNewTasks() throws Exception { final KStreamBuilder builder = new KStreamBuilder(); builder.setApplicationId(applicationId); builder.stream("t1").groupByKey().count("count-one"); builder.stream("t2").groupByKey().count("count-two"); final StreamThread thread = new StreamThread( builder, config, clientSupplier, applicationId, clientId, processId, metrics, mockTime, new StreamsMetadataState(builder, StreamsMetadataState.UNKNOWN_HOST), 0); final MockConsumer<byte[], byte[]> restoreConsumer = clientSupplier.restoreConsumer; restoreConsumer.updatePartitions("stream-thread-test-count-one-changelog", Collections.singletonList(new PartitionInfo("stream-thread-test-count-one-changelog", 0, null, new Node[0], new Node[0]))); restoreConsumer.updatePartitions("stream-thread-test-count-two-changelog", Collections.singletonList(new PartitionInfo("stream-thread-test-count-two-changelog", 0, null, new Node[0], new Node[0]))); final HashMap<TopicPartition, Long> offsets = new HashMap<>(); offsets.put(new TopicPartition("stream-thread-test-count-one-changelog", 0), 0L); offsets.put(new TopicPartition("stream-thread-test-count-two-changelog", 0), 0L); restoreConsumer.updateEndOffsets(offsets); restoreConsumer.updateBeginningOffsets(offsets); final Map<TaskId, Set<TopicPartition>> standbyTasks = new HashMap<>(); final TopicPartition t1 = new TopicPartition("t1", 0); standbyTasks.put(new TaskId(0, 0), Utils.mkSet(t1)); final Map<TaskId, Set<TopicPartition>> activeTasks = new HashMap<>(); final TopicPartition t2 = new TopicPartition("t2", 0); activeTasks.put(new TaskId(1, 0), Utils.mkSet(t2)); thread.setPartitionAssignor(new StreamPartitionAssignor() { @Override Map<TaskId, Set<TopicPartition>> standbyTasks() { return standbyTasks; } @Override Map<TaskId, Set<TopicPartition>> activeTasks() { return activeTasks; } }); thread.rebalanceListener.onPartitionsRevoked(Collections.<TopicPartition>emptyList()); thread.rebalanceListener.onPartitionsAssigned(Utils.mkSet(t2)); standbyTasks.clear(); activeTasks.clear(); standbyTasks.put(new TaskId(1, 0), Utils.mkSet(t2)); activeTasks.put(new TaskId(0, 0), Utils.mkSet(t1)); thread.rebalanceListener.onPartitionsRevoked(Collections.<TopicPartition>emptyList()); thread.rebalanceListener.onPartitionsAssigned(Utils.mkSet(t1)); }
void setPartitionAssignor(final StreamPartitionAssignor partitionAssignor) { this.partitionAssignor = partitionAssignor; }
StreamThread extends Thread { void setPartitionAssignor(final StreamPartitionAssignor partitionAssignor) { this.partitionAssignor = partitionAssignor; } }
StreamThread extends Thread { void setPartitionAssignor(final StreamPartitionAssignor partitionAssignor) { this.partitionAssignor = partitionAssignor; } StreamThread(final TopologyBuilder builder,
                        final StreamsConfig config,
                        final KafkaClientSupplier clientSupplier,
                        final String applicationId,
                        final String clientId,
                        final UUID processId,
                        final Metrics metrics,
                        final Time time,
                        final StreamsMetadataState streamsMetadataState,
                        final long cacheSizeBytes); }
StreamThread extends Thread { void setPartitionAssignor(final StreamPartitionAssignor partitionAssignor) { this.partitionAssignor = partitionAssignor; } StreamThread(final TopologyBuilder builder,
                        final StreamsConfig config,
                        final KafkaClientSupplier clientSupplier,
                        final String applicationId,
                        final String clientId,
                        final UUID processId,
                        final Metrics metrics,
                        final Time time,
                        final StreamsMetadataState streamsMetadataState,
                        final long cacheSizeBytes); @Override void run(); synchronized void close(); synchronized boolean isInitialized(); synchronized boolean stillRunning(); Map<TaskId, StreamTask> tasks(); Set<TaskId> prevActiveTasks(); Set<TaskId> cachedTasks(); void setStateListener(final StreamThread.StateListener listener); synchronized State state(); @Override String toString(); String toString(final String indent); }
StreamThread extends Thread { void setPartitionAssignor(final StreamPartitionAssignor partitionAssignor) { this.partitionAssignor = partitionAssignor; } StreamThread(final TopologyBuilder builder,
                        final StreamsConfig config,
                        final KafkaClientSupplier clientSupplier,
                        final String applicationId,
                        final String clientId,
                        final UUID processId,
                        final Metrics metrics,
                        final Time time,
                        final StreamsMetadataState streamsMetadataState,
                        final long cacheSizeBytes); @Override void run(); synchronized void close(); synchronized boolean isInitialized(); synchronized boolean stillRunning(); Map<TaskId, StreamTask> tasks(); Set<TaskId> prevActiveTasks(); Set<TaskId> cachedTasks(); void setStateListener(final StreamThread.StateListener listener); synchronized State state(); @Override String toString(); String toString(final String indent); final String applicationId; final String clientId; final UUID processId; }