Re:kafka connector[specificStartOffset cannot auto commit offset to zookeeper]

[孙森 <se...@creditease.cn>]

HI,all:
        I specify the exact offsets the consumer should start from for each partition.But the Kafka consumer connot periodically commit the offsets to Zookeeper.
I have disabled the checkpoint only if the job is stopped.This is my code:



val properties = new Properties()
properties.setProperty("bootstrap.servers", config.kafka_input.kafka_base_config.brokers)
properties.setProperty("zookeeper.connect", config.zookeeper_address)
properties.setProperty("group.id<http://group.id>", config.kafka_input.groupId)
properties.setProperty("session.timeout.ms", config.kafka_input.sessionTimeout)
properties.setProperty("enable.auto.commit", config.kafka_input.autoCommit.toString)
val flinkxConfigUtils = new WormholeFlinkxConfigUtils(config)
val topics = flinkxConfigUtils.getKafkaTopicList
val myConsumer = new FlinkKafkaConsumer010[(String, String, String, Int, Long)](topics, new WormholeDeserializationStringSchema, properties)

val specificStartOffsets = flinkxConfigUtils.getTopicPartitionOffsetMap
myConsumer.setStartFromSpecificOffsets(specificStartOffsets)


Can anyone explain the problem?
Thanks very much!