A tiny messages reconsumer based on kafkajs
npm install kafka-reconsume
This will reconsume messages from a specific topic, starting from the given timestamp:
import { kafkaReconsume } from "kafka-reconsume"
import { Kafka } from "kafkajs"
const kafka = new Kafka({ brokers: ["localhost:9092"] }),
(async function () {
const result = await kafkaReconsume(
kafka,
"my-topic",
1202301233, //timestamp
async (item) => {
const message = item.message.value!.toString()
console.log(item.topic, "==>", message)
}
)
for (const partition in result) {
console.log(
"partition:", partition,
"messages:", result[partition]
)
}
})()
- returns the number of reconsumed messages for each partition.
- Autocommit is always set to false, to ensure programmatic reconsume.
- By default a randomly generated consumer group id will be used, but it can be set proividing the whole consumer config as last optional argument.
Just two little wrappers around the main function:
await kafkaReconsumeByMillisecOffset(
kafka,
"my-topic",
10000, //starts from 10 seconds ago
async (item) => { /*....*/}
)
await kafkaReconsumeFromLocalDateTime(
kafka,
"my-topic",
new Date("2023-06-06 00:00:00")
async (item) => { /*....*/}
)