As traduções são geradas por tradução automática. Em caso de conflito entre o conteúdo da tradução e da versão original em inglês, a versão em inglês prevalecerá.

Usando o Kafka Streams com o MSK Serverless

O Kafka Streams é compatível com transformações com e sem estado. Transformações com estado, como contar, agregar ou unir, usam operadores que armazenam seu estado em tópicos internos do Kafka. Além disso, algumas transformações sem estado, como groupBy ou repartição, armazenam resultados em tópicos internos do Kafka. Por padrão, o Kafka Streams nomeia esses tópicos internos com base no operador correspondente. Se esses tópicos não existirem, o Kafka Streams criará os tópicos internos do Kafka. Para criar os tópicos internos, o Kafka Streams codifica a configuração segment.bytes e a define em 50 MB. O MSK Serverless protege algumas configurações de tópicos, incluindo segment.size durante a criação do tópico. Portanto, um aplicativo Kafka Streams com transformações com estado não consegue criar os tópicos internos usando o MSK Serverless.

Para executar esses aplicativos do Kafka Streams no MSK Serverless, você mesmo deve criar os tópicos internos. Para isso, primeiro identifique e nomeie os operadores do Kafka Streams que exigem tópicos. Em seguida, crie os tópicos internos correspondentes do Kafka.

Criando um aplicativo Kafka Streams usando o MSK Serverless

Se seu aplicativo Kafka Streams estiver application.id configurado comomsk-streams-processing, você poderá criar um aplicativo Kafka Streams usando o MSK Serverless. Para fazer isso, use o operador count(), que requer um tópico interno com o nome. Por exemplo, .msk-streams-processing-count-store

Para criar uma aplicação Kafka Streams, faça o seguinte:

Identifique e nomeie os operadores

Criação dos tópicos internos

O Kafka Streams coloca prefixos application.id em nomes de tópicos internos, em que application.id é definido pelo usuário. Por exemplo, .application.id-internal_topic_name Os tópicos internos são normais do Kafka, e você pode criar os tópicos usando as informações disponíveis em Criar um tópico do Apache Kafka ou AdminClient da API do Kafka.

Dependendo do seu caso de uso, você pode usar as políticas padrão de limpeza e retenção do Kafka Streams ou personalizar os valores. Você os define em cleanup.policy e retention.ms.

O exemplo a seguir cria os tópicos com a API AdminClient e define o application.id como msk-streams-processing.

try (AdminClient client = AdminClient.create(configs.kafkaProps())) {
    Collection<NewTopic> topics = new HashSet<>();
    topics.add(new NewTopic("msk-streams-processing-count-store", 3, (short) 3));
    client.createTopics(topics);
}

Depois que os tópicos forem criados no cluster, sua aplicação Kafka Streams poderá usar o tópico msk-streams-processing-count-store na operação count().

(Opcional) Verificar o nome do tópico

Você pode usar o descritor de topologia para descrever a topologia do seu stream e visualizar os nomes dos tópicos internos. O exemplo a seguir mostra como executar o descritor da topologia.

final StreamsBuilder builder = new StreamsBuilder();
Topology topology = builder.build();
System.out.println(topology.describe());

A saída a seguir mostra a topologia do fluxo para o exemplo anterior.

Topology Description:
Topologies:
   Sub-topology: 0
    Source: KSTREAM-SOURCE-0000000000 (topics: [input_topic])
      --> KSTREAM-AGGREGATE-0000000001
    Processor: KSTREAM-AGGREGATE-0000000001 (stores: [count-store])
      --> KTABLE-TOSTREAM-0000000002
      <-- KSTREAM-SOURCE-0000000000
    Processor: KTABLE-TOSTREAM-0000000002 (stores: [])
      --> KSTREAM-SINK-0000000003
      <-- KSTREAM-AGGREGATE-0000000001
    Sink: KSTREAM-SINK-0000000003 (topic: output_topic)
      <-- KTABLE-TOSTREAM-0000000002

Para saber como usar o descritor de topologia, consulte Nomeando operadores em uma aplicação DSL do Kafka Streams na documentação do Kafka Streams.

Exemplos de nomenclatura de operadores

Esta seção fornece exemplos de operações de nomenclatura.

Exemplo de operador de nomenclatura para group ByKey ()

groupByKey() -> groupByKey(Grouped.as("kafka-stream-groupby"))

Exemplo de operador de nomenclatura para contagem normal()

normal count() -> .count(Materialized.<String, Long, KeyValueStore<Bytes, byte[]>>as("kafka-streams-window")  // descriptive name for the store
    .withKeySerde(Serdes.String())
    .withValueSerde(Serdes.Long()))

Exemplo de operador de nomenclatura para contagem em janela()

windowed count() -> .count(Materialized.<String, Long, WindowStore<Bytes, byte[]>>as("kafka-streams-window")  // descriptive name for the store
    .withKeySerde(Serdes.String())
    .withValueSerde(Serdes.Long()))

Exemplo de operador de nomenclatura para suprimido em janela()

windowed suppressed() -> 
Suppressed<Windowed> suppressed = Suppressed
        .untilWindowCloses(Suppressed.BufferConfig.unbounded())
        .withName("kafka-suppressed");
    .suppress(suppressed)