Thiago Colen |

Firebase Functions e Firestore: executando localmente (comandos úteis)

2024-12-27T16:11:00.000-08:00

Neste post irei mostrar alguns comandos úteis para rodar e usar localmente o Google Firebase.

Primeiro você precisa inicializar seu projeto:

$ firebase init

Em seguida, será questionado sobre quais emuladores deseja instalar (por exemplo: Functions, Firestore, PubSub...).

Uma vez configurado, você precisa iniciar os emuladores:

$ firebase emulators:start

Com os emuladores no ar, inclusive a UI, você terá apoio visual, como abaixo:

No print acima, nota-se que há três emuladores: Firestore, Functions e PubSub.

E se você quiser testar alguma function? Primeiro execute o comando abaixo:

$ firebase functions:shell

Em seguida, chame a sua função pelo nome:

firebase > suaFuncao()

Depois de testar as funções e também os acessos às coleções do Firestore, faça o deploy remoto (produção):

firebase deploy --only functions

firebase deploy --only firestore

Kafka Confluent REST Proxy

2024-01-05T16:29:00.000-08:00

O Kafka da Confluent tem alguns extras interessantes. Um deles é o REST Proxy. Neste post mostro exemplos simples de como utilizá-lo.

Como pré-requisito, você precisa do Kafka da Confluent instalado, e este outro post aqui trata disso.

Feita a instalação, você precisa executar os dois comandos abaixo:

sudo systemctl start confluent-schema-registry confluent-kafka-rest

sudo systemctl enable confluent-schema-registry confluent-kafka-rest

Isto irá iniciar o Schema Registry e também o REST Proxy.

Para verificar se está tudo em ordem, execute:

sudo systemctl status confluent-schema-registry confluent-kafka-rest

Com eles no ar, vamos criar um tópico para iniciar a demonstração:

kafka-topics --bootstrap-server localhost:9092 --create --topic authors --partitions 1 --replication-factor 1

E agora vem o primeiro uso do REST Proxy: postar uma mensagem no tópico authors criado acima:

curl -X POST -H "Content-Type: application/vnd.kafka.json.v2+json" \

-H "Accept: application/vnd.kafka.v2+json" \

--data '{"records":[{"key":"EN","value":"Robert Hugh Benson"}]}' "http://localhost:8082/topics/authors"

Aqui é usado o o curl (com alguns headers), mas o principal é o endereço que estamos chamando (/topics/authors) e o payload com um autor (um json). Este é o primeiro uso do REST Proxy (/topics).

Executar essa chamada irá retornar algo assim:

{"offsets":[{"partition":0,"offset":0,"error_code":null,"error":null}],"key_schema_id":null,"value_schema_id":null}

Note que há algumas informações no retorno, como partição e offset.

Agora iremos consumir mensagens deste mesmo tópico. Este processo requer três etapas.

1) Criar um consumidor e uma instância deste consumidor:

curl -X POST -H "Content-Type: application/vnd.kafka.v2+json" \

--data '{"name": "consumer_authors_instance", "format": "json", "auto.offset.reset": "earliest"}' \

http://localhost:8082/consumers/consumer_authors

O ponto importante acima é o endereço: /consumers/<nome_do_seu_consumidor>. No caso eu escolhi consumer_authors.

Além disso, o payload também é importante. Nele estamos definindo o nome da instância (consumer_authors_instance), o formato e de onde iremos ler tópico ("auto.offset.reset": "earliest").

Obs.: auto.offset.reset:earliest serve para ler as mensagens a partir do início do tópico. A opção default é latest, e significa que a leitura das mensagens não se dará do início, mas a partir das mensagens mais novas (escritas depois que o consumidor entrou no ar).

Neste ponto temos outro uso do REST Proxy: /consumers.

Quando este comando for executado, ele irá retornar algo assim:

{"instance_id":"consumer_authors_instance","base_uri":"http://localhost:8082/consumers/consumer_authors/instances/consumer_authors_instance"}

Repare bem no atributo base_uri acima, em negrito. Seu valor será usado daqui pra frente.

2) Se inscrever no tópico (subscribe):

curl -X POST -H "Content-Type: application/vnd.kafka.v2+json" \

--data '{"topics":["authors"]}' \

http://localhost:8082/consumers/consumer_authors/instances/consumer_authors_instance/subscription

Repare no comando acima que no payload informamos em qual tópico queremos nos inscrever. E depois, repare na url. É a base_uri retornada anteriormente com a adição de /subscription. Em outras palavras, neste comando estamos dizendo que a instância do consumidor está associada ao tópico authors. Novamente estamos usando um endpoint do REST Proxy.

3) Consumir as mensagens:

curl -X GET -H "Accept: application/vnd.kafka.json.v2+json" \

http://localhost:8082/consumers/consumer_authors/instances/consumer_authors_instance/records

Novamente aparece aqui a base_uri com outro adendo: /records.

Ao executar esta chamada, será retornado algo semelhante a isso, com as mensagens que estão no tópico:

[{"topic":"authors","key":"EN","value":"Robert Hugh Benson","partition":0,"offset":0},{"topic":"authors","key":"EN","value":"G. K. Chesterton","partition

":0,"offset":1}]

Quando não precisar mais do consumer, você deve deletá-lo:

curl -X DELETE -H "Content-Type: application/vnd.kafka.v2+json" \

http://localhost:8082/consumers/consumer_authors/instances/consumer_authors_instance

Há muitos outros endpoints úteis. Dois exemplos:

Para listar os tópicos existentes: http://localhost:8082/topics. Isto retornará algo como:

["__confluent.support.metrics","_schemas","authors","rest-test-topic"]

Para obter informações sobre partições de um tópico: http://localhost:8082/topics/authors/partitions. Isto retornará algo assim:

[{"partition":0,"leader":1,"replicas":[{"broker":1,"leader":true,"in_sync":true}]}]

Instalação on-premise do Kafka da Confluent no Ubuntu (Systemd)

2023-12-26T17:32:00.000-08:00

Este post será dividido em três partes: instalação, configuração e inicialização. E vai considerar um cluster / ensemble de três servidores / brokers.

Obs.: a maior parte dos passos descritos aqui devem ser executados nos N hosts do cluster.

Instalação

Instale a chave pública da Confluent:

wget -qO - https://packages.confluent.io/deb/7.0/archive.key | sudo apt-key add -

Adicione o repositório da Confluent:

add-apt-repository "deb [arch=amd64] https://packages.confluent.io/deb/7.0 stable main"

Instale a versão Community:

apt-get update && sudo apt-get install -y openjdk-8-jre-headless confluent-community-2.13

Configuração

Aqui serão consideradas as configurações do ZooKeeper e do Kafka.

A primeira coisa é definir o id do servidor ZooKeeper. Para isso, crie um arquivo chamado myid no diretório /var/lib/zookeeper. Coloque neste arquivo o id, por exemplo 1. Obs.: em cada um dos N hosts, este arquivo deverá ser criado e preenchido com seu respectivo id. Portanto, no primeiro host, 1; no segundo host, 2; no terceiro host, 3 (considerando que é um cluster de 3 nós).

Depois, edite as configurações do ZooKeeper neste arquivo: /etc/kafka/zookeeper.properties.

tickTime=2000

dataDir=/var/lib/zookeeper/

clientPort=2181

initLimit=5

syncLimit=2

server.1=host1:2888:3888

server.2=host2:2888:3888

server.3=host3:2888:3888

O conteúdo acima pode conter mais ou menos propriedades e depende de vários fatores do seu ambiente e de seus requisitos. Segue uma breve explicação sobre os que foram usados aqui:

tickTime: é uma unidade, usada com initLimit e syncLimit (uma variável, ver abaixo);
dataDir: local onde o arquivo myid se encontra;
clientPort: porta na qual os membros irão se conectar ao ensemble;
initLimit: tempo para os followers se conectarem com um líder (inicialização). Este valor é multiplicado pelo tickTime;
syncLimit: tempo que um follower pode ficar dessincronizado de um líder. Este valor é multiplicado pelo tickTime;
server.N: onde N é o id do servidor (membro do cluster). Em seguida você tem o host e duas portas. A primeira é a que será utilizada na comunicação entre os nós. Enquanto a segunda porta é usada durante a eleição do líder.

Agora vamos às configurações do Kafka. Edite o arquivo /etc/kafka/server.properties e faça as seguintes alterações:

broker.id: defina um identificador único para o broker (único dentro do cluster);
zookeeper.connect: host e porta do Zookeeper. Você pode especificar N, separando-os por vírgula. É onde os metadados do cluster serão buscados.

Inicialização

Execute estes comandos:

systemctl start confluent-zookeeper

systemctl enable confluent-zookeeper

systemctl start confluent-kafka

systemctl enable confluent-kafka

Para verificar se os serviços estão rodando: systemctl status confluent*

Um outro teste útil, é listar os tópicos: kafka-topics --list --bootstrap-server localhost:9092

Isto deverá gerar uma saída com este valor: _confluent.support.metrics

Obs.: não fiz a configuração do Schema Registry aqui, mas farei em outra postagem futura.

Kafka: compactação de tópicos

2023-10-16T19:42:00.003-07:00

O Kafka provê um meio para compactar tópicos. Na prática, é um mecanismo de retenção baseado na chave da mensagem -- não confundir com a retenção por tempo, nem com a compressão de mensagens.

A compactação é útil em cenários onde você está interessado somente na mensagem / informação mais atualizada ou recente, por exemplo:

Endereço postal ou eletrônico de um cliente;
Estado de uma aplicação ou de um registro / objeto de negócio;
Preços;
Geolocalização.

Para fazer isso o Kafka elimina todas as mensagens com chaves repetidas e mantém apenas a mais atual.

Existem duas situações onde a compactação é iniciada / disparada:

Proporção: quando 50% do tópico contiver dados sujos, ou seja, dados ainda não analisados pelo processo de compactação. A partição é dividida em duas áreas (limpa e suja ou clean e dirty). A parte limpa é a que já não contém chaves repetidas (ou seja, onde o processo de compactação já rodou);
Tempo máximo para compactação: através da propriedade max.compaction.lag.ms é possível informar o tempo máximo, a partir do qual a compactação irá rodar, independente da proporção de dado sujo. Em outras palavras, é o tempo máximo entre a escrita da mensagem e ela estar disponível para compactação.

Abaixo irei mostrar como fazer o segundo cenário (tempo máximo). A primeira coisa é alterar o server.properties e adicionar duas linhas:

log.cleanup.policy=compact

max.compaction.lag.ms=10000

A primeira linha acima indica que iremos usar a compactação (ela é habilitada) e a segunda linha indica que o tempo máximo é de 10 minutos.

Importante: somente segmentos fechados ou inativos são elegíveis para compactação (este post aqui pode ajudar a entender o que é um log segment fechado).

Prosseguindo com a explicação, o próximo passo é postar mensagens com chaves repetidas. Após postar 25 mil mensagens (5 chaves diferentes, enviadas 5 mil vezes), os segmentos ficaram assim:

963K out 16 23:08 00000000000000000000.log

975K out 16 23:08 00000000000000009159.log

691K out 16 23:08 00000000000000018431.log

Acima, o log segment aberto ou ativo é o final 18431.

Após 10 minutos, se olhar novamente a pasta do tópico e da partição, você verá os segmentos um pouco diferentes:

0 out 16 23:08 00000000000000000000.log

599 out 16 23:08 00000000000000009159.log

691K out 16 23:08 00000000000000018431.log

Nota-se agora que os dois primeiros segmentos mudaram de tamanho, estão mais vazios. O primeiro zerado e o segundo bem pequeno. Já o segmento ativo (final 18431 segue com o mesmo tamanho).

Se você acessar o segundo segmento, verá que ele possui apenas 5 mensagens (as 5 chaves diferentes que mencionei acima):

| offset: 46114 CreateTime: 1697509329848 keySize: 8 valueSize: 87 sequence: 21114 headerKeys: [] key: Luke1,30 payload: Then the angel said to her, "Do not be afraid, Mary, for you have found favor with God.

| offset: 46115 CreateTime: 1697509329848 keySize: 8 valueSize: 94 sequence: 21115 headerKeys: [] key: Luke1,26 payload: In the sixth month, the angel Gabriel was sent from God to a town of Galilee called Nazareth,

| offset: 46116 CreateTime: 1697509329848 keySize: 8 valueSize: 100 sequence: 21116 headerKeys: [] key: Luke1,27 payload: to a virgin betrothed to a man named Joseph, of the house of David, and the virgin's name was Mary.

| offset: 46117 CreateTime: 1697509329848 keySize: 8 valueSize: 71 sequence: 21117 headerKeys: [] key: Luke1,28 payload: And coming to her, he said, "Hail, favored one! The Lord is with you."

| offset: 46118 CreateTime: 1697509329848 keySize: 8 valueSize: 96 sequence: 21118 headerKeys: [] key: Luke1,29 payload: But she was greatly troubled at what was said and pondered what sort of greeting this might be.

Testado no Kafka 3.4.0.

Kafka: como definir o tamanho de um log segment

2023-10-16T13:16:00.002-07:00

Log segment é um arquivo que armazena as mensagens no Kafka. O tamanho padrão dele é 1GB, mas você pode modificar isso através da propriedade log.segment.bytes (no arquivo server.properties). Ao definir o tamanho de 1.000.000 (~975K) ficará fácil ver N log segments no diretório da partição:

$pwd

/tmp/kafka-logs/verses-0

$ls -lah *.log

974K out 16 16:24 00000000000000000000.log

975K out 16 16:24 00000000000000009267.log

679K out 16 16:24 00000000000000018539.log

Acima você pode ver que estamos no diretório verses-0, ou seja, tópico verses, partição 0. Em seguida, aparecem 3 log segments. Os dois primeiros de ~975K (já fechados) e o segundo de um pouco menos.

Nota: estes números nos nomes dos arquivos são os offsets de início (starting offset). Por exemplo, o offset inicial do arquivo 00000000000000000000.log é 0. Já do arquivo 00000000000000009267.log é 9267 e por fim, do arquivo 00000000000000018539.log é 18539.

Alerta: definir um log segment muito pequeno reduzirá a performance, pois mais escritas serão necessárias, já que log segments serão fechados e alocados mais frequentemente.

Testado no Kafka 3.4.0.

Kafka: como visualizar as mensagens em uma partição (log segment)

2023-10-13T20:30:00.001-07:00

Para visualizar mensagens em uma partição, utilize o comando abaixo:

kafka-run-class.sh kafka.tools.DumpLogSegments --print-data-log --files /tmp/kafka-logs/comp-lz4-0/00000000000000000000.log

A flag print-data-log indica que você deseja ver o conteúdo da mensagem. Já a flag files indica qual arquivo será lido. Neste exemplo acima, comp-lz4-0 é o nome do tópico (comp-lz4) e 0 é a partição.

A saída deste comando será algo assim:

| offset: 0 CreateTime: 1697250280386 keySize: 1 valueSize: 71 sequence: 0 headerKeys: [] key: a payload: In principio erat Verbum, et Verbum erat apud Deum, et Deus erat Verbum

Note que é exibido o offset, a key e o payload entre outras coisas.

Testado no Kafka 3.4.0.

Kafka: compressão de mensagens

2023-10-13T20:04:00.006-07:00

Por que é bom comprimir mensagens? Porque isto reduz o uso da rede e também do disco. Portanto, faça isso se estes dois aspectos são motivos de preocupação para você.

Isto pode ser feito através do parâmetro compression.type e o padrão é sem compressão ou uncompressed. Os tipos de compressão (algoritmos) disponíveis são: zstd, lz4, snappy e gzip.

Lembrando que a mensagem é comprimida antes de ser enviada ao broker.

No teste que fiz, postei 2.000 mensagens num tópico. O resultado da compressão ficou como abaixo:

Portanto, a opção que mais comprime é zstd, seguida de gzip, lz4 e snappy. A primeira opção -- zstd -- representa uma economia de 96%. Enquanto a última -- snappy -- economiza 88%.

Segue um exemplo de como definir a compressão na criação de um tópico:

kafka-topics.sh --bootstrap-server localhost:9092 --create --replication-factor 1 --partitions 1 --config compression.type=zstd --topic comp-zstd

Para verificar o tamanho ocupado, basta verificar o diretório onde as mensagens são armazenadas (definido através da propriedade log.dirs no server.properties). Em seguida, basta acessar o diretório correspondente ao seu tópico (por exemplo: /tmp/kafka-logs/comp-snappy-0) e verificar o arquivo .log:

18K out 13 23:26 00000000000000000000.log

Kafka: como deletar tópicos

2023-10-13T19:17:00.005-07:00

Para deletar um tópico, basta usar o comando abaixo:

kafka-topics.sh --bootstrap-server localhost:9092 --delete --topic topic1

Para deletar vários tópicos, basta separá-los por vírgula:

kafka-topics.sh --bootstrap-server localhost:9092 --delete --topic topic1,topic2,topic3

Testado no Kafka 3.4.0.

Kafka: consultando configurações com AdminClient

2023-09-13T17:23:00.002-07:00

Aqui será mostrado como ler algumas configurações -- compressão e retenção -- com o AdminClient.

Properties properties = new Properties();
properties.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
AdminClient admin = AdminClient.create(properties);

ConfigResource resource = new ConfigResource(ConfigResource.Type.TOPIC, "seila");
DescribeConfigsResult result = admin.describeConfigs(Collections.singleton(resource));
Config configs = result.all().get().get(resource);

ConfigEntry compression = configs.get(TopicConfig.COMPRESSION_TYPE_CONFIG);
System.out.println(compression.name() + " " + compression.value() + " " + compression.source());

ConfigEntry retention = configs.get(TopicConfig.RETENTION_MS_CONFIG);
System.out.println(retention.name() + " " + retention.value() + " " + retention.source());

No código acima, criamos um ConfigResource informando que se trata de um tópico, cujo nome é "seila". Em seguida, o describeConfigs é chamado e as configurações são lidas e armazenadas no objeto configs. Por fim, lemos as configurações desejadas: compressão e retenção (respectivamente uma configuração de tópico e broker).

O tópico deste exemplo foi criado com esta linha de comando:

kafka-topics.sh --bootstrap-server localhost:9092 --create --replication-factor 1 --partitions 1 --config compression.type=gzip --topic seila

Portanto, a saída do código Java acima é algo assim:

compression.type gzip DYNAMIC_TOPIC_CONFIG

retention.ms 604800000 DEFAULT_CONFIG

Explicação: na criação do tópico foi definida a compressão gzip (como se vê na linha de comando acima e depois na primeira linha da saída do programa). Quanto ao valor DYNAMIC_TOPIC_CONFIG retornado pelo método compression.source(), ele indica que foi uma alteração manual / dinâmica e não uma configuração padrão. Já na segunda linha da saída (retenção), o valor retornado é o default (log.retention.hours=168) definido no server.properties do broker, pois 168 horas equivale a 604800000 milisegundos.

Testado com o kafka-clients 3.4.1 e Kafka 3.4.0.

Kafka: consultando tópicos com AdminClient

2023-09-07T15:44:00.000-07:00

Uma maneira de consultar informações sobre tópicos é usar o AdminClient (programaticamente). Aqui serão mostradas duas coisas: listar tópicos e descrever alguns detalhes deles.

A primeira coisa a se fazer é criar um objeto AdminClient e passar para ele pelo menos a indicação dos brokers do seu cluster (aqui passo apenas um endereço). Feito isso, você já pode chamar o listTopics, como no código abaixo, que exibe os nomes dos tópicos:

Properties properties = new Properties();
properties.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");

AdminClient admin = AdminClient.create(properties);

ListTopicsResult topics = admin.listTopics();
topics.names().get().forEach(System.out::println);

admin.close();

Outra coisa que é possível fazer é descrever um tópico e obter informações como: número de partições, id do tópico, nome do tópico, informação do líder e algumas outras.

A criação do AdminClient também é necessária aqui, mas é feita da mesma forma como foi feita acima.

Primeiro você chama o describeTopics passando uma lista de tópicos. Isto retornará um mapa e através dele é possível obter um TopicDescription. Mais abaixo você pode obter informações de partições.

DescribeTopicsResult result = admin.describeTopics(List.of("topic-20230724-aa"));
Map <String, KafkaFuture<TopicDescription>> map = result.topicNameValues();
KafkaFuture<TopicDescription> futureDesc = map.get("topic-20230724-aa");

TopicDescription desc = futureDesc.get();
System.out.println("Id: " + desc.topicId());
System.out.println("Nome: " + desc.name());
System.out.println("Interno? " + desc.isInternal());

List<TopicPartitionInfo> partitions = desc.partitions();
System.out.println("Partições: " + partitions.size());
for (TopicPartitionInfo info : partitions) {
    System.out.println("Partição: " + info.partition());
    System.out.println("Líder " + info.leader());
}

É possível fazer algo semelhante via linha de comando (só pra lembrar):

kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic topic-20230724-aa

Testado com o kafka-clients 3.4.1 e Kafka 3.4.0.

Kafka: partitioner customizado

2023-07-24T16:30:00.003-07:00

Se você precisa definir as partições de um tópico onde as mensagens serão gravadas, ou seja, se você precisa ter o controle disso, é preciso criar uma classe que implemente a interface Partitioner e no método partition implementar a sua lógica para determinar a partição.

public class MyPartitioner implements Partitioner {


    @Override
    public int partition(String topic, Object key, byte[] bytes,
                     Object value, byte[] bytes1, Cluster cluster) {

        String k = (String) key;

        if (topic.equals("topic-20230724-aa")) {
            if (k.equals("a")) {
                return 0;
            } else if (k.equals("b")) {
                return 1;
            } else if (k.equals("c")) {
                return 2;
            } else  {
                return 3;
            }
        } else {
            List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
            int numPartitions = partitions.size();
            return Math.abs(Utils.murmur2(bytes)) % (numPartitions - 1);
        }
    }

    // Outros métodos

}

No código acima, é aplicada uma regra para o tópico "topic-20230724-aa" e outra para os demais tópicos. No primeiro caso (if), as mensagens cuja chave é "a" sempre vão para a partição 0, "b" para partição 1, "c" para a 2 e o restante para 3. Isto é apenas um exemplo.

Na prática os critérios podem ser os mais variados e complexos.

Para informar que você vai usar este Partitioner customizado, é preciso passar uma propriedade informando o mesmo para o KafkaProducer:

kafkaProperties.put(
        ProducerConfig.PARTITIONER_CLASS_CONFIG,
        "partitioner.MyPartitioner");

KafkaProducer producer = new KafkaProducer<String, String>(kafkaProperties);

Testado com o Kafka 3.4.0, Java 11 e kafka-clients 3.4.1.

Kafka: mensagens sem chave (key)

2023-07-22T19:31:00.004-07:00

A chave/key é usada para determinar em qual partição a mensagem será escrita. Mensagens com a mesma chave são escritas na mesma partição.

No código abaixo, só o tópico e o valor da mensagem são informados:

ProducerRecord <String, String> record = 
        new ProducerRecord<>(
                "topic-20230722-c", 
                "test-" + new Random().nextInt(100));

Quando uma mensagem é enviada sem especificar a chave/key, a partição é escolhida usando round-robin.

Você pode conferir este comportamento, verificando a quantidade de mensagens no tópico (uma linha / comando para cada partição):

$ kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list localhost:9092 --topic topic-20230722-c --partitions 0

topic-20230722-c:0:10

$ kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list localhost:9092 --topic topic-20230722-c --partitions 1

topic-20230722-c:1:5

$ kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list localhost:9092 --topic topic-20230722-c --partitions 2

topic-20230722-c:2:10

Obs.: este outro post aqui aborda o comando acima.

Como se nota, as mensagens vão sendo distribuídas entre as partições, respectivamente 10, 5 e 10 mensagens.

Também é possível verificar as mensagens nas partições através do kafka-console-consumer (basta mudar o número da partição no final):

$ kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic topic-20230722-c --from-beginning --partition 0

Nota: se você rodar o consumer sem especificar a partição, todas as mensagens do tópico -- de todas as partições -- serão consumidas.

Kafka: Producer Interceptor

2023-06-24T11:05:00.007-07:00

Você pode criar um interceptor no contexto do Produtor Kafka. Para isso, basta implementar a interface ProducerInterceptor. Os principais métodos desta interface são:

onSend: este método é chamado antes da mensagem ser serializada, ou seja, antes dela ser enviada ao Kafka. Ele é útil para capturar informações da mensagem e também modificá-la (headers, tratar informações sensíveis, ...);
onAcknowledgement: é chamado quando o Kafka confirma o recebimento. É útil para obter informações da resposta.

Como se vê, é um recurso útil para monitoramento.

Segue uma implementação de exemplo:

public class MyProducerInterceptor
        implements ProducerInterceptor<String, String> {

    @Override
    public ProducerRecord<String, String> onSend(
            ProducerRecord<String, String> producerRecord) {

        System.out.println("Mensagem enviada, vou modificar");
        String valor = producerRecord.value();
        System.out.println("Valor original: " + valor);
        return new ProducerRecord<>(
                "teste-purge3",
                "a",
                valor + "SOMETHING");
    }

    @Override
    public void onAcknowledgement(RecordMetadata recordMetadata,
                                  Exception e) {

        System.out.println("Ack realizado");
        System.out.println(recordMetadata.offset());
    }

    @Override
    public void close() {
        System.out.println("Close");
    }

    @Override
    public void configure(Map<String, ?> map) {
        System.out.println("Configure");
        Set<String> keys = map.keySet();
        for (String k : keys) {
            System.out.println(k + ": " + map.get(k));
        }
    }

}

Além dos dois método citados acima, como pode ser visto no código a interface também conta com o método configure. Ele é o primeiro a ser executado e você tem acesso a toda configuração do produtor (parâmetros). Útil também para fazer algum setup do seu lado.

Para este interceptor funcionar, na configuração do seu produtor, utilize:

kafkaProperties.put(
        ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, 
        "xpto.interceptor.MyProducerInterceptor");

Kafka: buffer.memory VS batch.size

2023-06-21T16:56:00.002-07:00

Estes dois parâmetros dos produtores Kafka podem ser confundidos. Segue abaixo a diferença entre eles.

Quando enviamos mensagens (através do método send), é utilizado um buffer. O tamanho dele é controlado através do parâmetro buffer.memory. Em outras palavras, é um espaço relativo à mensagens que estão esperando para serem enviadas aos brokers. Se este buffer estiver cheio, o produtor bloqueia pelo tempo definido em outro parâmetro: max.block.ms. Se durante este tempo o buffer não ficar livre, uma exceção será lançada.

Já batch.size é usado num contexto diferente. As mensagens que são enviadas para a mesma partição são agrupadas no mesmo lote ou batch e ocupam um espaço. Este parâmetro trata deste espaço (em bytes). Atenção: configurar um batch muito pequeno provocará overhead, pois causará envios mais frequentes.

Kafka: produtor síncrono

2023-06-17T18:00:00.009-07:00

A primeira coisa ao criar um produtor que envia mensagens para o Kafka, é instanciar um KafkaProducer. Ele espera um Properties com algumas informações obrigatórias:

Lista de brokers Kafka (host e porta): não precisa ser a lista completa, pois o produtor consegue obter mais informações após se conectar, mas é recomendável que seja pelo menos dois;
Key serializer: as chaves das mensagens são serializadas, portanto você precisa especificar um serializador. Você pode criar o seu próprio (se necessário), bastando para isto implementar org.apache.kafka.common.serialization.Serializer. Há vários serializadores comuns, como por exemplo, StringSerializer (usado aqui);
Value serializer: a ideia é a mesma do anterior, mas agora com relação ao valor.

Properties kafkaProperties = new Properties();
kafkaProperties.put("bootstrap.servers", "localhost:9092");
kafkaProperties.put("key.serializer","org.apache.kafka.common.serialization.StringSerializer");
kafkaProperties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

KafkaProducer producer = new KafkaProducer<String, String>(kafkaProperties);

Feito isso, você pode criar uma mensagem / record (ProducerRecord).

ProducerRecord <String, String> record = new ProducerRecord<>("topic-xpto", "key", "value");

Em seguida, faz o envio deste record, através do método send.

Um Future será retornado e por meio dele você pode obter os metadados (RecordMetadata).

Future <RecordMetadata> future = producer.send(record);
RecordMetadata metadata = future.get();
System.out.println("Sucesso --->>> " + metadata.offset());

A biblioteca utilizada aqui foi:

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>3.4.1</version>
</dependency>

Kafka: produtor assíncrono (callback)

2023-06-16T20:39:00.007-07:00

Naturalmente as mensagens são produzidas no Kafka de modo assíncrono, pois ao fazer o envio é retornado um Future.

Aqui será mostrado como fazer o envio usando um callback.

Para isso você precisa implementar a interface org.apache.kafka.clients.producer.Callback. Ela pode ser encontrada kafka-clients:

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>3.4.1</version>
</dependency>

Basta implementar o método onCompletion:

public class MyFirstCallback implements Callback {

    @Override
    public void onCompletion(RecordMetadata recordMetadata, Exception e) {
        if (e == null) {
            System.out.println("Sucesso: " + recordMetadata.offset());
        } else {
            System.err.println("A casa caiu: " + e.getMessage());
        }
    }

}

Feito isso, basta instanciar este callback e passar como segundo parâmetro do método send do seu producer (um org.apache.kafka.clients.producer.KafkaProducer):

producer.send(record, new MyFirstCallback());

Também é possível fazer com Lambda:

producer.send(record, (r, e) -> {
    if (e == null) {
        System.out.println("Sucesso:: " + r.offset());
    } else {
        System.err.println("A casa caiu: " + e.getMessage());
    }
});

Obs.: se o Kafka retornar algum erro, o parâmetro do tipo Exception no onCompletion será diferente de nulo.

Kafka: número de mensagens num tópico

2023-06-10T19:41:00.003-07:00

Para saber a quantidade de mensagens que um tópico do Kafka já recebeu, utilize o comando:

kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list localhost:9092 --topic basic-teste --partitions 0

Segue uma breve explicação dele:

kafka-run-class.sh: está localizado na pasta bin do seu Kafka;
GetOffsetShell: é um utilitário relacionado a offSets;
broker-list: é a lista de brokers literalmente;
topic: o nome do tópico;
partitions: lista com os ids da partições que você deseja.

Isto irá retornar algo assim:

basic-teste:0:18

A saída acima tem três partes (separadas por dois pontos). A primeira parte é o tópico, a segunda é o id da partição e a terceira o offset. É justamente ele que nos diz quantas mensagens este tópico já recebeu (incluindo as mensagens que já foram deletadas por conta da configuração de retenção).

Podemos simplificar ainda mais, acrescentando o comando cut do Linux:

kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list localhost:9092 --topic basic-teste --partitions 0 | cut -d ':' -f 3

O cut permite manipular linhas. Acima estamos dizendo que o delimitador é : (dois pontos, com a opção -d) e que queremos apenas a última parte, a terceira parte, ou field (-f). Rodar este comando retorna: 18.

Observações:

O offset é um inteiro que é incrementado a cada mensagem produzida.
Para saber a quantidade corrente (atual, ainda não deletada) de mensagens num tópico, é preciso usar o ksqlDB, mas isso eu ainda preciso estudar melhor para escrever aqui.

Tuning Kafka: armazenamento dos logs segments (reduzir escritas)

2023-06-08T18:41:00.000-07:00

É recomendável que o ponto de montagem que armazena os log segments (mensagens) do Kafka, seja configurado com a opção noatime. Isto está relacionado com os atributos de timestamp do Linux. Resumidamente, o Linux tem um metadado relacionado a última vez em que um arquivo foi acessado (atime). Isso acaba gerando muitas atualizações, ou seja, muitas escritas no disco, pois as mensagens estão sendo lidas a todo momento.

Portanto, para diminuir o número de escritas, deve-se utilizar o noatime. Para entender melhor isso, e também como fazer na prática, veja este post aqui.

Podemos usar o noatime com segurança, pois o Kafka não utiliza essa informação.

Linux: performance e file timestamps

2023-06-08T16:26:00.004-07:00

Todos os arquivos no Linux possuem alguns timestamps. Eis alguns deles:

atime: é atualizado sempre que um arquivo é acessado (accessed);
mtime: é atualizado sempre que o conteúdo de um arquivo é modificado (modified);
ctime: é atualizado sempre que o conteúdo ou os atributos (alterar o nome, mudar permissões...) de um arquivo são alterados (changed).

Por conta de constantes atualizações -- pense, arquivos são acessados a todo instante -- deixar o atime habilitado causa uma queda de performance, pois ocorrem muitas escritas/atualizações no atime. Portanto, se você ou sua aplicação / servidor, não precisa do atime, desligue-o! Mesmo rodando com SSD.

Para desligar o atime, você precisa editar o /etc/fstab e acrescentar noatime na linha do ponto de montagem, como abaixo:

UUID=7b32eca9-e795-4506-8ece-a789a9f53a06 / ext4 errors=remount-ro,noatime 0 1

Depois de reiniciar a máquina, utilize este comando para ter certeza que o noatime foi habilitado:

cat /proc/mounts | grep " / "

A saída será:

/dev/nvme0n1p4 / ext4 rw,noatime,errors=remount-ro 0 0

Obs.: existe um meio termo para não ficar sem o atime: é o relatime. Com essa opção, o atime só é atualizado se ele for anterior ao mtime ou ctime atual (corrente), ou seja, o acesso é atualizado somente quando o arquivo sofre alguma modificação. Esse comportamento já é melhor que o original (quando o acesso é sempre atualizado) e oferece também compatibilidade. O relatime está disponível como default desde o Kernel 2.6.30. Mas se você quiser ser "radical" e precisar do extremo da performance, use noatime (neste caso precisa ter certeza de que seu ambiente não precisa do atime).

Para ter certeza que o relatime tá habilitado, utilize o comando:

cat /proc/mounts | grep " / "

No meu caso a saída foi (olha o relatime aí):

/dev/nvme0n1p4 / ext4 rw,relatime,errors=remount-ro 0 0

Kafka: retenção de mensagens por tempo

2023-05-29T19:41:00.002-07:00

Por default o Kafka retém mensagens por 168 horas, ou seja, 7 dias. Isto é definido através da propriedade log.retention.hours no arquivo server.properties do diretório config da instalação do Kafka.

Há outras duas opções: milissegundos (log.retention.ms) e minutos (log.retention.minutes). Recomenda-se a utilização de milissegundos, pois é a menor unidade e a menor sempre tem precedência sobre as outras (caso mais de uma seja especificada).

Testado no Kafka 3.4.0.

Tuning Kafka: Qual é o melhor Garbage Collector (GC) para o Kafka?

2023-05-22T18:19:00.004-07:00

Se você usa uma JVM 1.8 ou superior, o melhor é usar o G1GC (Garbage First Garbage Collector).

Mas por que? Resumidamente, o G1GC tem alta performance, suporta grandes volumes de memória e se ajusta automaticamente ao processamento que precisa fazer, ou seja, ele se adapta à circunstância (dinamismo). Suas pausas são consistentes, pois ele segmenta a memória em zonas menores, o que faz com que não percorra (varra / colete) toda a memória em cada pausa, logo executa em menor tempo. Ele também foca nas zonas onde há mais objetos que podem ser descartados (sem referência, que não estão sendo usados), logo libera espaço considerável.

Há dois parâmetros / opções úteis neste contexto:

MaxGCPauseMillis: indica o tempo aproximado em que o G1GC vai rodar (pode exceder). Seu valor default é 200 milisegundos. O algoritmo faz uma previsão de quantas zonas coletar / processar para atingir o tempo especificado aqui.
InitiatingHeapOccupancyPercent: indica qual o percentual da memória (new e old juntas) deverá estar em uso para que o G1GC inicie a execução. O default aqui é 45.

Em termos práticos, você faz essa configuração através desta variável:

export KAFKA_JVM_PERFORMANCE_OPTS="-server -Xmx4g -Xms4g -XX:UseG1GC -XX:MaxGCPauseMillis=50 -XX:InitiatingHeapOccupancyPercent=40 ..."

Kafka (standalone): subindo e testando

2023-05-06T18:11:00.006-07:00

No post anterior, mostrei como configurar minimamente o ZooKeeper como pré-requisito para rodar o Kafka standalone.

Agora o foco é o Kafka. A versão utilizada aqui é a 3.4.0.

Com as configurações padrão, a primeira coisa a fazer é o comando abaixo (para dar o start):

kafka-server-start.sh -daemon /SEU_DIRETORIO/kafka_2.13-3.4.0/config/server.properties

Agora para testar, faremos 4 coisas (tudo pela linha de comando):

Criar um tópico;
Pedir detalhes deste tópico;
Produzir mensagens;
Consumir mensagens.

Criação de um tópico chamado teste-simples:

kafka-topics.sh --bootstrap-server localhost:9092 --create --replication-factor 1 --partitions 1 --topic teste-simples

Para ver detalhes deste tópico:

kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic teste-simples

Para produzir mensagens:

kafka-console-producer.sh --bootstrap-server localhost:9092 --topic teste-simples

Para consumir mensagens:

kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic teste-simples --from-beginning

Kafka (standalone): subindo e testando o ZooKeeper

2023-05-06T15:34:00.011-07:00

O ZooKeeper é um pré-requisito para rodar o Kafka, pois ele armazena os metadados do(s) broker(s) (cluster).

Neste exemplo, estou utilizando o ZooKeeper 3.8.1.

A primeira coisa a se fazer é a configuração básica (standalone). Para isto, na pasta conf do ZooKeeper, crie um arquivo de configuração chamado zoo.cfg e coloque as linhas abaixo nele:

# The number of milliseconds of each tick
tickTime=2000
# The directory where the snapshot is stored
dataDir=/work/tools/kafka/zookeeper-data
# The port at which the clients will connect
clientPort=2181

Acima, as duas coisas mais importantes neste momento são: o diretório onde os dados serão armazenados e a porta para os clientes.

Feito isso, é hora de iniciar o ZooKeeper. Utilize este comando (localizado na basta bin do seu ZooKeeper):

zkServer.sh start

As linhas abaixo serão exibidas:

ZooKeeper JMX enabled by default
Using config: /work/tools/kafka/apache-zookeeper-3.8.1-bin/bin/../conf/zoo.cfg
Starting zookeeper ... STARTED

Para testar e ter certeza que o ZooKeeper está no ar, faça o telnet na porta configurada acima:

telnet localhost 2181

Quando conectar, digite: srvr. Serão exibidas algumas informações do servidor ZooKeeper:

Zookeeper version: 3.8.1-74db005175a4ec545697012f9069cb9dcc8cdda7, built on 2023-01-25 16:31 UTC
Latency min/avg/max: 0/0.0/0
Received: 1
Sent: 0
Connections: 1
Outstanding: 0
Zxid: 0x8d
Mode: standalone
Node count: 134

Pronto, ZooKeeper no ar!

SQL Server: ler objeto de um JSON (JSON_QUERY)

2023-03-08T19:00:00.000-08:00

Este post pode ser considerado uma continuação deste outro. Nele eu mostrei como ler um atributo simples (scalar value). Agora é um pouco diferente, o atributo que será lido contém um objeto, outro JSON.

Basicamente, ao invés de usar JSON_VALUE, basta usar JSON_QUERY:

select json_query(campo, '$.education') from tabela

O valor retornado será: {"name":"xpto", "year":"2005"}.

Testado no SQL Server 2017.

SQL Server: query que gera JSON

2023-03-02T02:39:00.001-08:00

Para fazer sua query gerar um JSON, utilize o for json path.

select top 1 id as id, name as nome
from tabela
for json path

A saída conterá um JSON assim: [{"id":1, "nome":"Fiodor"}]

Testado no SQL Server 2017.