KafkaIgniteStreamerSelfTest.java 文件源码

/**
 * Creates default consumer config.
 *
 * @param zooKeeper ZooKeeper address <server:port>.
 * @param grpId Group Id for kafka subscriber.
 * @return Kafka consumer configuration.
 */
private ConsumerConfig createDefaultConsumerConfig(String zooKeeper, String grpId) {
    A.notNull(zooKeeper, "zookeeper");
    A.notNull(grpId, "groupId");

    Properties props = new Properties();

    props.put("zookeeper.connect", zooKeeper);
    props.put("group.id", grpId);
    props.put("zookeeper.session.timeout.ms", "400");
    props.put("zookeeper.sync.time.ms", "200");
    props.put("auto.commit.interval.ms", "1000");
    props.put("auto.offset.reset", "smallest");

    return new ConsumerConfig(props);
}