class WorkerSinkTask extends WorkerTask {

    @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(transformationChain, "transformation chain");
        Utils.closeQuietly(retryWithToleranceOperator, "retry operator");
    }

    public  WorkerSinkTask(ConnectorTaskId id,
                          SinkTask task,
                          TaskStatus.Listener statusListener,
                          TargetState initialState,
                          WorkerConfig workerConfig,
                          ClusterConfigState configState,
                          ConnectMetrics connectMetrics,
                          Converter keyConverter,
                          Converter valueConverter,
                          HeaderConverter headerConverter,
                          TransformationChain<SinkRecord> transformationChain,
                          KafkaConsumer<byte[], byte[]> consumer,
                          ClassLoader loader,
                          Time time,
                          RetryWithToleranceOperator retryWithToleranceOperator,
                          WorkerErrantRecordReporter workerErrantRecordReporter,
                          StatusBackingStore statusBackingStore);

}