tutorial_kafka - Archi-Lab/fae-team-3 GitHub Wiki

Tutorial Kafka

Die Kafka docker-compose.yml Datei sieht ungefähr so aus.

Um auf Kafka zuzugreifen im internen Docker Netzwerk "kafka_backend" muss der Port 9092 genutzt werden. Andernfalls der Port 5000.

Kafka muss zuerst gestartet werden vor den anderen Microservices.

#Kafka Service
version: '3'
    
services:
  zookeeper:
    image: confluentinc/cp-zookeeper:latest
    ports:
      - 2181:2181
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000
    networks:
        - backend
  kafka:
    image: confluentinc/cp-kafka:latest
    ports:
      - 5000:5000
    depends_on:
      - zookeeper
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: INTERNAL://kafka:9092,EXTERNAL://localhost:5000
      KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
    networks:
        - backend
networks:
    backend:
        driver: bridge

Die application.properties Datei sollte ungefähr so aussehen:

eventing.brokers = kafka_kafka_1:9092,localhost:5000
topics.ungeRou = ungeRou.t
topics.ungeVer = ungeVer.t
topics.notlage = Notlage.t
topics.bpmeta = BPMeta.t
topics.bpcore = BPCore.t

group.ungeRou = NotlageungeRouConsumer
group.ungeVer = NotlageungeVerConsumer
group.bpReader = BPBPConsumer
group.bpMetaReader = FABPMetaConsumer
group.notlageReader = BPNotlageConsumer

Der Eintrag: eventing.brokers = kafka_kafka_1:9092

Es ist wichtig zu wissen, wie der echte Hostnaem von Kafka im kafka_backend Netzwerk sit. Dazu kann der Befehl: docker networks inspect kafka_backend genutzt werden.

Dieser Name muss dann als Bootstrap Server angegeben werden. Weiterhin sollte in der Datei auhc alle weitern Ressorucen aufgelistet werden wie Topics und Consumer Groups.

Die Config und Sende und Recceaver Klassen sollten ungefähr so aussehen:

ReceiverConfig:

package faeteam3.Notlage.kafka;

import faeteam3.Notlage.model.support.*;
import java.util.HashMap;
import java.util.Map;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.PropertySource;
import org.springframework.context.support.PropertySourcesPlaceholderConfigurer;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.KafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
import org.springframework.kafka.support.serializer.JsonDeserializer;
import org.springframework.kafka.support.serializer.JsonSerializer;

@Configuration
@EnableKafka
public class ReceiverConfig {

	private final String bootstrapServers ;

	private final String id_UngeVer_group;
	private final String id_UngeRou_group;

	//  @Bean
	public Map<String, Object> consumerConfigs(String bs, String id_group) {
		Map<String, Object> props = new HashMap<>();
		// list of host:port pairs used for establishing the initial connections to the Kafka cluster
		props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
				bs);
		props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
				StringDeserializer.class);
		props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
				StringDeserializer.class);
		// allows a pool of processes to divide the work of consuming and processing records
		props.put(ConsumerConfig.GROUP_ID_CONFIG, id_group);
		// automatically reset the offset to the earliest offset
		props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
		// client.id muss eizigartig sein, auch in bezug zu vielen instanzen. ist aber optionales feld
		//    props.put("client.id", "Notlage_Client_consumer"+id_for_client);

		return props;
	}

	public ReceiverConfig(
			@Value("${eventing.brokers}") final String servers,
			@Value("${group.ungeRou}")   final String ungeRou,
			@Value("${group.ungeVer}")   final String ungeVer) 
	{
		this.bootstrapServers = servers;
		this.id_UngeRou_group=ungeRou;
		this.id_UngeVer_group=ungeVer;
	}

	@Bean
	public ConsumerFactory<String, Object> consumerFactory() {
		return new DefaultKafkaConsumerFactory<>(consumerConfigs(bootstrapServers,"no_group_xx"));
	}

	@Bean
	public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, Object>> kafkaListenerContainerFactory() {
		ConcurrentKafkaListenerContainerFactory<String, Object> factory =
				new ConcurrentKafkaListenerContainerFactory<>();
		factory.setConsumerFactory(consumerFactory());

		return factory;
	}

	@Bean
	public ConsumerFactory<String, String> consumerFactoryY1() {
		return new DefaultKafkaConsumerFactory<>(consumerConfigs(bootstrapServers,id_UngeVer_group), new StringDeserializer(), new StringDeserializer());
	}

	@Bean
	public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> kafkaListenerContainerFactoryY1() {
		ConcurrentKafkaListenerContainerFactory<String, String> factory =
				new ConcurrentKafkaListenerContainerFactory<>();
		factory.setConsumerFactory(consumerFactoryY1());

		return factory;
	}

	@Bean
	public ConsumerFactory<String, String> consumerFactoryY2() {
		return new DefaultKafkaConsumerFactory<>(consumerConfigs(bootstrapServers,id_UngeRou_group), new StringDeserializer(), new StringDeserializer());
	}

	@Bean
	public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> kafkaListenerContainerFactoryY2() {
		ConcurrentKafkaListenerContainerFactory<String, String> factory =
				new ConcurrentKafkaListenerContainerFactory<>();
		factory.setConsumerFactory(consumerFactoryY2());

		return factory;
	}


	@Bean
	public ReceiverEinheit receiver() {
		return new ReceiverEinheit();
	}
}

ReceiverEinheit:

package faeteam3.Notlage.kafka;

import java.util.concurrent.CountDownLatch;
import java.util.stream.StreamSupport;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.apache.kafka.common.header.Headers;

import faeteam3.Notlage.controller.WorkerService;
import faeteam3.Notlage.model.support.UngeRou;
import faeteam3.Notlage.model.support.UngeVer;

public class ReceiverEinheit {

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

  private CountDownLatch latch = new CountDownLatch(1);

  public CountDownLatch getLatch() {
    return latch;
  }
  
//  private final String id_notlage_topic;
//  private final String id_UngeVer_topic;
//  private final String id_UngeRou_topic;
//	  
//	public ReceiverEinheit(
//			@Value("${topics.notlage}")   final String notlage,
//			@Value("${topics.ungeRou}")   final String ungeRou,
//			@Value("${topics.ungeVer}")   final String ungeVer) 
//	{
//		this.id_notlage_topic = notlage;
//		this.id_UngeVer_topic=ungeRou;
//		this.id_UngeRou_topic=ungeVer;
//	}
  
  
  
  @Autowired
  private WorkerService serivce;
  
  @KafkaListener(topics = "${topics.ungeRou}" , containerFactory = "kafkaListenerContainerFactoryY1") //  clientIdPrefix = "json",
  public void receive_topic_n1(ConsumerRecord<String, String> cr ) {
    
    LOGGER.info("Logger 1 [JSON] received key {}: Type [{}] | Payload: {} | Record: {}", cr.key(),
    		typeIdHeader(cr.headers()), cr.value(), cr.toString());
    serivce.bearbeiteMessageUngeVer(cr.value());
    latch.countDown();
  }
  
  @KafkaListener(topics = "${topics.ungeVer}" , containerFactory = "kafkaListenerContainerFactoryY2") //  clientIdPrefix = "json",
  public void receive_topic_n2(ConsumerRecord<String, String> cr ) {
    
    
    LOGGER.info("Logger 1 [JSON] received key {}: Type [{}] | Payload: {} | Record: {}", cr.key(),
    		typeIdHeader(cr.headers()), cr.value(), cr.toString());
    serivce.bearbeiteMessageUngeRou(cr.value());
    latch.countDown();
  }
  
  private static String typeIdHeader(Headers headers) {
      return StreamSupport.stream(headers.spliterator(), false)
              .filter(header -> header.key().equals("__TypeId__"))
              .findFirst().map(header -> new String(header.value())).orElse("N/A");
}
  
}

SendeConfig:

package faeteam3.Notlage.kafka;

import java.util.HashMap;
import java.util.Map;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.PropertySource;
import org.springframework.context.support.PropertySourcesPlaceholderConfigurer;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.kafka.support.serializer.JsonSerializer;

import faeteam3.Notlage.model.Notlage;
import faeteam3.Notlage.model.support.UngeRou;
import faeteam3.Notlage.model.support.UngeVer;



@Configuration
public class SenderConfig {

  private final String bootstrapServers;

//  @Bean
  public Map<String, Object> producerConfigs(String bs,String id_for_client) {
    Map<String, Object> props = new HashMap<>();
    // list of host:port pairs used for establishing the initial connections to the Kakfa cluster
    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
    		bs);
    props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
        StringSerializer.class);
    props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
    		JsonSerializer.class);
//    props.put("client.id", "Notlage_Client_producer"+id_for_client);

    return props;
  }
  

  private static final Logger LOGGER =
	      LoggerFactory.getLogger(SendeEinheit.class);
  
 // @PropertySource(value = "classpath:application.properties")
  public SenderConfig(@Value("${eventing.brokers}") final String servers) {

	  LOGGER.info(servers);

	    this.bootstrapServers = servers;
//	    log.info(servers);
	}
  
  @Bean
  public ProducerFactory<String, Object> producerFactory() {
    return new DefaultKafkaProducerFactory<>(producerConfigs(bootstrapServers,"p1"));
  }

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

  @Bean
  public ProducerFactory<String, Notlage> producerFactoryNotlage() {
    return new DefaultKafkaProducerFactory<>(producerConfigs(bootstrapServers,"p2"));
  }

  @Bean
  public KafkaTemplate<String, Notlage> kafkaTemplateNotlage() {
    return new KafkaTemplate<>(producerFactoryNotlage());
  }
  
  
  @Bean
  public ProducerFactory<String, UngeVer> producerFactoryNotlage2() {
    return new DefaultKafkaProducerFactory<>(producerConfigs(bootstrapServers,"p3"));
  }

  @Bean
  public KafkaTemplate<String, UngeVer> kafkaTemplateNotlage2() {
    return new KafkaTemplate<>(producerFactoryNotlage2());
  }
  
  
  @Bean
  public ProducerFactory<String, UngeRou> producerFactoryNotlage3() {
    return new DefaultKafkaProducerFactory<>(producerConfigs(bootstrapServers,"p4"));
  }

  @Bean
  public KafkaTemplate<String, UngeRou> kafkaTemplateNotlage3() {
    return new KafkaTemplate<>(producerFactoryNotlage3());
  }

  @Bean
  public SendeEinheit sender() {
    return new SendeEinheit();
  }
}

SendeEinheit:

package faeteam3.Notlage.kafka;


import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

import faeteam3.Notlage.model.Notlage;
import faeteam3.Notlage.model.support.UngeRou;
import faeteam3.Notlage.model.support.UngeVer;

@Service
public class SendeEinheit {

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

  @Autowired
  private KafkaTemplate<String, Notlage> kafkaTemplate;
  
  @Autowired
  private KafkaTemplate<String, UngeVer> kafkaTemplate2;
  
  @Autowired
  private KafkaTemplate<String, UngeRou> kafkaTemplate3;
  
 
  
    @Value("${topics.notlage}")
    private String id_notlage_topic;
    @Value("${topics.ungeRou}")
    private String id_UngeVer_topic;
    @Value("${topics.ungeVer}") 
	private String id_UngeRou_topic;
	  
//	public SendeEinheit(
//			@Value("${topics.notlage}")   final String notlage,
//			@Value("${topics.ungeRou}")   final String ungeRou,
//			@Value("${topics.ungeVer}")   final String ungeVer) 
//	{
//		this.id_notlage_topic = notlage;
//		this.id_UngeVer_topic=ungeRou;
//		this.id_UngeRou_topic=ungeVer;
//	}
  

  public void sendNotlage(Notlage payload) {
    LOGGER.info("sending payload='{}'", payload);
    kafkaTemplate.send(id_notlage_topic, payload);
  }
  
  public void sendUngeVer(UngeVer payload) {
	    LOGGER.info("sending payload='{}'", payload);
	    kafkaTemplate2.send(id_UngeVer_topic, payload);
	  }
  
  public void sendUngeRou(UngeRou payload) {
	    LOGGER.info("sending payload='{}'", payload);
	    kafkaTemplate3.send(id_UngeRou_topic, payload);
	  }
}
⚠️ **GitHub.com Fallback** ⚠️