kafka节点-如何检索压缩主题上的所有消息

t9eec4r0  于 2021-06-06  发布在  Kafka
关注(0)|答案(2)|浏览(436)

我正在尝试使用kafka节点读取来自kafka主题的压缩消息。
问题是,最近插入的消息留在eol之上,并且在插入其他消息之前无法访问。实际上,下线和高位偏移之间存在一个间隙,这会阻止读取最新消息。不清楚为什么会这样。
已使用创建主题

kafka-topics.sh --zookeeper ${KAFKA_HOST}:2181 --create --topic atopic --config "cleanup.policy=compact" --config "delete.retention.ms=100" --config "segment.ms=100" --config "min.cleanable.dirty.ratio=0" --partitions 1 --replication-factor 1

主题中产生了许多键值。有些钥匙是一样的。

var client = new kafka.KafkaClient({kafkaHost: "<host:port>",autoConnect: true})
var producer = new HighLevelProducer(client);
  producer.send(payload, function(error, result) {
  debug('Sent payload to Kafka: ', payload);
  if (error) {
    console.error(error);
  } else {
   res(true)
  }
  client.close()
 });
});

以下是插入的键和值

key - 1
key2 - 1
key3 - 1
key - 2
key2 - 2
key3 - 2
key1 - 3
key - 3
key2 - 3
key3 - 3

然后请求主题键集。

var options = {
        id: 'consumer1',
        kafkaHost: "<host:port>",
        groupId: "consumergroup1",
        sessionTimeout: 15000,
        protocol: ['roundrobin'],
        fromOffset: 'earliest'
      };
      var consumerGroup = new ConsumerGroup(options, topic);
        consumerGroup.on('error', onError);
        consumerGroup.on('message', onMessage);
        consumerGroup.on('done', function(message) {
          consumerGroup.close(true,function(){ });
        })
        function onError (error) {
          console.error(error);
        }
        function onMessage (message) {)
            console.log('%s read msg Topic="%s" Partition=%s Offset=%d HW=%d', this.client.clientId, message.topic, message.partition, message.offset, message.highWaterOffset, message.value);
        }
      })

结果令人惊讶:

consumer1 read msg Topic="atopic" Partition=0 Offset=4 highWaterOffset=10 Key=key2 value={"name":"key2","url":"2"}
consumer1 read msg Topic="atopic" Partition=0 Offset=5 highWaterOffset=10 Key=key3 value={"name":"key3","url":"2"}
consumer1 read msg Topic="atopic" Partition=0 Offset=6 highWaterOffset=10 Key=key1 value={"name":"key1","url":"3"}
consumer1 read msg Topic="atopic" Partition=0 Offset=7 highWaterOffset=10 Key=key value={"name":"key","url":"3"}
consumer1 read msg Topic="atopic" Partition=0 Offset=0 highWaterOffset=10 Key= value=
consumer1 read msg Topic="atopic" Partition=0 Offset=0 highWaterOffset=10 Key= value=
consumer1 read msg Topic="atopic" Partition=0 Offset=0 highWaterOffset=10 Key= value=
consumer1 read msg Topic="atopic" Partition=0 Offset=0 highWaterOffset=10 Key= value=

有一个代表最新值10的高水位偏移量。然而,消费者看到的偏移值只有7。不知何故,压缩阻止消费者看到最新的消息。
目前还不清楚如何避免这种限制,让消费者看到最新的消息。
谢谢你的建议。谢谢。

x33g5p2x

x33g5p2x1#

不知何故,压缩阻止消费者看到最新的消息。
是的,你错过了一些信息,但你也看到其他人。
压缩是删除先前的关键点。
注意这里没有 url - 1 价值观

Key=key2 value={"name":"key2","url":"2"}
Key=key3 value={"name":"key3","url":"2"}
Key=key1 value={"name":"key1","url":"3"}
Key=key value={"name":"key","url":"3"}

这是因为您为同一个键发送了新值。
你发了10条信息,所以这个主题的高潮偏移量是10
你的代码看起来不一定是错的,但是你应该有两个以上的3值。打印的偏移量与此逻辑相对应。

key - 1 | 0
key2 - 1 | 1
key3 - 1 | 2
key - 2 | 3
key2 - 2 | 4
key3 - 2 | 5
key1 - 3 | 6
key  - 3 | 7
key2 - 3 | 8
key3 - 3 | 9

一般来说,我建议不要让kafka尝试压缩主题,以每秒10倍的速度编写日志段,也不要使用不同的库,例如 node-rdkafka

oalqel3c

oalqel3c2#

在对kafka做了更多的工作之后,kafka节点api似乎有以下行为(我认为这实际上源于kafka本身)。
当在highwateroff之前查询消息时,只有高达highwateroffset的消息才会返回到consumergroup。如果消息没有被复制,这是有意义的,因为组中的另一个使用者不一定会看到这些消息。
仍然可以使用使用者(而不是使用者组)并通过查询特定分区来请求和接收超出highwateroffset的消息。
另外,当偏移量不一定在latestoffset时,“done”事件似乎会被触发。在这种情况下,有必要在message.offset+1处提交进一步的查询。如果你继续这样做,你可以得到所有的消息到最新的偏移量。
我不清楚Kafka为什么会有这种行为,但有一个可能是一些较低层次的细节,表面这种紧急行为。

相关问题