micronaut kafka with auto.commit.enable=false:如何手动提交偏移量

2mbi3lxu  于 2021-06-04  发布在  Kafka
关注(0)|答案(1)|浏览(622)

我只想提交成功保存在数据库中的消息。
我知道我已关闭此应用程序的自动提交。yml

  1. micronaut:
  2.   application:
  3.     name: demoGrpcKafka
  4.   executors:
  5.     consumer:
  6.       type: fixed
  7.       nThreads: 1
  9. # kafka.bootstrap.servers: localhost:9092
  11. kafka:
  12.   bootstrap:
  13.     servers: localhost:9092
  14.   consumers:
  15.     default:
  16.       auto:
  17.         commit:
  18.           enable: false
  19.   producers:
  20.     #default:
  21.     demo-producer:
  22.       retries: 2

消费者

  1. package com.tolearn.consumer
  3. import io.micronaut.configuration.kafka.annotation.KafkaKey
  4. import io.micronaut.configuration.kafka.annotation.KafkaListener
  5. import io.micronaut.configuration.kafka.annotation.Topic
  7. @KafkaListener(groupId="myGroup")
  8. class DemoConsumer {
  9.     @Topic("testkey")
  10.     fun receive(@KafkaKey key: String?,
  11.                 msg: String,
  12.                 offset: Long,
  13.                 partition: Int,
  14.                 topic: String,
  15.                 timestamp: Long
  16.     ){
  17.         println("Key = $key " +
  18.                 "msg = $msg " +
  19.                 "offset = $offset " +
  20.                 "partition = $partition " +
  21.                 "topic = $topic " +
  22.                 "timestamp = $timestamp ")
  24.         //saved to database
  26.         // THE ISSUE IS HERE: how commit like consumer.commitOffsets(true) ?????
  27.     }
  28. }

换句话说，在使用micronaut-kafka时，如何使用commitofset或commitsync（）或任何其他方法手动提交消息？

***第二版

我回到application.yaml

  1. consumers:
  2.     default:
  3.       auto:
  4.         commit:
  5.           enable: false

***第三次编辑

我尝试添加io.micronaut.configuration.kafka.acknowledgement（已弃用）或import io.micronaut.messaging.acknowledgement，其中一个导致

  1. com.fasterxml.jackson.core.JsonParseException: Unrecognized token 'name': was expecting (JSON String, Number, Array, Object or token 'null', 'true' or 'false')

似乎我必须做些别的事情来注入这样的对象。我错过了什么？

  1. package com.tolearn.consumer
  3. import io.micronaut.configuration.kafka.Acknowledgement
  4. import io.micronaut.configuration.kafka.annotation.KafkaKey
  5. import io.micronaut.configuration.kafka.annotation.KafkaListener
  6. import io.micronaut.configuration.kafka.annotation.OffsetStrategy
  7. import io.micronaut.configuration.kafka.annotation.Topic
  8. //import io.micronaut.messaging.Acknowledgement
  9. //import io.micronaut.messaging.annotation.Header
  11. @KafkaListener(
  12.         groupId="myGroup",
  13.         offsetStrategy=OffsetStrategy.SYNC_PER_RECORD
  14. )
  15. class DemoConsumer {
  16.     @Topic("demotopic")
  17.     fun receive(@KafkaKey key: String?,
  18.                 acknowledgement: Acknowledgement,
  19.                 msg: String,
  20.                 offset: Long,
  21.                 partition: Int,
  22.                 topic: String,
  23.                 timestamp: Long
  24.                 //,header: Header
  25.     ){
  26.         println("Key = $key " +
  27.                 "msg = $msg " +
  28.                 "offset = $offset " +
  29.                 "partition = $partition " +
  30.                 "topic = $topic " +
  31.                 "timestamp = $timestamp "
  32.                 // + "header = $header"
  33.         )
  35.         //saved to database
  37.         // how commit like consumer.commitOffsets(true)
  38.         //Consumer.commitSync()
  40.         acknowledgement.ack();
  41.     }
  42. }

整个日志是

  1. 18:13:13.812 [consumer-executor-thread-1] ERROR i.m.c.k.e.KafkaListenerExceptionHandler - Kafka consumer [com.tolearn.consumer.DemoConsumer@17e970dd] failed to deserialize value: Error deserializing key/value for partition demotopic-0 at offset 4. If needed, please seek past the record to continue consumption.
  2. org.apache.kafka.common.errors.SerializationException: Error deserializing key/value for partition demotopic-0 at offset 4. If needed, please seek past the record to continue consumption.
  3. Caused by: io.micronaut.core.serialize.exceptions.SerializationException: Error deserializing object from JSON: Unrecognized token 'name': was expecting (JSON String, Number, Array, Object or token 'null', 'true' or 'false')
  4.  at [Source: (byte[])"name: "Hello"
  5. "; line: 1, column: 6]
  6.     at io.micronaut.jackson.serialize.JacksonObjectSerializer.deserialize(JacksonObjectSerializer.java:73)
  7.     at io.micronaut.configuration.kafka.serde.JsonSerde.deserialize(JsonSerde.java:82)
  8.     at org.apache.kafka.common.serialization.Deserializer.deserialize(Deserializer.java:60)
  9.     at org.apache.kafka.clients.consumer.internals.Fetcher.parseRecord(Fetcher.java:1365)
  10.     at org.apache.kafka.clients.consumer.internals.Fetcher.access$3400(Fetcher.java:130)
  11.     at org.apache.kafka.clients.consumer.internals.Fetcher$CompletedFetch.fetchRecords(Fetcher.java:1596)
  12.     at org.apache.kafka.clients.consumer.internals.Fetcher$CompletedFetch.access$1700(Fetcher.java:1432)
  13.     at org.apache.kafka.clients.consumer.internals.Fetcher.fetchRecords(Fetcher.java:684)
  14.     at org.apache.kafka.clients.consumer.internals.Fetcher.fetchedRecords(Fetcher.java:635)
  15.     at org.apache.kafka.clients.consumer.KafkaConsumer.pollForFetches(KafkaConsumer.java:1308)
  16.     at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1237)
  17.     at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1210)
  18.     at io.micronaut.configuration.kafka.processor.KafkaConsumerProcessor.lambda$process$8(KafkaConsumerProcessor.java:396)
  19.     at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)
  20.     at java.base/java.util.concurrent.FutureTask.run$$$capture(FutureTask.java:264)
  21.     at java.base/java.util.concurrent.FutureTask.run(FutureTask.java)
  22.     at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
  23.     at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
  24.     at java.base/java.lang.Thread.run(Thread.java:834)
  25. Caused by: com.fasterxml.jackson.core.JsonParseException: Unrecognized token 'name': was expecting (JSON String, Number, Array, Object or token 'null', 'true' or 'false')
  26.  at [Source: (byte[])"name: "Hello"
  27. "; line: 1, column: 6]
  28.     at com.fasterxml.jackson.core.JsonParser._constructError(JsonParser.java:1851)
  29.     at com.fasterxml.jackson.core.base.ParserMinimalBase._reportError(ParserMinimalBase.java:717)
  30.     at com.fasterxml.jackson.core.json.UTF8StreamJsonParser._reportInvalidToken(UTF8StreamJsonParser.java:3588)
  31.     at com.fasterxml.jackson.core.json.UTF8StreamJsonParser._reportInvalidToken(UTF8StreamJsonParser.java:3564)
  32.     at com.fasterxml.jackson.core.json.UTF8StreamJsonParser._matchToken2(UTF8StreamJsonParser.java:2899)
  33.     at com.fasterxml.jackson.core.json.UTF8StreamJsonParser._matchNull(UTF8StreamJsonParser.java:2870)
  34.     at com.fasterxml.jackson.core.json.UTF8StreamJsonParser._nextTokenNotInObject(UTF8StreamJsonParser.java:844)
  35.     at com.fasterxml.jackson.core.json.UTF8StreamJsonParser.nextToken(UTF8StreamJsonParser.java:757)
  36.     at com.fasterxml.jackson.databind.ObjectMapper._initForReading(ObjectMapper.java:4664)
  37.     at com.fasterxml.jackson.databind.ObjectMapper._readMapAndClose(ObjectMapper.java:4513)
  38.     at com.fasterxml.jackson.databind.ObjectMapper.readValue(ObjectMapper.java:3529)
  39.     at io.micronaut.jackson.serialize.JacksonObjectSerializer.deserialize(JacksonObjectSerializer.java:71)
  40.     ... 18 common frames omitted
  41. 18:13:13.812 [consumer-executor-thread-1] INFO  o.a.k.clients.consumer.KafkaConsumer - [Consumer clientId=demo-grpc-kafka-demo-consumer, groupId=myGroup] Seeking to offset 5 for partition demotopic-0
apache-kafkamicronautmicronaut-kafka

1条答案

按热度按时间
6qfn3psc

6qfn3psc1#

根据文档，您可以设置 offsetStrategy 在你的 KafkaListener 注解式

  1. @KafkaListener(groupId="myGroup", offsetStrategy=OffsetStrategy.SYNC)
  2. class DemoConsumer {
  3.     @Topic("testkey")
  4.     fun receive(@KafkaKey key: String?,
  5. [...]

以下选项之一：

  1. ASYNC: Asynchronously commit offsets using Consumer.commitAsync() after each batch of messages is processed.
  2. ASYNC_PER_RECORD: Asynchronously commit offsets using Consumer.commitSync() after each ConsumerRecord is consumed.
  3. AUTO: Automatically commit offsets with the Consumer.poll(long) loop.
  4. DISABLED: Do not commit offsets.
  5. SYNC: Synchronously commit offsets using Consumer.commitSync() after each batch of messages is processed.
  6. SYNC_PER_RECORD: Synchronously commit offsets using Consumer.commitSync() after each ConsumerRecord is consumed.

如果我对你的问题理解正确，你想把它设为 SYNC .

展开查看全部
赞(0）分享  回复(0）举报  2021-06-05
我来回答

相关问题

    查看更多

    热门标签

    更多
    JavaquerypythonNode开发语言requestUtil数据库Table后端算法LoggerMessageElementParser

    最新问答

    更多
    • 蜀ICP备13028337号-1 大数据知识库 https://www.saoniuhuo.com © All rights reserved
    • 本站内容来源互联网,如果侵犯您的权益请联系我们删除, 联系方式：448109455@qq.com