Loading MQ::STREAM...
“Consumer Lag is the delta between the latest message written to a partition (Log-End-Offset) and the last message processed by the consumer (Current Offset). High lag indicates downstream bottlenecks, slow database queries, or unhandled exceptions.”
Detecting processing bottlenecks by monitoring Log-End-Offset vs Consumer-Committed-Offset.
// Batching Messages in Consumer Loop
await consumer.run({
eachBatch: async ({ batch, resolveOffset, heartbeat }) => {
const records = batch.messages.map(m => JSON.parse(m.value));
await db.analytics.insertMany(records); // Bulk insert
resolveOffset(batch.lastOffset());
await heartbeat();
}
});Batching downstream database inserts (e.g. 500 events per bulk write) drops consumer CPU overhead by 90% and slashes lag.