class StreamsRebalanceListener implements ConsumerRebalanceListener {

    @Override
    public void onPartitionsAssigned(final Collection<TopicPartition> partitions) {
        // NB: all task management is already handled by:
        // org.apache.kafka.streams.processor.internals.StreamsPartitionAssignor.onAssignment
        if (assignmentErrorCode.get() == AssignorError.INCOMPLETE_SOURCE_TOPIC_METADATA.code()) {
            log.error("Received error code {}", assignmentErrorCode.get());
            throw new MissingSourceTopicException("One or more source topics were missing during rebalance");
        }

        streamThread.setState(State.PARTITIONS_ASSIGNED);
        taskManager.handleRebalanceComplete();
    }

      StreamsRebalanceListener(final Time time,
                             final TaskManager taskManager,
                             final StreamThread streamThread,
                             final Logger log,
                             final AtomicInteger assignmentErrorCode);

}