class Worker {

    static Map<String, Object> consumerConfigs(ConnectorTaskId id,
                                               WorkerConfig config,
                                               ConnectorConfig connConfig,
                                               Class<? extends Connector> connectorClass,
                                               ConnectorClientConfigOverridePolicy connectorClientConfigOverridePolicy,
                                               String clusterId) {
        // Include any unknown worker configs so consumer configs can be set globally on the worker
        // and through to the task
        Map<String, Object> consumerProps = new HashMap<>();

        consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, SinkUtils.consumerGroupId(id.connector()));
        consumerProps.put(ConsumerConfig.CLIENT_ID_CONFIG, "connector-consumer-" + id);
        consumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
                  Utils.join(config.getList(WorkerConfig.BOOTSTRAP_SERVERS_CONFIG), ","));
        consumerProps.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
        consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArrayDeserializer");
        consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArrayDeserializer");

        consumerProps.putAll(config.originalsWithPrefix("consumer."));
        //add client metrics.context properties
        ConnectUtils.addMetricsContextProperties(consumerProps, config, clusterId);
        // Connector-specified overrides
        Map<String, Object> consumerOverrides =
            connectorClientConfigOverrides(id, connConfig, connectorClass, ConnectorConfig.CONNECTOR_CLIENT_CONSUMER_OVERRIDES_PREFIX,
                                           ConnectorType.SINK, ConnectorClientConfigRequest.ClientType.CONSUMER,
                                           connectorClientConfigOverridePolicy);
        consumerProps.putAll(consumerOverrides);

        return consumerProps;
    }

    public  Worker(
        String workerId,
        Time time,
        Plugins plugins,
        WorkerConfig config,
        OffsetBackingStore offsetBackingStore,
        ConnectorClientConfigOverridePolicy connectorClientConfigOverridePolicy);
    @SuppressWarnings("deprecation")  Worker(
            String workerId,
            Time time,
            Plugins plugins,
            WorkerConfig config,
            OffsetBackingStore offsetBackingStore,
            ExecutorService executorService,
            ConnectorClientConfigOverridePolicy connectorClientConfigOverridePolicy
    );

}