Spring for Apache Kafka

• Développé chez LinkedIn, et maintenu au sein de la fondation Apache depuis 2012.
• Poussé par Confluent, plateforme créée autour de Kafka.

Mais c’est quoi Apache Kafka ?

• Système de stockage de flux de messages (streams of records)
• Asynchrone
• Permet de gérer des gros volumes de données, faible latence
• Tout en assurant une sécurité des données
• Aucun type de message, que des streams

Messages

• Composés d’une valeur, d’une clé, et d’un timestamp
• Sont organisés en catégories appelées topics

Plateforme de streaming distribuée

RIP opt-kakfa

• Nouvelle stratégie SMO : ne plus maintenir de lib "maison"
• Abandon de la lib opt-kafka (utilisation de la lib Apache Kafka) au profit de Spring for Apache Kafka

Spring Apache for Kafka

   org.springframework.kafka
   spring-kafka

compile 'org.springframework.kafka:spring-kafka'

spring.kafka.bootstrap-servers=${KAFKA_BROKERS_HOST}

développement en local

version: "3"
services:

    zookeeper:
    image: wurstmeister/zookeeper:3.4.6
    container_name: zookeeper
    expose:
        - "2181"

    kafka:
    image: wurstmeister/kafka:2.11-2.0.0
    container_name: kafka
    depends_on:
        - zookeeper
    ports:
        - "9092:9092"
    environment:
        KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
        KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092
        KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
                    

Producer

• Produit des données sur les topics
• Peut attendre un ACK (accusé réception) pour savoir si la donnée a bien été transmise

Configuration

@Configuration
public class KafkaConfiguration {

    @Autowired
    private KafkaProperties kafkaProperties;

    public static final String ENTREPRISE_KAFKA_TOPIC = "isee.entreprise" ;
    public static final String ETABLISSEMENT_KAFKA_TOPIC = "isee.etablissement" ;

    @Bean
    public ProducerFactory<String, String> producerFactory() {
        return new DefaultKafkaProducerFactory<>(producerConfigs());
    }

    @Bean
    public Map<String, Object> producerConfigs() {
        Map<String, Object> kafkaProps = new HashMap<>();
        kafkaProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProperties.getBootstrapServers());
        kafkaProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        kafkaProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        kafkaProps.put(ProducerConfig.ACKS_CONFIG, "all");
        kafkaProps.put(ProducerConfig.RETRIES_CONFIG, "1");
        kafkaProps.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, "1");
        kafkaProps.put(ProducerConfig.MAX_REQUEST_SIZE_CONFIG, "22020096");
        return kafkaProps;
    }

    @Bean
    public KafkaTemplate<String, String> kafkaTemplate() {
        return new KafkaTemplate<>(producerFactory());
    }

    @Bean
    public KafkaAdmin admin() {
        Map<String, Object> configs = new HashMap<>();
        configs.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProperties.getBootstrapServers());
        return new KafkaAdmin(configs);
    }

}

Producer avec ACK

@Service
public class ProducerService {

    private static final Logger LOGGER = LoggerFactory.getLogger(ProducerService.class);

    private KafkaTemplate<String, String> kafkaTemplate;

    public ProducerService(KafkaTemplate<String, String> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    public SendResult<String, String> push(String topic, String key, Object data, ObjectMapper objectMapper) {
        try {
            String json = objectMapper.writeValueAsString(data);
            ProducerRecord<String, String> record = new ProducerRecord>(topic, key, json);
            LOGGER.debug("Envoi Kafka : topic=[{}], key=[{}], message=[{}]", record.topic(), record.key(), record.value());
            return kafkaTemplate.send(record).get();

        } catch (JsonProcessingException e) {
            throw new RuntimeException(String.format("Erreur de serialization de l'objet %s destiné au topic %s", key, topic), e);
        } catch (InterruptedException | ExecutionException e) {
            throw new RuntimeException(String.format("Erreur lors de l'envoi de l'objet %s destiné au topic %s", key, topic), e);
        }
    }
}

Producer mode asynchrone

@Service
public class ProducerService {
  private KafkaTemplate<String, String> kafkaTemplate;

    public KafkaService(KafkaTemplate<String, String> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

 // Envoi asynchrone du message
 public void pushAsync(MessageDTO messageDTO) {
    ObjectMapper objectMapper = new ObjectMapper();
    String json = null;
    try {
         json = objectMapper.writeValueAsString(messageDTO);
         String uuid = UUID.randomUUID().toString();
         ListenableFuture<SendResult<String, String>> future =
              kafkaTemplate.send(messageTopic, json);

         String finalJson = json;
         future.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
           @Override
            public void onSuccess(SendResult<String, String> result) {
                System.out.println("Sent message=[" + finalJson +
                   "] with offset=[" + result.getRecordMetadata().offset() + "]");
             }
           @Override
           public void onFailure(Throwable ex) {
              System.out.println("Unable to send message=["
                    + finalJson + "] due to : " + ex.getMessage());
           }
         });

    } catch (JsonProcessingException e) {
        e.printStackTrace();
    }
 }
}

Consumer

• Appartient à un consumer group name
• Chaque message publié dans un topic est consommé par un seul consumer d'un group name

@Component
public class KafkaConsumer {

   private final Logger log = LoggerFactory.getLogger(KafkaConsumer.class);

   @KafkaListener(id = "messageListener", topics = "${opt.kafka.topics.message}", groupId = "messageG1")
   public void messageListener(ConsumerRecord<String, String> record) throws JsonProcessingException {
      log.info("Reception enregistrement brut de messageListener : [{}]", record);
      ObjectMapper objectMapper = new ObjectMapper();
      MessageDTO messageDTO = objectMapper.readValue(record.value(), MessageDTO.class);
      log.info("Reception message de messageListener : [{}]", messageDTO);
   }
}

Consumer Stream

• Permet de développer des unités de traitement de messages au fil de l’eau (streaming).
• Permet de faire des opérations entre topic en temps réel: aggregation / jointure ...

Cas d'utilisation - Consumer-SIG

Cas d'utilisation - Consumer-SIG

Tests

   org.springframework.kafka
   spring-kafka-test
   test

testCompile 'org.springframework.kafka:spring-kafka-test'

Maintenance de Kafka

• Disponibilité
• Méthode d'upgrade
• Maintenance consumers

Prochain Upgrade 2.4.1 > 2.5.0

Problèmatique de la création de nouveaux topics

• Tel que configuré, Kafka laisse le producer créer automatiquement le topic s'il n'existe pas
• Risque de duplication de la donnée

Stratégie

• Consulter le GLIA avant toute création de topic
• En cours : mise en place par le GLIA d'un référenciel des topics dans Confluence
• Rédaction de la documentation d'interface avant toute MEP

Design for failure

• De plus en plus de clients Kafka vont etre développés
• Kakfa ne stocke que 48h de données
• Le consumer doit savoir quel est le dernier offset traité avec succès.

Projet utilisant org.springframework.kafka

• OPTISEE
• SYNCRO-TIARHE-SIRH (en cours)

Clients Kafka

• CLIK
• SIG (BPWEB)
• REFCOM
• REFTEL
• OPTISEE
• ...

Démo

• Producer
• Consumer
• KStream

Questions