class StreamsRebalanceListener implements ConsumerRebalanceListener {

    @Override
    public void onPartitionsRevoked(final Collection<TopicPartition> partitions) {
        log.debug("Current state {}: revoked partitions {} because of consumer rebalance.\n" +
                      "\tcurrently assigned active tasks: {}\n" +
                      "\tcurrently assigned standby tasks: {}\n",
                  streamThread.state(),
                  partitions,
                  taskManager.activeTaskIds(),
                  taskManager.standbyTaskIds());

        if (streamThread.setState(State.PARTITIONS_REVOKED) != null && !partitions.isEmpty()) {
            final long start = time.milliseconds();
            try {
                taskManager.handleRevocation(partitions);
            } finally {
                log.info("partition revocation took {} ms.", time.milliseconds() - start);
            }
        }
    }

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

}