import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;
import java.util.Collection;
import java.util.regex.Pattern;

public class KafkaConsumerExample {
    
    public static void main(String[] args) {
        // Configuration du consommateur
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("group.id", "my-consumer-group");
        props.put("key.deserializer", StringDeserializer.class.getName());
        props.put("value.deserializer", StringDeserializer.class.getName());
        
        // Paramètres avancés
        props.put("enable.auto.commit", "false");  // Désactiver la validation automatique des offsets
        props.put("auto.offset.reset", "earliest");  // Lire depuis le début si pas d'offset connu
        props.put("max.poll.records", 100);  // Nombre max de messages par poll
        props.put("session.timeout.ms", 30000);  // Délai avant considération de panne
        props.put("heartbeat.interval.ms", 10000);  // Fréquence des battements de cœur
        
        // Création du consommateur
        try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props)) {
            
            // S'abonner à un ou plusieurs topics
            consumer.subscribe(Collections.singletonList("my-topic"), 
                new ConsumerRebalanceListener() {
                    @Override
                    public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
                        // Appelé avant le rebalancement, pour valider les offsets
                        System.out.println("Partitions révoquées: " + partitions);
                    }
                    
                    @Override
                    public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
                        // Appelé après l'assignation des partitions
                        System.out.println("Nouvelles partitions assignées: " + partitions);
                        // On pourrait ici restaurer des offsets personnalisés si nécessaire
                    }
                });
            
            // Boucle de consommation
            try {
                while (true) {
                    // Poll avec un timeout de 100ms
                    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                    
                    // Traitement des messages reçus
                    for (ConsumerRecord<String, String> record : records) {
                        System.out.printf("Offset = %d, Key = %s, Value = %s, Partition = %d%n",
                                record.offset(), record.key(), record.value(), record.partition());
                        
                        // Traitement du message...
                        processMessage(record);
                    }
                    
                    // Validation manuelle des offsets (commit asynchrone)
                    if (!records.isEmpty()) {
                        consumer.commitAsync((offsets, exception) -> {
                            if (exception != null) {
                                System.err.println("Erreur lors du commit: " + exception.getMessage());
                            } else {
                                System.out.println("Offsets validés: " + offsets);
                            }
                        });
                    }
                }
            } catch (Exception e) {
                System.err.println("Erreur lors de la consommation: " + e.getMessage());
            } finally {
                try {
                    // Dernier commit synchrone avant la fermeture
                    consumer.commitSync();
                } finally {
                    consumer.close();
                }
            }
        }
    }
    
    private static void processMessage(ConsumerRecord<String, String> record) {
        // Implémentation du traitement du message
        // Cette méthode peut être personnalisée selon les besoins
        try {
            // Simulation d'un traitement
            System.out.println("Traitement du message: " + record.value());
            // Traitement métier ici...
        } catch (Exception e) {
            System.err.println("Erreur lors du traitement du message: " + e.getMessage());
            // Gestion des erreurs de traitement
        }
    }
}