public void partitionsAssigned(Collection<TopicPartition> partitions) {
        partitions.forEach(partition -> this.partitions.computeIfAbsent(partition, PartitionHistory::new).assigned());
    }
Adds a set of partitions to the (sink) task's assignment

@param partitions the newly-assigned partitions
connect/runtime/src/test/java/org/apache/kafka/connect/integration/TaskHandle.java