@Override
    protected void close() {
        // FIXME Kafka needs to add a timeout parameter here for us to properly obey the timeout
        // passed in
        try {
            task.stop();
        } catch (Throwable t) {
            log.warn("Could not stop task", t);
        }
        taskStopped = true;
        Utils.closeQuietly(consumer, "consumer");
        Utils.closeQuietly(headerConverter, "header converter");
        /*
            Setting partition count explicitly to 0 to handle the case,
            when the task fails, which would cause its consumer to leave the group.
            This would cause onPartitionsRevoked to be invoked in the rebalance listener, but not onPartitionsAssigned,
            so the metrics for the task (which are still available for failed tasks until they are explicitly revoked
            from the worker) would become inaccurate.
        */
        sinkTaskMetricsGroup.recordPartitionCount(0);
    }
{@link WorkerTask} that uses a {@link SinkTask} to export data from Kafka.
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSinkTask.java