正在过期的xxxxx的1条记录：自批处理创建后已过30030毫秒加上停留时间

Question

下面是我用来推送主题的方法：

public DomainEntity push(DomainEntity pDomainEntity) throws Exception {
    logger.log(Level.INFO, "streaming...");
    wKafkaProperties.put("bootstrap.servers", "localhost:9092");
    wKafkaProperties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    wKafkaProperties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    KafkaProducer wKafkaProducer = new KafkaProducer(wKafkaProperties);
    ProducerRecord wProducerRecord = new ProducerRecord("DomainEntityCommandStream", getJSON(pDomainEntity));
    wKafkaProducer.send(wProducerRecord, (RecordMetadata r, Exception e) -> {
        if (e != null) {
            logger.log(Level.SEVERE, e.getMessage());
        }
    }).get();
    return pDomainEntity;
}

使用命令shell脚本

./kafka-console-producer.sh--broker-list 10.0.1.15:9092--主题DomainEntityCommandStream

./kafka-console-consumer.sh--boostrap-server 10.0.1.15:9092-topic DomainEntityCommandStream--从头开始

工作得很好。

通过Stackoverflow的一些相关问题，我试图清除这个主题：

有效负载小得离谱，为什么我要更改batch.size？

<soapenv:Envelope xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/"
                  xmlns:gs="http://soap.problem.com">
   <soapenv:Header/>
   <soapenv:Body>
      <gs:streamDomainEntityRequest>
         <gs:domainEntity>
                <gs:name>12345</gs:name>
                <gs:value>Quebec</gs:value>
                <gs:version>666</gs:version>
            </gs:domainEntity>
      </gs:streamDomainEntityRequest>
   </soapenv:Body>
</soapenv:Envelope>

Answer 1

使用Docker和Kafka 0.11.0.1图像，您需要向容器中添加以下环境参数：

KAFKA_ZOOKEEPER_CONNECT=x.x.x.x:xxxx（您的zookeeper IP或域：端口默认2181）

KAFKA_ADVERTISED_HOST_NAME=x.x.x.x（您的kafka IP或域）

可选：

KAFKA_BROKER_ID=999（某个值）

kafka_create_topics=test:1:1（开始时要创建一些主题名称）