Informações sobre KCL 1.x e 2.x

KCL 1.x

Para ter mais informações, consulte Desenvolver aplicações de consumo da KCL 1.x

KCL 2.x

Para ter mais informações, consulte Desenvolver aplicações de consumo da KCL 2.x

Aplicação de consumo da KCL: uma aplicação personalizada que usa a KCL e é projetada para ler e processar registros de fluxos de dados.

Instância de aplicação de consumo: as aplicações de cliente da KCL normalmente são distribuídas, com uma ou mais instâncias executadas simultaneamente para coordenar falhas e balancear dinamicamente a carga de processamento dos registros de dados.

Operador: uma classe de alto nível que uma instância de aplicação de consumo da KCL usa para começar a processar dados.

O operador inicializa e supervisiona várias tarefas, incluindo a sincronização de informações de fragmentos e concessões, o monitoramento de atribuições de fragmentos e o processamento dos dados dos fragmentos. Um trabalhador fornece à KCL as informações de configuração do aplicativo consumidor, como o nome do fluxo de dados cujos registros de dados esse aplicativo consumidor KCL processará e AWS as credenciais necessárias para acessar esse fluxo de dados. O operador também inicia a instância específica da aplicação de consumo da KCL para entregar registros de dados do fluxo de dados aos processadores de registros.

Concessão: dado que define a ligação entre um operador e um fragmento. As aplicações de cliente distribuídas da KCL usam concessões para particionar o processamento de registros de dados em uma frota de operadores. A qualquer momento, cada fragmento de registros de dados é associado a um determinado operador por uma concessão identificada pela variável leaseKey.

Por padrão, um trabalhador pode manter um ou mais arrendamentos (sujeito ao valor da variável maxLeasesForWorker) ao mesmo tempo.

Por exemplo, se houver uma instância da aplicação de consumo A com o operador A que está processando um fluxo de dados com quatro fragmentos, o operador A poderá reter as concessões aos fragmentos 1, 2, 3 e 4 ao mesmo tempo. Mas, se houver duas instâncias de aplicações de consumo A e B com os operadores A e B, e essas instâncias estiverem processando um fluxo de dados com quatro fragmentos, o operador A e o operador B não poderão reter a concessão ao fragmento 1 ao mesmo tempo. Um operador retém a concessão a um fragmento específico até estar pronto para parar de processar os registros de dados do fragmento ou até que uma falha ocorra. Quando um operador libera a concessão, outro operador a assume e a retém.

Para obter mais informações (esses são os repositórios Java KCL), consulte https://github.com/awslabs/amazon-kinesis-client/blob/v1.x/src/main/java/com/amazonaws/services/kinesis/leases/impl/Lease.java para KCL 1.x e https://github.com/awslabs/amazon-kinesis-client/blob/master/amazon-kinesis-client/src/main/java/software/amazon/kinesis/leases/Lease.java para KCL 2.x.

Tabela de concessões: uma tabela exclusiva do Amazon DynamoDB usada para monitorar os fragmentos em um fluxo de dados do KDS vinculados a uma concessão e sendo processados pelos operadores da aplicação de consumo da KCL. A tabela de concessões precisa permanecer sincronizada (em um operador e entre todos os operadores) com as informações mais recentes do fragmento do fluxo de dados enquanto a aplicação de consumo da KCL está em execução. Para obter mais informações, consulte Usar uma tabela de concessões para monitorar os fragmentos processados pela aplicação de consumo da KCL.

Processador de registros: a lógica que define como a aplicação de consumo da KCL processa os dados obtidos dos fluxos de dados. Em runtime, uma instância da aplicação de consumo da KCL inicia um operador, que, por sua vez, inicia um processador de registros para cada fragmento cuja concessão retém.

checkpoint: número de sequência do ponto de verificação mais recente do fragmento. Esse valor é exclusivo entre todos os fragmentos no fluxo de dados.

checkpointSubSequenceNúmero: ao usar o recurso de agregação da Kinesis Producer Library, essa é uma extensão do ponto de verificação que rastreia registros individuais de usuários dentro do registro do Kinesis.

leaseCounter: usado para versionamento de concessão, para que os operadores possam detectar se a própria concessão foi assumida por outro operador.

leaseKey: um identificador exclusivo para uma concessão. Cada concessão é específica a um fragmento no fluxo de dados e é retida por um operador por vez.

leaseOwner: o operador que está retendo essa concessão.

ownerSwitchesSincePonto de verificação: Quantas vezes esse contrato mudou de trabalhadores desde a última vez que um posto de controle foi escrito.

parentShardId: usado para garantir que o fragmento principal seja totalmente processado antes do início do processamento nos fragmentos secundários. Isso garante que os registros sejam processados na mesma ordem em que foram colocados no fluxo.

hashrange: usado pelo PeriodicShardSyncManager para executar sincronizações periódicas a fim de encontrar fragmentos ausentes na tabela de concessões e criar concessões para eles, se necessário.

childshards: usado por LeaseCleanupManager para revisar o status de processamento do fragmento filho e decidir se o fragmento pai pode ser excluído da tabela de concessões.

shardID: o ID do fragmento.

stream name o identificador do fluxo de dados no formato account-id:StreamName:streamCreationTimestamp.

Busca todos os fragmentos de dados do fluxo sendo processado

Busca todas as concessões do fragmento da tabela de concessões

Filtra cada fragmento aberto sem uma concessão na tabela de concessões

Itera em todos os fragmentos abertos encontrados e, para cada fragmento aberto sem pai aberto:

Na inicialização da aplicação, se a tabela de concessões estiver vazia, a KCL utilizará a opção de filtragem da API ListShard (o parâmetro de solicitação ShardFilter opcional) para recuperar e criar concessões somente para um instantâneo dos fragmentos abertos no momento especificado pelo parâmetro ShardFilter. O parâmetro ShardFilter permite filtrar a resposta da API ListShards. A única propriedade obrigatória do parâmetro ShardFilter é Type. A KCL usa a propriedade de filtro Type e os seguintes valores válidos para identificar e retornar um instantâneo dos fragmentos abertos que podem exigir novas concessões:

ShardFilter é usado ao criar uma concessão em uma tabela de concessões vazia para inicializar concessões para um instantâneo dos fragmentos especificados em RetrievalConfig#initialPositionInStreamExtended.

Para obter mais informações sobre o ShardFilter, consulte https://docs.aws.amazon.com/kinesis/latest/APIReference/API_ShardFilter.html.

Em vez de todos os trabalhadores realizarem a lease/shard synchronization to keep the lease table up to date with the latest shards in the data stream, a single elected worker leader performs the lease/shard sincronização.

O KCL 2.3 usa o parâmetro de ChildShards retorno do GetRecords e do SubscribeToShard APIs para realizar a sincronização de arrendamento/fragmento que acontece em SHARD_END para fragmentos fechados, permitindo que um trabalhador da KCL crie concessões somente para os fragmentos secundários do fragmento que ele concluiu o processamento. Em aplicações de consumo com throughput compartilhada, a otimização da sincronização de concessão/fragmento usa o parâmetro ChildShards da API GetRecords. Em aplicações de consumo com throughput dedicada (distribuição avançada), a otimização da sincronização de concessão/fragmento usa o parâmetro ChildShards da API SubscribeToShard. Para ter mais informações, consulte GetRecords, SubscribeToShards e ChildShard.

Com as mudanças acima, o comportamento da KCL está passando de um modelo no qual todos os operadores obtêm informações de todos os fragmentos existentes para um modelo em que os operadores só obtêm informações dos filhos do fragmento que possui. Portanto, além da sincronização que ocorre durante a inicialização da aplicação de consumo e nos eventos de refragmentação, a KCL agora também realiza varreduras periódicas adicionais de fragmento/concessão para identificar possíveis falhas na tabela de concessões (em outras palavras, para identificar todos os novos fragmentos), confirmando que o intervalo de hash completo do fluxo de dados está sendo processado e criar concessões para eles, se necessário. PeriodicShardSyncManager é o componente responsável pela execução periódica de varreduras de concessão/fragmento.

Para obter mais informações sobre o PeriodicShardSyncManager KCL 2.3, consulte https://github.com/awslabs/amazon-kinesis-client/blob/master/amazon-kinesis-client/src/main/java/software/amazon/kinesis/leases/LeaseManagementConfig.java #L201 -L213.

A KCL 2.3 tem novas opções disponíveis para configuração de PeriodicShardSyncManager em LeaseManagementConfig:

Agora, novas CloudWatch métricas também são emitidas para monitorar a integridade doPeriodicShardSyncManager. Para obter mais informações, consulte PeriodicShardSyncManager.

Inclui uma otimização de HierarchicalShardSyncer para criar apenas concessões em uma camada de fragmentos.

Na inicialização da aplicação, se a tabela de concessões estiver vazia, a KCL utilizará a opção de filtragem da API ListShard (o parâmetro de solicitação ShardFilter opcional) para recuperar e criar concessões somente para um instantâneo dos fragmentos abertos no momento especificado pelo parâmetro ShardFilter. O parâmetro ShardFilter permite filtrar a resposta da API ListShards. A única propriedade obrigatória do parâmetro ShardFilter é Type. A KCL usa a propriedade de filtro Type e os seguintes valores válidos para identificar e retornar um instantâneo dos fragmentos abertos que podem exigir novas concessões:

ShardFilter é usado ao criar uma concessão em uma tabela de concessões vazia para inicializar concessões para um instantâneo dos fragmentos especificados em KinesisClientLibConfiguration#initialPositionInStreamExtended.

Para obter mais informações sobre o ShardFilter, consulte https://docs.aws.amazon.com/kinesis/latest/APIReference/API_ShardFilter.html.

Em vez de todos os trabalhadores realizarem a lease/shard synchronization to keep the lease table up to date with the latest shards in the data stream, a single elected worker leader performs the lease/shard sincronização.

O KCL 1.14 usa o parâmetro de ChildShards retorno do GetRecords e do SubscribeToShard APIs para realizar a sincronização de arrendamento/fragmento que acontece em SHARD_END para fragmentos fechados, permitindo que um trabalhador da KCL crie concessões somente para os fragmentos secundários do fragmento que ele concluiu o processamento. Para ter mais informações, consulte GetRecords e ChildShard.

Com as mudanças acima, o comportamento da KCL está passando de um modelo no qual todos os operadores obtêm informações de todos os fragmentos existentes para um modelo em que os operadores só obtêm informações dos filhos do fragmento que possui. Portanto, além da sincronização que ocorre durante a inicialização da aplicação de consumo e nos eventos de refragmentação, a KCL agora também realiza varreduras periódicas adicionais de fragmento/concessão para identificar possíveis falhas na tabela de concessões (em outras palavras, para identificar todos os novos fragmentos), confirmando que o intervalo de hash completo do fluxo de dados está sendo processado e criar concessões para eles, se necessário. PeriodicShardSyncManager é o componente responsável pela execução periódica de varreduras de concessão/fragmento.

Quando KinesisClientLibConfiguration#shardSyncStrategyType é definido como ShardSyncStrategyType.SHARD_END, PeriodicShardSync leasesRecoveryAuditorInconsistencyConfidenceThreshold é usado para determinar o limite do número de varreduras consecutivas contendo lacunas na tabela de concessões após o qual é necessário impor uma sincronização de fragmentos. Quando KinesisClientLibConfiguration#shardSyncStrategyType é definido como ShardSyncStrategyType.PERIODIC, leasesRecoveryAuditorInconsistencyConfidenceThreshold é ignorado.

Para obter mais informações sobre o PeriodicShardSyncManager KCL 1.14, consulte https://github.com/awslabs/amazon-kinesis-client/blob/v1.x/src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/KinesisClientLibConfiguration.java #L987 -L999.

A KCL 1.14 tem uma nova opção disponível para configuração de PeriodicShardSyncManager em LeaseManagementConfig:

Agora, novas CloudWatch métricas também são emitidas para monitorar a integridade doPeriodicShardSyncManager. Para obter mais informações, consulte PeriodicShardSyncManager.

A KCL 1.14 agora também oferece suporte à limpeza adiada de concessões. As concessões são excluídas de forma assíncrona por LeaseCleanupManager ao chegar ao SHARD_END quando um fragmento ultrapassar o período de retenção do fluxo de dados ou quando for fechado por uma operação de refragmentação.

Novas opções disponíveis para configuração de LeaseCleanupManager:

Inclui uma otimização de KinesisShardSyncer para criar apenas concessões em uma camada de fragmentos.

MultistreamTracker interface

Para criar um aplicativo de consumidor que possa processar vários fluxos ao mesmo tempo, você deve implementar uma nova interface chamada MultistreamTracker. Essa interface inclui o método streamConfigList, que retorna a lista de fluxos de dados, e suas configurações, a serem processados pela aplicação de consumo da KCL. Observe que os fluxos de dados sendo processados podem ser alterados durante o runtime da aplicação de consumo. streamConfigList é chamado periodicamente pela KCL para obter informações das mudanças nos fluxos de dados a serem processados.

O streamConfigList método preenche a StreamConfiglista.

package software.amazon.kinesis.common;

import lombok.Data;
import lombok.experimental.Accessors;

@Data
@Accessors(fluent = true)
public class StreamConfig {
    private final StreamIdentifier streamIdentifier;
    private final InitialPositionInStreamExtended initialPositionInStreamExtended;
    private String consumerArn;
}
                

Observe que os campos StreamIdentifier e InitialPositionInStreamExtended são obrigatórios, enquanto consumerArn é opcional. Só é necessário fornecer consumerArn se a KCL 2.x estiver sendo usada para implementar uma aplicação de consumo de distribuição avançada.

Para obter mais informações sobreStreamIdentifier, consulte https://github.com/awslabs/amazon-kinesis-client/blob/v2.5.8/amazon-kinesis-client/src/main/java/software/amazon/kinesis/common/StreamIdentifier.java #L129. Para criar um StreamIdentifier, recomenda-se a criação de uma instância multifluxo a partir do streamArn e do streamCreationEpoch que esteja disponível na v2.5.0 e versões posteriores. Na KCL v2.3 e v2.4, que não oferecem suporte ao streamArm, crie uma instância multifluxo usando o formato account-id:StreamName:streamCreationTimestamp. Esse formato será descontinuado e não terá mais suporte a partir da próxima versão principal.

MultistreamTracker também inclui uma estratégia para excluir concessões de fluxos antigos na tabela de concessões (formerStreamsLeasesDeletionStrategy). Observe que a estratégia NÃO PODE ser alterada durante o runtime da aplicação de consumo. Para obter mais informações, consulte https://github.com/awslabs/amazon-kinesis-client/blob/0c5042dadf794fe988438436252a5a8fe70b6b0b//amazon-kinesis-client.java src/main/java/software/amazon/kinesis/processor/FormerStreamsLeasesDeletionStrategy

ConfigsBuilderé uma classe de todo o aplicativo que você pode usar para especificar todas as configurações do KCL 2.x a serem usadas ao criar seu aplicativo consumidor KCL. ConfigsBuildera classe agora tem suporte para a MultistreamTracker interface. Você pode inicializar ConfigsBuilder com o nome do único fluxo de dados do qual consumir registros:

 /**
     * Constructor to initialize ConfigsBuilder with StreamName
     * @param streamName
     * @param applicationName
     * @param kinesisClient
     * @param dynamoDBClient
     * @param cloudWatchClient
     * @param workerIdentifier
     * @param shardRecordProcessorFactory
     */
    public ConfigsBuilder(@NonNull String streamName, @NonNull String applicationName,
            @NonNull KinesisAsyncClient kinesisClient, @NonNull DynamoDbAsyncClient dynamoDBClient,
            @NonNull CloudWatchAsyncClient cloudWatchClient, @NonNull String workerIdentifier,
            @NonNull ShardRecordProcessorFactory shardRecordProcessorFactory) {
        this.appStreamTracker = Either.right(streamName);
        this.applicationName = applicationName;
        this.kinesisClient = kinesisClient;
        this.dynamoDBClient = dynamoDBClient;
        this.cloudWatchClient = cloudWatchClient;
        this.workerIdentifier = workerIdentifier;
        this.shardRecordProcessorFactory = shardRecordProcessorFactory;
    }

Ou você pode inicializar ConfigsBuilder com MultiStreamTracker se quiser implementar um aplicativo consumidor KCL que processe vários fluxos ao mesmo tempo.

* Constructor to initialize ConfigsBuilder with MultiStreamTracker
     * @param multiStreamTracker
     * @param applicationName
     * @param kinesisClient
     * @param dynamoDBClient
     * @param cloudWatchClient
     * @param workerIdentifier
     * @param shardRecordProcessorFactory
     */
    public ConfigsBuilder(@NonNull MultiStreamTracker multiStreamTracker, @NonNull String applicationName,
            @NonNull KinesisAsyncClient kinesisClient, @NonNull DynamoDbAsyncClient dynamoDBClient,
            @NonNull CloudWatchAsyncClient cloudWatchClient, @NonNull String workerIdentifier,
            @NonNull ShardRecordProcessorFactory shardRecordProcessorFactory) {
        this.appStreamTracker = Either.left(multiStreamTracker);
        this.applicationName = applicationName;
        this.kinesisClient = kinesisClient;
        this.dynamoDBClient = dynamoDBClient;
        this.cloudWatchClient = cloudWatchClient;
        this.workerIdentifier = workerIdentifier;
        this.shardRecordProcessorFactory = shardRecordProcessorFactory;
    }                    
                

Com o suporte multifluxo implementado na aplicação de consumo da KCL, cada linha da tabela de concessões da aplicação contém o ID do fragmento e o nome do fluxo que a aplicação processa.

Quando o suporte multifluxo para sua aplicação de consumo da KCL é implementado, leaseKey assume a estrutura account-id:StreamName:streamCreationTimestamp:ShardId. Por exemplo, 111111111:multiStreamTest-1:12345:shardId-000000000336.