Como estabelecer conexão com o seu cluster usando credenciais de acesso

Após criar um segredo e associá-lo ao cluster, você poderá conectar o cliente ao cluster. O procedimento a seguir demonstra como conectar um cliente a um cluster que usa SASL/SCRAM autenticação. Também mostra como produzir e consumir a partir de um tópico de exemplo.

Conectando um cliente ao cluster usando SASL/SCRAM autenticação

1. Execute o comando a seguir em uma máquina que tenha sido AWS CLI instalada. clusterARNSubstitua pelo ARN do seu cluster.

   aws kafka get-bootstrap-brokers --cluster-arn clusterARN

   No resultado JSON desse comando, salve o valor associado à string chamadaBootstrapBrokerStringSaslScram. Você usará esse valor em etapas posteriores.

2. Em sua máquina cliente, crie um arquivo de configuração JAAS contendo as credenciais de usuário armazenadas em seu segredo. Por exemplo, para o usuário alice, crie um arquivo chamado users_jaas.conf com o conteúdo a seguir.

   KafkaClient {
      org.apache.kafka.common.security.scram.ScramLoginModule required
      username="alice"
      password="alice-secret";
   };

3. Use o comando a seguir para exportar seu arquivo de configuração JAAS como um parâmetro de ambiente KAFKA_OPTS.

   export KAFKA_OPTS=-Djava.security.auth.login.config=<path-to-jaas-file>/users_jaas.conf

4. Crie um arquivo chamado kafka.client.truststore.jks em um diretório /tmp.

5. (Opcional) Use o comando a seguir para copiar o arquivo de armazenamento de chaves do JDK da sua cacerts pasta JVM para o kafka.client.truststore.jks arquivo que você criou na etapa anterior. JDKFolderSubstitua pelo nome da pasta JDK na sua instância. Por exemplo, sua pasta do JDK pode ter o nome java-1.8.0-openjdk-1.8.0.201.b09-0.amzn2.x86_64.

   cp /usr/lib/jvm/JDKFolder/lib/security/cacerts /tmp/kafka.client.truststore.jks

6. No diretório bin da instalação do Apache Kafka, crie um arquivo de propriedades do cliente chamado client_sasl.properties com o conteúdo a seguir. Esse arquivo define o mecanismo e o protocolo SASL.

   security.protocol=SASL_SSL
   sasl.mechanism=SCRAM-SHA-512

7. Para criar um tópico de exemplo, execute o comando a seguir. BootstrapBrokerStringSaslScramSubstitua pela string do bootstrap broker que você obteve na etapa 1 deste tópico.

   <path-to-your-kafka-installation>/bin/kafka-topics.sh --create --bootstrap-server BootstrapBrokerStringSaslScram --command-config <path-to-client-properties>/client_sasl.properties --replication-factor 3 --partitions 1 --topic ExampleTopicName

8. Para produzir o tópico de exemplo que você criou, execute o seguinte comando em sua máquina cliente. BootstrapBrokerStringSaslScramSubstitua pela string do bootstrap broker que você recuperou na etapa 1 deste tópico.

   <path-to-your-kafka-installation>/bin/kafka-console-producer.sh --broker-list BootstrapBrokerStringSaslScram --topic ExampleTopicName --producer.config client_sasl.properties

9. Para consumir do tópico que você criou, execute o comando a seguir em sua máquina cliente. BootstrapBrokerStringSaslScramSubstitua pela string do bootstrap broker que você obteve na etapa 1 deste tópico.

   <path-to-your-kafka-installation>/bin/kafka-console-consumer.sh --bootstrap-server BootstrapBrokerStringSaslScram --topic ExampleTopicName --from-beginning --consumer.config client_sasl.properties

Solução de problemas de conexão

Ao executar comandos do cliente Kafka, você pode encontrar erros de memória de pilha Java, especialmente ao trabalhar com grandes tópicos ou conjuntos de dados. Esses erros ocorrem porque as ferramentas do Kafka são executadas como aplicativos Java com configurações de memória padrão que podem ser insuficientes para sua carga de trabalho.

Para resolver Out of Memory Java Heap erros, você pode aumentar o tamanho do heap Java modificando a variável de KAFKA_OPTS ambiente para incluir configurações de memória.

O exemplo a seguir define o tamanho máximo da pilha como 1 GB ()-Xmx1G. Você pode ajustar esse valor com base na memória e nos requisitos disponíveis do sistema.

export KAFKA_OPTS="-Djava.security.auth.login.config=<path-to-jaas-file>/users_jaas.conf -Xmx1G"

Para consumir tópicos extensos, considere usar parâmetros baseados em tempo ou em offset em vez de limitar o uso --from-beginning de memória:

<path-to-your-kafka-installation>/bin/kafka-console-consumer.sh --bootstrap-server BootstrapBrokerStringSaslScram --topic ExampleTopicName --max-messages 1000 --consumer.config client_sasl.properties