我在单节点上运行kafka,我想看到kafka生产者的行为当我关闭我的kafka代理,然后我在几秒钟内重新启动我的代理,所以我创建了spring boot projet,在那里我可以发送1000个customer json对象并打印每次发送的json对象的偏移量。我的应用程序运行得很好,但是当我关闭kafka代理并在几秒钟后重新启动代理时,我的生产者返回正常地发送来自最新偏移量的对象。问题是,在我的示例中,当控制台中的offset=983时,我关闭了我的kafka,当我再次启动我的代理时,我看到kafka开始从offset=984发送消息,但我发现有3或4条消息由于此消息错误而丢失!!!
983
offset acked
2019-04-01 16:20:34.635 INFO 7656 --- [ntainer#0-0-C-1] o.a.k.c.c.internals.AbstractCoordinator : [Consumer clientId=consumer-2, groupId=jsa-group] Group coordinator dali-X556UJ:9092 (id: 2147483647 rack: null) is unavailable or invalid, will attempt rediscovery
2019-04-01 16:20:34.638 INFO 7656 --- [ntainer#0-0-C-1] o.a.kafka.clients.FetchSessionHandler : [Consumer clientId=consumer-2, groupId=jsa-group] Error sending fetch request (sessionId=1910675333, epoch=31) to node 0: org.apache.kafka.common.errors.DisconnectException.
2019-04-01 16:20:34.689 WARN 7656 --- [ntainer#0-0-C-1] org.apache.kafka.clients.NetworkClient : [Consumer clientId=consumer-2, groupId=jsa-group] Connection to node 0 could not be established. Broker may not be available.
2019-04-01 16:20:34.844 WARN 7656 --- [ntainer#0-0-C-1] org.apache.kafka.clients.NetworkClient : [Consumer clientId=consumer-2, groupId=jsa-group] Connection to node 0 could not be established. Broker may not be available.
2019-04-01 16:20:35.050 WARN 7656 --- [ntainer#0-0-C-1] org.apache.kafka.clients.NetworkClient : [Consumer clientId=consumer-2, groupId=jsa-group] Connection to node 0 could not be established. Broker may not be available.
2019-04-01 16:20:35.557 WARN 7656 --- [ntainer#0-0-C-1] org.apache.kafka.clients.NetworkClient : [Consumer clientId=consumer-2, groupId=jsa-group] Connection to node 0 could not be established. Broker may not be available.
.........
dali 36
2019-04-01 16:20:55.592 INFO 7656 --- [ntainer#0-0-C-1] o.a.k.c.c.internals.AbstractCoordinator : [Consumer clientId=consumer-2, groupId=jsa-group] Discovered group coordinator dali-X556UJ:9092 (id: 2147483647 rack: null)
2019-04-01 16:20:55.594 ERROR 7656 --- [ntainer#0-0-C-1] o.a.k.c.c.internals.ConsumerCoordinator : [Consumer clientId=consumer-2, groupId=jsa-group] Offset commit failed on partition jsa-kafka-topic-0 at offset 984: This is not the correct coordinator.
2019-04-01 16:20:55.594 INFO 7656 --- [ntainer#0-0-C-1] o.a.k.c.c.internals.AbstractCoordinator : [Consumer clientId=consumer-2, groupId=jsa-group] Group coordinator dali-X556UJ:9092 (id: 2147483647 rack: null) is unavailable or invalid, will attempt rediscovery
2019-04-01 16:20:55.696 INFO 7656 --- [ntainer#0-0-C-1] o.a.k.c.c.internals.AbstractCoordinator : [Consumer clientId=consumer-2, groupId=jsa-group] Discovered group coordinator dali-X556UJ:9092 (id: 2147483647 rack: null)
2019-04-01 16:20:55.696 INFO 7656 --- [ntainer#0-0-C-1] o.a.k.c.c.internals.AbstractCoordinator : [Consumer clientId=consumer-2, groupId=jsa-group] Group coordinator dali-X556UJ:9092 (id: 2147483647 rack: null) is unavailable or invalid, will attempt rediscovery
984
offset acked
985
offset acked
986
offset acked
987
offset acked
988
offset acked
989
offset acked
990
offset acked
991
offset acked
992
offset acked
993
offset acked
994
offset acked
995
offset acked
996
offset acked
997
offset acked
998
offset acked
999
offset acked
1000
offset acked
1001
offset acked
1002
offset acked
1003
offset acked
1004
offset acked
2019-04-01 16:20:55.799 INFO 7656 --- [ntainer#0-0-C-1] o.a.k.c.c.internals.AbstractCoordinator : [Consumer clientId=consumer-2, groupId=jsa-group] Discovered group coordinator dali-X556UJ:9092 (id: 2147483647 rack: null)
2019-04-01 16:20:55.803 ERROR 7656 --- [ntainer#0-0-C-1] o.a.k.c.c.internals.ConsumerCoordinator : [Consumer clientId=consumer-2, groupId=jsa-group] Offset commit failed on partition jsa-kafka-topic-0 at offset 984: This is not the correct coordinator.
2019-04-01 16:20:55.803 INFO 7656 --- [ntainer#0-0-C-1] o.a.k.c.c.internals.AbstractCoordinator : [Consumer clientId=consumer-2, groupId=jsa-group] Group coordinator dali-X556UJ:9092 (id: 2147483647 rack: null) is unavailable or invalid, will attempt rediscovery
2019-04-01 16:20:55.904 INFO 7656 --- [ntainer#0-0-C-1] o.a.k.c.c.internals.AbstractCoordinator : [Consumer clientId=consumer-2, groupId=jsa-group] Discovered group coordinator dali-X556UJ:9092 (id: 2147483647 rack: null)
2019-04-01 16:20:55.904 INFO 7656 --- [ntainer#0-0-C-1] o.a.k.c.c.internals.AbstractCoordinator : [Consumer clientId=consumer-2, groupId=jsa-group] Group coordinator dali-X556UJ:9092 (id: 2147483647 rack: null) is unavailable or invalid, will attempt rediscovery
2019-04-01 16:20:56.006 INFO 7656 --- [ntainer#0-0-C-1] o.a.k.c.c.internals.AbstractCoordinator : [Consumer clientId=consumer-2, groupId=jsa-group] Discovered group coordinator dali-X556UJ:9092 (id: 2147483647 rack: null)
2019-04-01 16:20:56.010 ERROR 7656 --- [ntainer#0-0-C-1] o.a.k.c.c.internals.ConsumerCoordinator : [Consumer clientId=consumer-2, groupId=jsa-group] Offset commit failed on partition jsa-kafka-topic-0 at offset 984: This is not the correct coordinator.
2019-04-01 16:20:56.011 INFO 7656 --- [ntainer#0-0-C-1] o.a.k.c.c.internals.AbstractCoordinator : [Consumer clientId=consumer-2, groupId=jsa-group] Group coordinator dali-X556UJ:9092 (id: 2147483647 rack: null) is unavailable or invalid, will attempt rediscovery
2019-04-01 16:20:56.111 INFO 7656 --- [ntainer#0-0-C-1] o.a.k.c.c.internals.AbstractCoordinator : [Consumer clientId=consumer-2, groupId=jsa-group] Discovered group coordinator dali-X556UJ:9092 (id: 2147483647 rack: null)
2019-04-01 16:20:56.111 INFO 7656 --- [ntainer#0-0-C-1] o.a.k.c.c.internals.AbstractCoordinator : [Consumer clientId=consumer-2, groupId=jsa-group] Group coordinator dali-X556UJ:9092 (id: 2147483647 rack: null) is unavailable or invalid, will attempt rediscovery
2019-04-01 16:20:56.213 INFO 7656 --- [ntainer#0-0-C-1] o.a.k.c.c.internals.AbstractCoordinator : [Consumer clientId=consumer-2, groupId=jsa-group] Discovered group coordinator dali-X556UJ:9092 (id: 2147483647 rack: null)
2019-04-01 16:20:56.215 ERROR 7656 --- [ntainer#0-0-C-1] o.a.k.c.c.internals.ConsumerCoordinator : [Consumer clientId=consumer-2, groupId=jsa-group] Offset commit failed on partition jsa-kafka-topic-0 at offset 984: This is not the correct coordinator.
2019-04-01 16:20:56.215 INFO 7656 --- [ntainer#0-0-C-1] o.a.k.c.c.internals.AbstractCoordinator : [Consumer clientId=consumer-2, groupId=jsa-group] Group coordinator dali-X556UJ:9092 (id: 2147483647 rack: null) is unavailable or invalid, will attempt rediscovery
2019-04-01 16:20:56.317 INFO 7656 --- [ntainer#0-0-C-1] o.a.k.c.c.internals.AbstractCoordinator : [Consumer clientId=consumer-2, groupId=jsa-group] Discovered group coordinator dali-X556UJ:9092 (id: 2147483647 rack: null)
2019-04-01 16:20:56.317 INFO 7656 --- [ntainer#0-0-C-1] o.a.k.c.c.internals.AbstractCoordinator : [Consumer clientId=consumer-2, groupId=jsa-group] Group coordinator dali-X556UJ:9092 (id: 2147483647 rack: null) is unavailable or invalid, will attempt rediscovery
2019-04-01 16:20:56.419 INFO 7656 --- [ntainer#0-0-C-1] o.a.k.c.c.internals.AbstractCoordinator : [Consumer clientId=consumer-2, groupId=jsa-group] Discovered group coordinator dali-X556UJ:9092 (id: 2147483647 rack: null)
2019-04-01 16:20:56.422 ERROR 7656 --- [ntainer#0-0-C-1] o.a.k.c.c.internals.ConsumerCoordinator : [Consumer clientId=consumer-2, groupId=jsa-group] Offset commit failed on partition jsa-kafka-topic-0 at offset 984: This is not the correct coordinator.
2019-04-01 16:20:56.422 INFO 7656 --- [ntainer#0-0-C-1] o.a.k.c.c.internals.AbstractCoordinator : [Consumer clientId=consumer-2, groupId=jsa-group] Group coordinator dali-X556UJ:9092 (id: 2147483647 rack: null) is unavailable or invalid, will attempt rediscovery
2019-04-01 16:20:56.523 INFO 7656 --- [ntainer#0-0-C-1] o.a.k.c.c.internals.AbstractCoordinator : [Consumer clientId=consumer-2, groupId=jsa-group] Discovered group coordinator dali-X556UJ:9092 (id: 2147483647 rack: null)
2019-04-01 16:20:56.523 INFO 7656 --- [ntainer#0-0-C-1] o.a.k.c.c.internals.AbstractCoordinator : [Consumer clientId=consumer-2, groupId=jsa-group] Group coordinator dali-X556UJ:9092 (id: 2147483647 rack: null) is unavailable or invalid, will attempt rediscovery
dali 37
1005
offset acked
2019-04-01 16:20:56.625 INFO 7656 --- [ntainer#0-0-C-1] o.a.k.c.c.internals.AbstractCoordinator : [Consumer clientId=consumer-2, groupId=jsa-group] Discovered group coordinator dali-X556UJ:9092 (id: 2147483647 rack: null)
2019-04-01 16:20:56.628 ERROR 7656 --- [ntainer#0-0-C-1] o.a.k.c.c.internals.ConsumerCoordinator : [Consumer clientId=consumer-2, groupId=jsa-group] Offset commit failed on partition jsa-kafka-topic-0 at offset 984: This is not the correct coordinator.
2019-04-01 16:20:56.628 INFO 7656 --- [ntainer#0-0-C-1] o.a.k.c.c.internals.AbstractCoordinator : [Consumer clientId=consumer-2, groupId=jsa-group] Group coordinator dali-X556UJ:9092 (id: 2147483647 rack: null) is unavailable or invalid, will attempt rediscovery
2019-04-01 16:20:56.729 INFO 7656 --- [ntainer#0-0-C-1] o.a.k.c.c.internals.AbstractCoordinator : [Consumer clientId=consumer-2, groupId=jsa-group] Discovered group coordinator dali-X556UJ:9092 (id: 2147483647 rack: null)
2019-04-01 16:20:56.729 INFO 7656 --- [ntainer#0-0-C-1] o.a.k.c.c.internals.AbstractCoordinator : [Consumer clientId=consumer-2, groupId=jsa-group] Group coordinator dali-X556UJ:9092 (id: 2147483647 rack: null) is unavailable or invalid, will attempt rediscovery
2019-04-01 16:20:56.831 INFO 7656 --- [ntainer#0-0-C-1] o.a.k.c.c.internals.AbstractCoordinator : [Consumer clientId=consumer-2, groupId=jsa-group] Discovered group coordinator dali-X556UJ:9092 (id: 2147483647 rack: null)
2019-04-01 16:20:56.834 ERROR 7656 --- [ntainer#0-0-C-1] o.a.k.c.c.internals.ConsumerCoordinator : [Consumer clientId=consumer-2, groupId=jsa-group] Offset commit failed on partition jsa-kafka-topic-0 at offset 984: This is not the correct coordinator.
2019-04-01 16:20:56.834 INFO 7656 --- [ntainer#0-0-C-1] o.a.k.c.c.internals.AbstractCoordinator : [Consumer clientId=consumer-2, groupId=jsa-group] Group coordinator dali-X556UJ:9092 (id: 2147483647 rack: null) is unavailable or invalid, will attempt rediscovery
2019-04-01 16:20:56.936 INFO 7656 --- [ntainer#0-0-C-1] o.a.k.c.c.internals.AbstractCoordinator : [Consumer clientId=consumer-2, groupId=jsa-group] Discovered group coordinator dali-X556UJ:9092 (id: 2147483647 rack: null)
2019-04-01 16:20:56.936 INFO 7656 --- [ntainer#0-0-C-1] o.a.k.c.c.internals.AbstractCoordinator : [Consumer clientId=consumer-2, groupId=jsa-group] Group coordinator dali-X556UJ:9092 (id: 2147483647 rack: null) is unavailable or invalid, will attempt rediscovery
2019-04-01 16:20:57.037 INFO 7656 --- [ntainer#0-0-C-1] o.a.k.c.c.internals.AbstractCoordinator : [Consumer clientId=consumer-2, groupId=jsa-group] Discovered group coordinator dali-X556UJ:9092 (id: 2147483647 rack: null)
2019-04-01 16:20:57.040 ERROR 7656 --- [ntainer#0-0-C-1] o.a.k.c.c.internals.ConsumerCoordinator : [Consumer clientId=consumer-2, groupId=jsa-group] Offset commit failed on partition jsa-kafka-topic-0 at offset 984: This is not the correct coordinator.
2019-04-01 16:20:57.040 INFO 7656 --- [ntainer#0-0-C-1] o.a.k.c.c.internals.AbstractCoordinator : [Consumer clientId=consumer-2, groupId=jsa-group] Group coordinator dali-X556UJ:9092 (id: 2147483647 rack: null) is unavailable or invalid, will attempt rediscovery
2019-04-01 16:20:57.140 INFO 7656 --- [ntainer#0-0-C-1] o.a.k.c.c.internals.AbstractCoordinator : [Consumer clientId=consumer-2, groupId=jsa-group] Discovered group coordinator dali-X556UJ:9092 (id: 2147483647 rack: null)
2019-04-01 16:20:57.141 INFO 7656 --- [ntainer#0-0-C-1] o.a.k.c.c.internals.AbstractCoordinator : [Consumer clientId=consumer-2, groupId=jsa-group] Group coordinator dali-X556UJ:9092 (id: 2147483647 rack: null) is unavailable or invalid, will attempt rediscovery
2019-04-01 16:20:57.245 INFO 7656 --- [ntainer#0-0-C-1] o.a.k.c.c.internals.AbstractCoordinator : [Consumer clientId=consumer-2, groupId=jsa-group] Discovered group coordinator dali-X556UJ:9092 (id: 2147483647 rack: null)
2019-04-01 16:20:57.252 ERROR 7656 --- [ntainer#0-0-C-1] o.a.k.c.c.internals.ConsumerCoordinator : [Consumer clientId=consumer-2, groupId=jsa-group] Offset commit failed on partition jsa-kafka-topic-0 at offset 984: This is not the correct coordinator.
2019-04-01 16:20:57.252 INFO 7656 --- [ntainer#0-0-C-1] o.a.k.c.c.internals.AbstractCoordinator : [Consumer clientId=consumer-2, groupId=jsa-group] Group coordinator dali-X556UJ:9092 (id: 2147483647 rack: null) is unavailable or invalid, will attempt rediscovery
2019-04-01 16:20:57.358 INFO 7656 --- [ntainer#0-0-C-1] o.a.k.c.c.internals.AbstractCoordinator : [Consumer clientId=consumer-2, groupId=jsa-group] Discovered group coordinator dali-X556UJ:9092 (id: 2147483647 rack: null)
2019-04-01 16:20:57.358 INFO 7656 --- [ntainer#0-0-C-1] o.a.k.c.c.internals.AbstractCoordinator : [Consumer clientId=consumer-2, groupId=jsa-group] Group coordinator dali-X556UJ:9092 (id: 2147483647 rack: null) is unavailable or invalid, will attempt rediscovery
2019-04-01 16:20:57.460 INFO 7656 --- [ntainer#0-0-C-1] o.a.k.c.c.internals.AbstractCoordinator : [Consumer clientId=consumer-2, groupId=jsa-group] Discovered group coordinator dali-X556UJ:9092 (id: 2147483647 rack: null)
2019-04-01 16:20:57.466 ERROR 7656 --- [ntainer#0-0-C-1] o.a.k.c.c.internals.ConsumerCoordinator : [Consumer clientId=consumer-2, groupId=jsa-group] Offset commit failed on partition jsa-kafka-topic-0 at offset 984: This is not the correct coordinator.
2019-04-01 16:20:57.467 INFO 7656 --- [ntainer#0-0-C-1] o.a.k.c.c.internals.AbstractCoordinator : [Consumer clientId=consumer-2, groupId=jsa-group] Group coordinator dali-X556UJ:9092 (id: 2147483647 rack: null) is unavailable or invalid, will attempt rediscovery
2019-04-01 16:20:57.567 INFO 7656 --- [ntainer#0-0-C-1] o.a.k.c.c.internals.AbstractCoordinator : [Consumer clientId=consumer-2, groupId=jsa-group] Discovered group coordinator dali-X556UJ:9092 (id: 2147483647 rack: null)
2019-04-01 16:20:57.568 INFO 7656 --- [ntainer#0-0-C-1] o.a.k.c.c.internals.AbstractCoordinator : [Consumer clientId=consumer-2, groupId=jsa-group] Group coordinator dali-X556UJ:9092 (id: 2147483647 rack: null) is unavailable or invalid, will attempt rediscovery
dali 38
1006
offset acked
2019-04-01 16:20:57.671 INFO 7656 --- [ntainer#0-0-C-1] o.a.k.c.c.internals.AbstractCoordinator : [Consumer clientId=consumer-2, groupId=jsa-group] Discovered group coordinator dali-X556UJ:9092 (id: 2147483647 rack: null)
2019-04-01 16:20:57.681 ERROR 7656 --- [ntainer#0-0-C-1] o.a.k.c.c.internals.ConsumerCoordinator : [Consumer clientId=consumer-2, groupId=jsa-group] Offset commit failed on partition jsa-kafka-topic-0 at offset 984: This is not the correct coordinator.
2019-04-01 16:20:57.681 INFO 7656 --- [ntainer#0-0-C-1] o.a.k.c.c.internals.AbstractCoordinator : [Consumer clientId=consumer-2, groupId=jsa-group] Group coordinator dali-X556UJ:9092 (id: 2147483647 rack: null) is unavailable or invalid, will attempt rediscovery
2019-04-01 16:20:57.785 INFO 7656 --- [ntainer#0-0-C-1] o.a.k.c.c.internals.AbstractCoordinator : [Consumer clientId=consumer-2, groupId=jsa-group] Discovered group coordinator dali-X556UJ:9092 (id: 2147483647 rack: null)
2019-04-01 16:20:57.786 INFO 7656 --- [ntainer#0-0-C-1] o.a.k.c.c.internals.AbstractCoordinator : [Consumer clientId=consumer-2, groupId=jsa-group] Group coordinator dali-X556UJ:9092 (id: 2147483647 rack: null) is unavailable or invalid, will attempt rediscovery
2019-04-01 16:20:57.889 INFO 7656 --- [ntainer#0-0-C-1] o.a.k.c.c.internals.AbstractCoordinator : [Consumer clientId=consumer-2, groupId=jsa-group] Discovered group coordinator dali-X556UJ:9092 (id: 2147483647 rack: null)
2019-04-01 16:20:57.896 ERROR 7656 --- [ntainer#0-0-C-1] o.a.k.c.c.internals.ConsumerCoordinator : [Consumer clientId=consumer-2, groupId=jsa-group] Offset commit failed on partition jsa-kafka-topic-0 at offset 984: The coordinator is loading and hence can't process requests.
2019-04-01 16:20:58.007 ERROR 7656 --- [ntainer#0-0-C-1] o.a.k.c.c.internals.ConsumerCoordinator : [Consumer clientId=consumer-2, groupId=jsa-group] Offset commit failed on partition jsa-kafka-topic-0 at offset 984: The coordinator is loading and hence can't process requests.
2019-04-01 16:20:58.125 INFO 7656 --- [ntainer#0-0-C-1] o.a.k.c.c.internals.ConsumerCoordinator : [Consumer clientId=consumer-2, groupId=jsa-group] Revoking previously assigned partitions [jsa-kafka-topic-0]
2019-04-01 16:20:58.125 INFO 7656 --- [ntainer#0-0-C-1] o.s.k.l.KafkaMessageListenerContainer : partitions revoked: [jsa-kafka-topic-0]
2019-04-01 16:20:58.125 INFO 7656 --- [ntainer#0-0-C-1] o.a.k.c.c.internals.AbstractCoordinator : [Consumer clientId=consumer-2, groupId=jsa-group] (Re-)joining group
2019-04-01 16:20:58.149 INFO 7656 --- [ntainer#0-0-C-1] o.a.k.c.c.internals.AbstractCoordinator : [Consumer clientId=consumer-2, groupId=jsa-group] Successfully joined group with generation 31
2019-04-01 16:20:58.149 INFO 7656 --- [ntainer#0-0-C-1] o.a.k.c.c.internals.ConsumerCoordinator : [Consumer clientId=consumer-2, groupId=jsa-group] Setting newly assigned partitions [jsa-kafka-topic-0]
2019-04-01 16:20:58.154 INFO 7656 --- [ntainer#0-0-C-1] o.s.k.l.KafkaMessageListenerContainer : partitions assigned: [jsa-kafka-topic-0]
dali 39
1007
offset acked
dali 40
1008
offset acked
dali 41
1009
offset acked
dali 42
1010
offset acked
dali 43
1011
offset acked
//...continue to send normally
为什么当我再次启动kafka代理时,offset=1004、offset=1005和d offset=1006的消息以及offset=983的第一条消息(当我关闭代理时)无法发送。
//我的主课
public static void main(String[] args) {
SpringApplication.run(SpringKafkaSendConsumeJavaObjectApplication.class, args);
}
@Bean
ApplicationRunner run(CustomerRepository personRepository) {
return args -> {
List<Customer> list = new ArrayList<>(100);
for (int i = 0; i < 100; i++) {
list.add(new Customer("dali "+i, 25, "homme"));
Runnable runnable = new MyRunnable(list,producer);
runnable.run();
};
}
这是我的run方法,我每秒钟发送一个客户
public class MyRunnable implements Runnable {
private static final Logger LOGGER = LoggerFactory.getLogger(MyRunnable.class);
private List<Customer> customers;
private KafkaProducer kafkaProducer;
public MyRunnable(List<Customer> customers,KafkaProducer kafkaProducer) {
this.customers = customers.stream().collect(Collectors.toList());
this.kafkaProducer = kafkaProducer;
}
@Override
public void run() {
customers.forEach(customer -> {
System.out.println(customer.getName());
kafkaProducer.send(customer);
try {
Thread.sleep(1000);//1second
} catch (InterruptedException e) {
e.printStackTrace();
}
});
}
//这是我的发送方法
@Service
public class KafkaProducer {
private static final Logger LOGGER=LoggerFactory.getLogger(KafkaProducer.class);
@Autowired
private KafkaTemplate<String, Customer> kafkaTemplate;
@Value("${jsa.kafka.topic}")
String kafkaTopic = "jsa-test";
@Async
public void send(Customer customer) {
//LOGGER.info("sending data= '{}' " , customer);
ListenableFuture<SendResult<String,Customer>> listenableFuture = kafkaTemplate.send(kafkaTopic ,customer);
listenableFuture.addCallback(new ListenableFutureCallback<SendResult<String,Customer>>() {
@Override
public void onFailure(Throwable throwable) {
System.out.println(throwable.getMessage());
}
@Override
public void onSuccess(final SendResult<String, Customer> message) {
System.out.println(message.getRecordMetadata().offset());
System.out.println("offset acked");
}
});
}
//这是application.properties中的kafka配置
# Kafka Cluster
jsa.kafka.bootstrap-servers=localhost:9092
# consumer group id
jsa.kafka.consumer.group-id=jsa-group
# topic name
jsa.kafka.topic=jsa-kafka-topic
# server port
server.port=9000
# reconnect.backoff.ms=10000
# Integer.MAX_VALUE
retries=2147483647
retry.backoff.ms=1000
# 5 minutes
request.timeout.ms=305000
# Integer.MAX_VALUE
max.block.ms=2147483647
1条答案
按热度按时间ktca8awb1#
你要么等着
Future<?>
由KafkaTemplate.send()
方法,或调用kafkaTemplate.flush()
在停止应用程序之前。kafkaproducer.flush()javadocs文件