@Override
    public void onPartitionsLost(final Collection<TopicPartition> partitions) {
        log.info("at state {}: partitions {} lost due to missed rebalance.\n" +
                     "\tlost active tasks: {}\n" +
                     "\tlost assigned standby tasks: {}\n",
                 streamThread.state(),
                 partitions,
                 taskManager.activeTaskIds(),
                 taskManager.standbyTaskIds());

        final long start = time.milliseconds();
        try {
            // close all active tasks as lost but don't try to commit offsets as we no longer own them
            taskManager.handleLostAll();
        } finally {
            log.info("partitions lost took {} ms.", time.milliseconds() - start);
        }
    }
@Test
    public void shouldHandleLostPartitions() {
        streamsRebalanceListener.onPartitionsLost(Collections.singletonList(new TopicPartition("topic", 0)));

        verify(taskManager).handleLostAll();
    }