如何使用Kafka节点从Kafka主题中删除消息

Question

我创建了一个制作人和一个消费者，使用“Kafka节点”包发送和消费Kafka主题的消息。生产者和消费者通过API进行调用。POST方法用于向主题发送消息，而GET方法用于在消费者处从主题获取消息。

当我向KAFKA发送消息后调用consumer API时，之前的所有消息都会在consumer中得到安慰。on（'message'，函数（message）{}）。

我只需要最后一条消息，这是生产者发送的。

var kafka = require('kafka-node');
const Admin = require('kafka-node/lib/admin');
var Producer = kafka.Producer,
client = new kafka.KafkaClient(),
producer = new Producer(client);
producer.on('ready', function () {
    console.log('Producer is ready');
});

producer.on('error', function (err) {
    console.log('Producer is in error state');
    console.log(err);
})

const createProducer = async(req,res,next) => {
    var sentMessage = JSON.stringify(req.body.messages);
    payloads = [
        { topic: req.body.topic, messages: sentMessage, partition: 0}
    ];
    producer.createTopics(req.body.topic, function(err,data){
        console.log("createtopic error"+err);
    })
    producer.send(payloads, function( err, data) {
        console.log(payloads);
    })
    res.send({
        message: 'Successfully send data from producer',
        payloads: payloads
    })
}

module.exports = {createProducer}

var kafka = require('kafka-node');

const createConsumer = async(req,res,next) =>{
    var topic = req.params.topic;
    
    Consumer = kafka.Consumer,
    client = new kafka.KafkaClient(),
    // offset = new kafka.Offset(client),
    consumer = new Consumer(client,
        [{ topic: topic, offset: 0}],
        {
            autoCommit: false,

        }
    );
    consumer.on('message', function (message) {
        console.log('message : ',message.value);
    });

    res.send({
        message: "consumer created : ",
        topic: topic
    })
    consumer.on('error', function (err) {
        console.log('Error:',err);
    })
    
    consumer.on('offsetOutOfRange', function (err) {
        console.log("absbs");
        topic.maxNum = 2;
        offset.fetch([topic], function (err, offsets) {
            if (err) {
              return console.error(err);
            }
            var min = Math.min.apply(null, offsets[topic.topic][topic.partition]);
            consumer.setOffset(topic.topic, topic.partition, min);
        });
    })
}

module.exports = {createConsumer}

const express = require("express");
const kafkaConsumer = require("../kafka/consumer") // create routes for consumer
const kafkaproducer = require("../kafka/producer") // create routes for producer

router.post("/kafka/producer",kafkaproducer.createProducer);
router.get("/kafka/consumer/:topic",kafkaConsumer.createConsumer);

module.exports = router;

var express = require('express');
var app = express();
var kafka = require('kafka-node');
var router = require('./routes/routes');
const {
    PORT
  } = require("./config/config");

var bodyParser = require('body-parser');
app.use( bodyParser.json());
app.use(bodyParser.urlencoded({
  extended: true
}));

app.listen(PORT, function(){
    console.log('Kafka producer running at 5000');
});

app.use('/', router)

module.export = app;

如何在不使用任何数组或任何东西的情况下获取最后一条消息。

有没有办法删除这个话题？

谢谢

Answer 1

我对代码做了一些更改，现在我可以从我的主题中检索最新消息。

我已经创建了消费者内部的offset.fetchLatestOffset（[主题]， cb），并在消费者选项中进行了一些更改。

请检查这个答案

Answer 2

您必须设置消费者偏移量最新的，这是给定主题的最后一条发布消息。

consumer = new Consumer(client,
    [{ topic: topic, offset: latest}],
    {
        autoCommit: false,

    }

删除一个主题的命令如下

kafka-topics.sh --zookeeper <zk>:2181 --delete --topic <topic name>

注意：如果删除，这将不会产生影响。话题enable未设置为true。