from kafka import KafkaProducer
import json
from datetime import datetime

# Kafka configuration
producer = KafkaProducer(
    bootstrap_servers=['localhost:9092'],
    value_serializer=lambda v: json.dumps(v).encode('utf-8')
)

# Event data
event_data = {
  "event_id": "abc123",
  "timestamp": datetime.utcnow().isoformat() + 'Z',
  "product_id": "prod987",
  "adjusted_quantity": 10,
  "new_quantity": 150,
  "adjustment_reason": "restock",
  "adjusted_by": "user123"
}

# Send event to Kafka topic
producer.send('inventory.adjusted', event_data)
producer.flush()

import { Kafka } from 'kafkajs';

// Kafka configuration
const kafka = new Kafka({
  clientId: 'inventory-producer',
  brokers: ['localhost:9092']
});

const producer = kafka.producer();

// Event data
const eventData = {
  event_id: "abc123",
  timestamp: new Date().toISOString(),
  product_id: "prod987", 
  adjusted_quantity: 10,
  new_quantity: 150,
  adjustment_reason: "restock",
  adjusted_by: "user123"
};

// Send event to Kafka topic
async function produceEvent() {
  await producer.connect();
  await producer.send({
    topic: 'inventory.adjusted',
    messages: [
      { value: JSON.stringify(eventData) }
    ],
  });
  await producer.disconnect();
}

produceEvent().catch(console.error);

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;
import com.fasterxml.jackson.databind.ObjectMapper;
import java.util.Properties;
import java.util.HashMap;
import java.util.Map;
import java.time.Instant;

public class InventoryProducer {
    public static void main(String[] args) {
        // Kafka configuration
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ProducerConfig.CLIENT_ID_CONFIG, "inventory-producer");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

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

        try {
            // Event data
            Map<String, Object> eventData = new HashMap<>();
            eventData.put("event_id", "abc123");
            eventData.put("timestamp", Instant.now().toString());
            eventData.put("product_id", "prod987");
            eventData.put("adjusted_quantity", 10);
            eventData.put("new_quantity", 150);
            eventData.put("adjustment_reason", "restock");
            eventData.put("adjusted_by", "user123");

            // Create producer record
            ProducerRecord<String, String> record = new ProducerRecord<>(
                "inventory.adjusted",
                mapper.writeValueAsString(eventData)
            );

            // Send event to Kafka topic
            producer.send(record, (metadata, exception) -> {
                if (exception != null) {
                    System.err.println("Error producing message: " + exception);
                }
            });

        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            producer.flush();
            producer.close();
        }
    }
}