import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

public class KafkaProducerExample {
    
    public static void main(String[] args) {
        // Configuration du producteur
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("key.serializer", StringSerializer.class.getName());
        props.put("value.serializer", StringSerializer.class.getName());
        
        // Paramètres avancés
        props.put("acks", "all");  // Attendre la confirmation de tous les réplicas
        props.put("retries", 3);   // Nombre de tentatives en cas d'échec
        props.put("batch.size", 16384);  // Taille du lot en octets
        props.put("linger.ms", 1);  // Temps d'attente avant envoi du lot
        props.put("buffer.memory", 33554432);  // Taille du buffer

        try (Producer<String, String> producer = new KafkaProducer<>(props)) {
            // Envoi d'un message simple
            ProducerRecord<String, String> record1 = 
                new ProducerRecord<>("my-topic", "key1", "Hello, Kafka!");
            
            // Envoi avec callback pour gérer le résultat de manière asynchrone
            producer.send(record1, (metadata, exception) -> {
                if (exception != null) {
                    System.err.println("Erreur lors de l'envoi: " + exception.getMessage());
                } else {
                    System.out.printf("Message envoyé avec succès au topic %s, partition %d, offset %d%n",
                            metadata.topic(), metadata.partition(), metadata.offset());
                }
            });
            
            // Envoi synchrone (attente de la confirmation)
            ProducerRecord<String, String> record2 = 
                new ProducerRecord<>("my-topic", "key2", "Message synchrone");
            try {
                RecordMetadata metadata = producer.send(record2).get();
                System.out.println("Message synchrone envoyé avec succès à l'offset: " + metadata.offset());
            } catch (InterruptedException | ExecutionException e) {
                System.err.println("Erreur lors de l'envoi synchrone: " + e.getMessage());
            }
            
            // Envoi de plusieurs messages dans une boucle
            for (int i = 0; i < 10; i++) {
                String key = "key-" + i;
                String value = "Message #" + i;
                producer.send(new ProducerRecord<>("my-topic", key, value));
            }
            
            // S'assurer que tous les messages sont bien envoyés avant de fermer
            producer.flush();
            System.out.println("Tous les messages ont été envoyés avec succès!");
            
        } catch (Exception e) {
            System.err.println("Erreur avec le producteur Kafka: " + e.getMessage());
        }
    }
}