Changes

Azt, hogy egy producer melyik partícióba dobja az üzenetet vagy a kulcs határozza meg, vagy round-rubin robin módon mindig egy másikba teszi. Ha van kulcs, akkor az abból készült hash fogja meghatározni, hogy melyik partícióba kerüljön. Ugyan az a kulcs így mindig ugyan abba a partícióba fog kerülni. De a kulcs nem kötelező. A sorrend tartás csak egy partíción belül garantált, de ott nagyon. Ha nagyon kritikus bizonyos üzenetek sorrendje, akkor azokat egy partícióba kell rakni azonos kulcsot használva. Loggolásnál ez nem kritikus, egyrészt mert a logstash sorba rakja az üzeneteket, másrészt mikor elastichsearch-be szúrjuk, ott a dátum lesz az egyik attribútum, ami alapján már sorba lehet majd újra rendezni a logokat. Az meg amúgy sem kritikus, ha a log egy része enyhe csúszással kerül be az adatbázisba, lényeg, hogy végül helyes lesz a sorrend.

 =Java consumer==Pom.xml=== <source lang="xml"> <dependencies> ...  <dependency> <groupId>com.github.danielwegener</groupId> <artifactId>logback-kafka-appender</artifactId> <version>0.2.0-RC1</version> <scope>runtime</scope> </dependency> <dependency> <groupId>org.apache.kafka</groupId> <artifactId>connect-json</artifactId> <version>2.1.1</version> </dependency>   <dependency> <groupId>ch.qos.logback</groupId> <artifactId>logback-classic</artifactId> <version>1.2.3</version> <scope>runtime</scope> </dependency>  <dependency> <groupId>org.slf4j</groupId> <artifactId>slf4j-api</artifactId> <version>1.7.25</version> </dependency> .... </dependencies></source>

=Sorting messages with Logstash==Logback.xml===

A logstashlogback-t rakhatjuk ek Kafka topic-okba a '''com.github.danielwegener.logback.kafka elé és a kafka után is. Első lépésként a KafkaAppender''' osztállyal lehet írni ami egy szabványos Logback appender. Teljes leírás itt: https://github.com/danielwegener/logback-kafka elé fogjuk tenni, ami szortírozni fogja a logokat különböző topic-okbaappender<br>logback.xml<source lang="xml"><configuration> <shutdownHook class="ch.qos.logback.core.hook. DelayingShutdownHook"/>

<appender name="kafkaAppender" class=Producer"com.github.danielwegener.logback.kafka.KafkaAppender"> <encoder class="net.logstash.logback.encoder.LogstashEncoder"> <customFields>{"application":"this is the extra field"}</customFields> </encoder> <topic>test2-topic</topic> <keyingStrategy class="com.github.danielwegener.logback.kafka.keying.HostNameKeyingStrategy" /> <deliveryStrategy class="com.github.danielwegener.logback.kafka.delivery.AsynchronousDeliveryStrategy" /> <appendTimestamp>true</appendTimestamp> <producerConfig>bootstrap.servers=kafka:9092</producerConfig> </appender>

<root level="info"> <appender-ref ref="kafkaAppender" /> </root></configuration></source>  Nagyon fontos, hogy leállítsuk a logger context-et mielőtt a VM leáll, mert ha a VM leállítása nagyon közel van a log beíráshoz, akkor még azelőtt leáll az egész VM, hogy a logokat kiírtuk volna. Nagyon rövid életű programokban, mint amilyen a mi példa programunk, hamarabb leállhat a VM, mint hogy el tudta volna küldeni a logback a Kafka-nak az üzenetet. Ha még a VM leállítása előtt meghívjuk a sthudownHook-ot, akkor leállás előtt még ki fogja írni a logokat.<source lang="xml"><shutdownHook class="ch.qos.logback.core.hook.DelayingShutdownHook"/></source>  A kulcs kezelési stratégiát a '''keyingStrategy''' paraméterben kell definiálni. A futtatásához szükséges környezet kulcsok kitöltése nem kötelező, de szintén hatással lehet a performanciára. Ha a kulcs minden üzenetben ugyan az, akkor az összes üzenet ugyanabba a partícióba fog kerülni, ami nem a legjobb, de cserébe sorrendtartó lesz. Több kulcskezelési stratégia közül választhatunk. A '''NoKeyKeyingStrategy''' hatására nem fog kulcsot generálni, így round robin módon fog mindig egy két nodeúj partíciót választani. Ha a '''HostNameKeyingStrategy''' stratégiát választjuk, akkor a host név lesz a kulcs, tehát az azonos hostról érkező logsorok mindig ugyan abba a partícióba fognak kerülni. <source lang="xml"><keyingStrategy class="com.github.danielwegener.logback.kafka.keying.NoKeyKeyingStrategy" />VAGY<keyingStrategy class="com.github.danielwegener.logback.kafka.keying.HostNameKeyingStrategy" /></source>  A Kafka brókerek listáját a '''bootstrap.servers''' producerConfig paraméterben kell megadni. Nagyon fontos, hogy ugyan azzal a host névvel tegyük ezt ide, mint ahogy a swarm-ban létrehoztuk, és az itt megadott nevet fel kell venni a hosts-ba. A producerConfig-os swarm cluster leszok teljes listája itt olvasható: ttps://kafka.apache.org/documentation.html#producerconfigs<source lang="xml"><producerConfig>bootstrap.servers=kafka:9092</producerConfig></source> A '''customFields''' paraméterben tetszőleges log paramétereket adhatunk a Kafka üzenethez. <source lang="xml"><customFields>{"application":"this is the extra field"}</customFields></source> <br> ===Java kód=== <source lang="java">import org.slf4j.Logger;import org.slf4j.LoggerFactory; public class LogbackExample {  private static final Logger logger = LoggerFactory.getLogger(LogbackExample.class.getSimpleName());  public static void main(String... args) throws InterruptedException { logger. info("this is the message:"); }}</source>  ===Futtatás===Indítsuk el a kafka-console-consumer -t a test2-topic-ra, hogy lássuk, hogy a logback milyen üzeneteket tesz be:

# docker node ls./kafka-console-consumer \ID --bootstrap-server kafka:29092 \ --topic test2-topic \ --from-beginning</pre> Futtassuk le a LogbackExample java programot, ekkor a consumer ki fogja írni a logback által beküldött üzenetet: <pre>{"@timestamp":"2019-04-21T12:44:32.430+02:00","@version":"1","message":"this is the message:","logger_name":"LogbackExample","thread_name":"main","level":"INFO","level_value":20000,"HOSTNAME STATUS AVAILABILITY MANAGER STATUS ENGINE VERSION":"adamDell2","application":"this is the extra field"}</pre>Látható, hogy bekerült az üzenetbe a timestamp is és az extra mező is, amit a logback appender-ben adtunk hozzá. <br>  <br> ===Custom log object===Ahogy a logback hagyományos használata mellett, itt is lehetőség van egyedi üzenet objektumok használatára. maigxlyagj1fl4sgcf6rnn9pc * mg0 Ready Active Leader 181.05Az egyik lehetőség az MDC - Mapped Diagnostic Context (https://logback.0qos.ch/manual/mdc.html) használata, amivel egyedi mezőket adhatunk hozzá a log-cehoz, ami a kafka üzenetben is meg fog jelenni: <source lang="java">import org.slf4j.MDC;...MDC.put("transactionId", "1111");logger.info("this is the message:");</source> A fenit üzenet a kafka consumer-ben így fog megjelenni: <pre>{"@timestamp":"2019-04-21T16:20:34.620+02:00","@version":"1","message":"this is the message:","logger_name":"LogbackExample","thread_name":"main","level":"INFO","level_value":20000,"HOSTNAME":"adamDell2","transactionId":"1111","application":"this is the extra field"}</pre> <br><br>2. A másik lehetőség a '''net.logstash.logback.marker.Markers''' használata, amivel tetszőleges java POJO-kat írhatunk be JSON formátumban a logba. A példában az alábbi '''LogMessage''' java objektumot fogjuk használni. <source lang="java">public class LogMessage { vox99u5s1g1su742mc6npm370 worker0 Ready Active 18private String feild1; private String field2;  public LogMessage(String feild1, String field2) { this.feild1 = feild1; this.field2 = field2; }  public String getFeild1() { return feild1; } public String getField2() { return field2; } public void setFeild1(String feild1) { this.feild1 = feild1; } public void setField2(String field2) { this.field2 = field2; }}</source>A java osztályban az import-ok közé felvesszüka Markers.append-t. <source lang="java">import org.slf4j.Logger;import org.slf4j.LoggerFactory; import static net.logstash.logback.marker.Markers.append; public class LogbackExample {  private static final Logger logger = LoggerFactory.getLogger(LogbackExample.class.getSimpleName());  public static void main(String..05.0args) throws InterruptedException {  LogMessage message = new LogMessage("first", "second"); logger.info(append("customFieldName", message), "this is the message"); }}</source>A consume-ben az üzenet az alábbi lesz: <pre>{"@timestamp":"2019-ce04-21T16:20:35.140+02:00","@version":"1","message":"this is the message","logger_name":"LogbackExample","thread_name":"main","level":"INFO","level_value":20000,"HOSTNAME":"adamDell2","transactionId":"1111","customFieldName":{"feild1":"first","field2":"second"},"application":"this is the extra field"}

A korábban ismertetett docker stack-et ki fogjuk bővíteni egy logstash service-el. :[[File:ClipCapIt-190327190421-224419163339.PNG]]

A loggolásra logback-et fogunk használni, aki a logstash 51415-ös portjára fogja küldeni TCPsocket-en. Az üzeneteket a '''kafka-console-consumer'''-el fogjuk kiolvasni.

A lostash a TCP socket-en keresztül várja majd a logback-től a logokat. A logberben Marker-eket fogunk használni, amik a [tags] tömbbe fog tenni a logstash. A logstash a Kafka output plugin segítségével fogja beírni a megfelelő topic-ba az üzeneteket.  :[[File:ClipCapIt-190421-163511.PNG]] Az alap logstash image már tartalmazz mind a Kafka input és output plugin-t is, így kapásból tudunk a logstash-el Kafka-ból írni és olvasni. Listázzuk ki a logstash plugin-eket a '''bin/logstash-plugin list''' paranccsal. Láthatjuk hogy a kafka mind az input mind az output-ban ott van. <pre># docker run -it docker.elastic.co/logstash/logstash:6.6.2 bin/logstash-plugin list...logstash-input-kafka...logstash-output-kafka</pre> 

Fontos, hogy a logstash-ből a 6.6-os szériát használjuk, mert a korábbi verziókban van egy kafak specifikus hiba. A logstash konfigurációt volume driver-er fogjuk felcsatolnihost gépről. (A volume dirver-ekről részletek itt: https://wiki.berki.org/index.php/Docker_volume_orchestration)

===logoutlogback.xml===

<!-- <shutdownHook/> --> <!-- Nagyon fontos, hogy leállítsuk a logger context-et mielőtt a VM leáll, mert ha a VM leállítása nagyon közel van a log beíráshoz, akkor még azelőtt leáll az egész VM, hogy a logokat kiírtuk volna. de ha még a VM leállítása előtt meghívjuk a sthudownHook-ot, akkor leállás előtt még ki fogja írni a logokat. --> <shutdownHook class="ch.qos.logback.core.hook.DelayingShutdownHook"/>

===Java producerlogger example===

Logstash Mivel a logstash config-ba beletettük az '''stdout''' output-ot is, ezért a log-ba is be fog írni minden üzenetet:

 =Consumer= ==Command line consumer==A legegyszerűbben a '''kafka-console-consumer'''-el olvashatunk egy topic-ot, ez is része a standard Apache Kafka és a Confluent csomagnak is. Akárcsak a kafka-console-producer, ez is a bin mappában találathó és szintén a Kafka brokert és a topic nevét kell megadni. <pre>./kafka-console-consumer \ --bootstrap-server kafka:29092 \ --topic test2-topic \ --from-beginning</pre> Ha elindítottuk a '''kafka-console-consumer'''-t, akkor a Kafka service logjában láthatjuk, hogy a console consumer regisztrálta magát és ő lett a csoport vezetője. Minden consumer csoportnak van egy vezetője, akin keresztül a többi consumer a csoportban megkapja a konfigurációs változásokat. <pre># docker service logs -f confluent_kafka...INFO [GroupCoordinator 2]: Assignment received from leader for group console-consumer-73627 for generation 1 (kafka.coordinator.group.GroupCoordinator)</pre> A '''kafka-console-consumer''' mi fog loggolni minden egyes új üzenetet a topic-ról. <pre>{"@timestamp":"2019-04-21T12:44:32.896+02:00","@version":"1","message":"LogMessage@49d904ec","logger_name":"LogbackExample","thread_name":"main","level":"INFO","level_value":20000,"HOSTNAME":"adamDell2","transactionId":"444","metric":{"feild1":"first","field2":"second"},"application":"this is the extra field"}</pre><br> ==Java consumer==A Java consumer-nek ugyan azokra a maven függőségekre van szüksége mint a producer-nek, ezrét ezeket itt nem ismételjük meg.<source lang="java">import org.apache.kafka.clients.consumer.*;import org.apache.kafka.clients.consumer.Consumer;import org.apache.kafka.common.serialization.LongDeserializer;import org.apache.kafka.common.serialization.LongSerializer;import org.apache.kafka.common.serialization.StringDeserializer; import java.time.Duration;import java.util.Collections;import java.util.Properties; public class KafkaConsumerExample { private final static String TOPIC = "test2-topic";  private static Consumer<String, String> createConsumer() { final Properties props = new Properties(); props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka:29092"); props.put(ConsumerConfig.GROUP_ID_CONFIG, "KafkaExampleConsumer"); props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, LongDeserializer.class.getName()); props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());  final Consumer<String, String> consumer = new KafkaConsumer<>(props);  consumer.subscribe(Collections.singletonList(TOPIC)); return consumer; }  static void runConsumer() throws InterruptedException { final Consumer<String, String> consumer = createConsumer(); final int giveUp = 100; int noRecordsCount = 0; while (true) { final ConsumerRecords<String, String> consumerRecords = consumer.poll(Duration.ofSeconds(1)); if (consumerRecords.count() == 0) { noRecordsCount++; if (noRecordsCount > giveUp) break; else continue; } consumerRecords.forEach(record -> { try { System.out.printf("Consumer Record:(key: %s value: %s, partition: %d, offset: %d)\n",record.key(), record.value(), record.partition(), record.offset()); } catch (Exception e) { e.printStackTrace(); } }); consumer.commitAsync(); } consumer.close(); System.out.println("DONE"); }  public static void main(String... args) throws Exception { runConsumer(); }}</source> A Kafka specifikus beállításokat szintén a Properties map-ben kell megadni, és megegyeznek a producer-nél bemutatott paraméterekkel. <source lang="java">props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka:29092");props.put(ConsumerConfig.GROUP_ID_CONFIG, "KafkaExampleConsumer");props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, LongDeserializer.class.getName());props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());</source> Fontos, hogy csak olyan deszerializátort használjunk, ami kompatibilis az üzenet fajtájával, vagyis olyat, ami kompatibilis a producer-nél használt serializátórral. Pl. egy JsonSerializer-el írt üzenetet ki lehet olvasni egy StringDeerializer-el is. <source lang="java">props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());VAGYprops.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArrayDeserializer");VAGYprops.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.connect.json.JsonDeserializer");</source> <br> ==Spring-Kafa consumer== https://www.baeldung.com/spring-kafka  ==Alpakka-kafka== https://doc.akka.io/docs/alpakka-kafka/current/home.html ==Logstash consumer==https://www.elastic.co/guide/en/logstash/6.7/plugins-inputs-kafka.htmlAhogy azt már láthattuk, a logstash lehet Kafka producer és consumer szerepben is, mind a Kafka input és output plugin-t is tartalmazza az alap logstash image.  :[[File:ClipCapIt-190421-191220.PNG]]  Ugyan azt a docker stack-et fogjuk használni, amit a logstash producer-nél használtunk, csak a konfigurációt fogjuk módosítani, hogy a kafa az input ne az output plugin-ben legyen: https://wiki.berki.org/index.php/Apache_Kafka#Logstash_producer_with_logback Mivel a logstash most Kafka consumer szerepben lesz, ezért most a Kafka input plugin-t fogjuk használni. A logstash konfigurációt docker volume-al fogjuk felcsatolni a service-t futtató konténerbe. A logstash a belső, kafka-net overlay hálózaton keresztül közvetlen el fogja érni a Kafka brókert, ezért a konfigurációban a Kafka service nevét kell megadni, amit a swarm fel fog oldani a service-ben lévő konténerek IP címére.  /usr/share/logstash/pipeline/logstash.conf<pre>input { kafka {  decorate_events => true value_deserializer_class => "org.apache.kafka.common.serialization.StringDeserializer"  topics => ["test2-topic"]  bootstrap_servers => "kafka:29092" group_id => "AvroConsumerGroupId" client_id => "AvroConsumerClientId"   }} output {  stdout { codec => rubydebug }}</pre>  Ha a logstash elindult, akkor a swarm service logjában láthatjuk, hogy rákapcsolód a test2-topic-ra. <pre># docker service logs -f confluent_logstash ...[Consumer clientId=AvroConsumerClientId-0, groupId=AvroConsumerGroupId] Resetting offset for partition test2-topic-0 to offset 0.</pre> Írjunk be egy üzenetet a test2-topic-ba a '''kafka-console-producer'''-el. <pre>$ ./kafka-console-producer \> --broker-list kafka:29092 \> --topic test2-topic>this is the test message </pre> Ekkor a logstash logjában meg fog jelenni a beírt üzenet message paraméterben. A logstash kiegészíti két meta paraméterrel az üzenetet (timestamp és version). <pre>confluent_logstash.1.3qsuyylnulxa@worker0 | {confluent_logstash.1.3qsuyylnulxa@worker0 | "@timestamp" => 2019-04-21T17:55:08.694Z,confluent_logstash.1.3qsuyylnulxa@worker0 | "message" => "this is the test message",confluent_logstash.1.3qsuyylnulxa@worker0 | "@version" => "1"confluent_logstash.1.3qsuyylnulxa@worker0 | }</pre> =Adminisztrációs felületekeszközök= * nodefluent/kafka-rest* nodefluent/kafka-rest-ui* sheepkiller/kafka-manager* tobilg/zookeeper-webui