在WorkerSinkTask.java的createConsumer()方法中:java
Map<String, Object> props = new HashMap<>(); props.put(ConsumerConfig.GROUP_ID_CONFIG, "connect-" + id.connector()); # consumer-group-id props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, Utils.join(workerConfig.getList(WorkerConfig.BOOTSTRAP_SERVERS_CONFIG), ",")); props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false"); # 自動提交offset props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); # 重置offset的位置 props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArrayDeserializer"); props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArrayDeserializer"); props.putAll(workerConfig.originalsWithPrefix("consumer."));