Producing Messages

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

KafkaProducer<String, String> producer = new KafkaProducer<>(props);

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("acks", "all");
props.put("retries", 3);
props.put("enable.idempotence", true);

{
  "bootstrap.servers": "broker1:9092,broker2:9092",
  "client.id": "my-producer",
  "acks": "all"
}

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.JsonSerializer");

KafkaProducer<String, MyObject> producer = new KafkaProducer<>(props);

const { Kafka } = require("kafkajs");

const kafka = new Kafka({
  clientId: "my-app",
  brokers: ["localhost:9092"]
});

const producer = kafka.producer();

async function sendMessage() {
  await producer.connect();
  await producer.send({
    topic: "my-topic",
    messages: [
      {
        key: "user-123",
        value: JSON.stringify({ name: "John", age: 30 })
      }
    ]
  });
  await producer.disconnect();
}

sendMessage().catch(console.error);

KafkaProducer<String, String> producer = new KafkaProducer<>(props);
ProducerRecord<String, String> record = new ProducerRecord<>("my-topic", "key1", "value1");
try {
    RecordMetadata metadata = producer.send(record).get();
    System.out.println("Message sent to partition " + metadata.partition() + " with offset " + metadata.offset());
} catch (Exception e) {
    System.err.println("Error sending message: " + e.getMessage());
}
producer.close();

KafkaProducer<String, String> producer = new KafkaProducer<>(props);
producer.send(new ProducerRecord<>("my-topic", "key1", "value1"), (metadata, exception) -> {
    if (exception != null) {
        System.err.println("Error sending message: " + exception.getMessage());
    } else {
        System.out.println("Message sent to partition " + metadata.partition() + " with offset " + metadata.offset());
    }
});
producer.flush();

KafkaProducer<String, String> producer = new KafkaProducer<>(props);
producer.send(new ProducerRecord<>("topic", "key1", "message value"));
producer.close();

ProducerRecord<String, String> record = new ProducerRecord<>(
    "my-topic",
    2, // Partition number
    "key-1",
    "message-value"
);

const producer = new KafkaProducer({
  brokers: ['localhost:9092'],
  clientId: 'my-app'
});

const message = {
  topic: 'orders',
  messages: [
    {
      value: JSON.stringify({ orderId: '123', amount: 99.99 }),
      headers: {
        'user-id': '12345',
        'event-type': 'purchase'
      }
    }
  ]
};

await producer.send(message);

{
  "acks": "all",
  "retries": 3,
  "timeout": 30000
}

const { Kafka } = require('kafkajs');

const kafka = new Kafka({
  clientId: 'my-app',
  brokers: ['localhost:9092']
});

const producer = kafka.producer();

await producer.connect();

// Send messages with flush
await producer.send({
  topic: 'test-topic',
  messages: [
    { value: 'Hello Kafka!' }
  ]
});

// Flush the producer buffer
await producer.flush();

console.log('All messages flushed');