kafka-console

const { exec } = require('child_process'); async function runKafkaCommand(command) { return new Promise((resolve, reject) => { // KAFKA_BROKERS environment variable can be used, or default to localhost const brokers = process.env.KAFKA_BROKERS || 'localhost:9092'; exec(`npx kafka-console ${command} -b ${brokers}`, (error, stdout, stderr) => { if (error) { console.error(`Command failed: npx kafka-console ${command} -b ${brokers}\nStderr: ${stderr}`); return reject(error); } if (stderr) { console.warn(`Command stderr for ${command}:\n${stderr}`); } console.log(`Stdout for ${command}:\n${stdout}`); resolve(stdout); }); }); } (async () => { const topic = 'my-console-topic'; // Ensure this topic exists on your Kafka cluster console.log('--- Listing topics ---'); await runKafkaCommand('list'); console.log('\n--- Producing a single message ---'); const message = JSON.stringify({ key: "user-1", value: { name: "Alice", timestamp: Date.now() } }); // Using echo with the heredoc-like syntax (<<<) for stdin await runKafkaCommand(`produce ${topic} <<< '${message}'`); console.log('\n--- Consuming 1 message from the beginning ---'); // Note: --from 0 might re-read messages if group not used, or if no messages since start await runKafkaCommand(`consume ${topic} --from 0 --count 1 --group cli-quickstart`); console.log('\nQuickstart complete. Verify Kafka brokers are running and topic exists for full functionality.'); })().catch(console.error);