Apache Kafka

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;


import java.util.Random;
import java.util.function.Consumer;
import java.util.function.Supplier;


@Configuration
public class FirstConfiguration {
private static final String[] chaves = new String[]{"1", "2", "3", "4"};
private static final Random RANDOM = new Random(System.currentTimeMillis());
private static final Logger logger = LoggerFactory.getLogger(FirstConfiguration.class);


    @Bean
    public Consumer<Message<String>> listenFirst() {
        return mensagem -> logger.info(
                String.format("%s [%s]",
                        mensagem.getPayload(),
                        mensagem.getHeaders().get(KafkaHeaders.RECEIVED_PARTITION_ID)));
    }


    @Bean
    public Supplier<Message<String>> sendFirst() {
        return () -> {
            String chave = chaves[RANDOM.nextInt(chaves.length)];
            logger.info(String.format("Enviando %s", chave));
            return MessageBuilder
                    .withPayload(String.format("Message from Spring [%s]", chave))
                    .setHeader("partitionKey", chave)
                    .build();
        };
    }
}

spring.cloud.stream.poller.fixedDelay: # Delay fixo entre cada chamada do Supplier
spring.cloud.stream.poller.initialDelay: # Delay para iniciar as chamadas
spring.cloud.stream.poller.cron: # Expressão cron para o polling

spring:
  cloud:
    stream:
      kafka:
        bindings:
          listenFirst-in-0:
            consumer:
              startOffset: latest
      bindings:
        listenFirst-in-0:
          destination: first
          group: app5
        sendFirst-out-0:
          destination: first
          producer:
            partition-key-expression: headers['partitionKey']
            partition-count: 3
    function:
      definition: listenFirst;sendFirst

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.Message;
import reactor.core.publisher.EmitterProcessor;
import reactor.core.publisher.Flux;


import java.util.function.Consumer;
import java.util.function.Supplier;


@Configuration
public class FirstConfiguration {
private static final Logger logger = LoggerFactory.getLogger(FirstConfiguration.class);


    @Bean
    public Consumer<Message<String>> listenFirst() {
        return mensagem -> logger.info(
                String.format("%s [%s]",
                        mensagem.getPayload(),
                        mensagem.getHeaders().get(KafkaHeaders.RECEIVED_PARTITION_ID)));
    }




    @Bean
    public EmitterProcessor<Message<String>> firstProcessor() {
        return EmitterProcessor.create();
    }


    @Bean
    public Supplier<Flux<Message<String>>> sendFirst() {
        return this::firstProcessor;
    }
}

import org.springframework.http.HttpStatus;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.ResponseStatus;
import org.springframework.web.bind.annotation.RestController;
import reactor.core.publisher.EmitterProcessor;


import java.util.Random;


@RestController
public class SendFirstResource {
private static final String[] chaves = new String[]{"1", "2", "3", "4"};
private static final Random RANDOM = new Random(System.currentTimeMillis());
private final EmitterProcessor<Message<String>> firstEmitterProcessor;


    public SendFirstResource(EmitterProcessor<Message<String>> firstEmitterProcessor) {
        this.firstEmitterProcessor = firstEmitterProcessor;
    }


    @PostMapping(path = "/", consumes = "*/*")
    @ResponseStatus(HttpStatus.ACCEPTED)
    public void handleRequest(@RequestBody String body) {
        String chave = chaves[RANDOM.nextInt(chaves.length)];
        Message<String> message = MessageBuilder
                .withPayload(String.format("%s [%s]", body, chave))
                .setHeader("partitionKey", chave)
                .build();
        firstEmitterProcessor.onNext(message);
    }
}

spring:
  cloud:
    stream:
      kafka:
        bindings:
          input:
            consumer:
              startOffset: latest
      bindings:
        input:
          destination: first
          group: app-simple
        output:
          destination: first
          producer:
            partition-key-expression: headers['partitionKey']
            partition-count: 3

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.messaging.Sink;
import org.springframework.cloud.stream.messaging.Source;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;


import java.util.Random;


@EnableBinding({Source.class, Sink.class})
public class FirstService {
private static final String[] chaves = new String[]{"1", "2", "3", "4"};
private static final Random RANDOM = new Random(System.currentTimeMillis());
private static final Logger logger = LoggerFactory.getLogger(FirstService.class);
private final Source source;


    public FirstService(Source source) {
        this.source = source;
    }




    @StreamListener(Sink.INPUT)
    public void recebeMensagemFirst(Message<String> mensagem) {
        logger.info(String.format("%s [%s]",
                mensagem.getPayload(),
                mensagem.getHeaders().get(KafkaHeaders.RECEIVED_PARTITION_ID)));
    }


    public void enviaMensagemFirst() {
        String chave = chaves[RANDOM.nextInt(chaves.length)];
        logger.info(String.format("Enviando chave: %s", chave));
        source.output().send(MessageBuilder
                .withPayload(String.format("Mensagem do Spring - Estilo imperativo [%s]", chave))
                .setHeader("partitionKey", chave)
                .build());
    }


}

import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;


@EnableScheduling
@Component
public class FirstServiceScheduler {


    private final FirstService firstService;


    public FirstServiceScheduler(FirstService firstService) {
        this.firstService = firstService;
    }


    @Scheduled(fixedDelay = 1000)
    public void enviarMensagem() {
        firstService.enviaMensagemFirst();
    }
}

docker run --rm -it --name producer --net=cp-all-in-one_default confluentinc/cp-server:5.4.1 bash

docker-compose up -d ksql-cli

docker exec -ti ksql-cli bash

# ksql http://ksql-server:8088


                  ===========================================
                  =        _  __ _____  ____  _             =
                  =       | |/ // ____|/ __ \| |            =
                  =       | ' /| (___ | |  | | |            =
                  =       |  <  \___ \| |  | | |            =
                  =       | . \ ____) | |__| | |____        =
                  =       |_|\_\_____/ \___\_\______|       =
                  =                                         =
                  =  Streaming SQL Engine for Apache Kafka® =
                  ===========================================


Copyright 2017-2019 Confluent Inc.


CLI v5.4.1, Server v5.4.1 located at http://ksql-server:8088


Having trouble? Type 'help' (case-insensitive) for a rundown of how things work!


ksql>

ksql> SET 'auto.offset.reset'='earliest';
ksql> SET 'cache.max.bytes.buffering'='20000000’;
ksql> list properties;

ksql> create stream first_stream (nome VARCHAR, credito INT) with (KAFKA_TOPIC=‘first’, VALUE_FORMAT=‘DELIMITED’);
VALUE_FORMAT define o formato dos valores no topic de origem. Existem 4 tipos de formato: DELIMITED, JSON, AVRO e KAFKA.

ksql> select nome, credito from first_stream emit changes;

docker-compose up -d ksql-datagen

docker exec -ti ksql-datagen bash

ksql-datagen bootstrap-server=broker:29092 schema=./userprofile.avro format=json topic=USERPROFILE key=userid maxInterval=5000 iterations=100

create stream userprofile (userid INT, firstname VARCHAR, lastname VARCHAR, countrycode VARCHAR, rating DOUBLE) with (KAFKA_TOPIC='USERPROFILE', VALUE_FORMAT='json’);
select * from userprofile up left join countrytable ct on ct.countrycode = up.countrycode emit changes;

docker-compose up -d ksql-datagen

docker exec -ti ksql-datagen bash
ksql-datagen bootstrap-server=broker:29092 schema=userprofile.avro iterations=1000 maxInterval=3000 value-format=json topic=USERPROFILE key=userid

kafka-topics --bootstrap-server broker:29092 --topic COUNTRY-CSV --create --partitions 1 --replication-factor 1
kafka-console-producer --broker-list broker:29092 --topic COUNTRY-CSV --property "parse.key=true"  --property "key.separator=:”
>AU:AU,Australia
>IN:IN,India
>GB:GB,UK
>US:US,United States

create table countrytable (countrycode VARCHAR, countryname VARCHAR) with (KAFKA_TOPIC='COUNTRY-CSV', VALUE_FORMAT='DELIMITED', KEY='countrycode’);
describe extended COUNTRYTABLE;
select rowkey, countrycode, countryname from countrytable emit changes;

>GB:GB,UK
>BR:BR,Brasil

select * from userprofile up left join countrytable ct on ct.countrycode = up.countrycode emit changes;

create stream userdetails as select up.firstname + ' ' + ucase(up.lastname) + ' from ' + ct.countryname + ' has a rating of ' + cast(up.rating as varchar) + ' stars' as description from userprofile up left join countrytable ct on ct.countrycode = up.countrycode emit changes;

select description from userdetails;

kafka-console-consumer --bootstrap-server localhost:9092 --topic USERDETAILS --from-beginning

create stream driverLocations (driverId INTEGER, countrycode VARCHAR, city VARCHAR, drivername VARCHAR) WITH (kafka_topic='driverlocations', key='driverId', value_format='json', partitions=1);

insert into driverLocations (driverId, countrycode, city, drivername) VALUES (1, 'AU', 'Sydney', 'Alice');
insert into driverLocations (driverId, countrycode, city, drivername) VALUES (2, 'US', 'New York', 'Bob');
insert into driverLocations (driverId, countrycode, city, drivername) VALUES (3, 'BR', 'Brasília', 'Roberto');
insert into driverLocations (driverId, countrycode, city, drivername) VALUES (4, 'IN', 'Mumbai', 'Arun’);
insert into driverLocations (driverId, countrycode, city, drivername) VALUES (5, 'AU', 'Melbourne', 'Jack’);

create table countryDrivers as select countrycode, count(*) as numDrivers from driverLocations group by countrycode;
select * from countryDrivers where rowkey='AU';