Kafka Console CLI Tool

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);