Menghubungkan ke klaster Anda dengan kredensyal masuk

Setelah Anda membuat rahasia dan mengaitkannya dengan cluster Anda, Anda dapat menghubungkan klien Anda ke cluster. Prosedur berikut menunjukkan cara menghubungkan klien ke cluster yang menggunakan SASL/SCRAM otentikasi. Ini juga menunjukkan bagaimana memproduksi dan mengkonsumsi dari topik contoh.

Menghubungkan klien ke cluster menggunakan SASL/SCRAM otentikasi

1. Jalankan perintah berikut pada mesin yang telah AWS CLI diinstal. Ganti clusterARN dengan ARN cluster Anda.

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

   Dari hasil JSON dari perintah ini, simpan nilai yang terkait dengan string bernamaBootstrapBrokerStringSaslScram. Anda akan menggunakan nilai ini di langkah selanjutnya.

2. Di mesin klien Anda, buat file konfigurasi JAAS yang berisi kredensyal pengguna yang disimpan dalam rahasia Anda. Misalnya, untuk pengguna alice, buat file yang dipanggil users_jaas.conf dengan konten berikut.

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

3. Gunakan perintah berikut untuk mengekspor file konfigurasi JAAS Anda sebagai parameter KAFKA_OPTS lingkungan.

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

4. Buat file bernama kafka.client.truststore.jks dalam /tmp direktori.

5. (Opsional) Gunakan perintah berikut untuk menyalin file penyimpanan kunci JDK dari cacerts folder JVM Anda ke kafka.client.truststore.jks file yang Anda buat pada langkah sebelumnya. Ganti JDKFolder dengan nama folder JDK pada instance Anda. Misalnya, folder JDK Anda mungkin diberi nama. 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. Di bin direktori instalasi Apache Kafka Anda, buat file properti klien yang disebut client_sasl.properties dengan konten berikut. File ini mendefinisikan mekanisme dan protokol SASL.

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

7. Untuk membuat contoh topik, jalankan perintah berikut. Ganti BootstrapBrokerStringSaslScram dengan string broker bootstrap yang Anda peroleh di langkah 1 topik ini.

   <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. Untuk menghasilkan contoh topik yang Anda buat, jalankan perintah berikut di mesin klien Anda. Ganti BootstrapBrokerStringSaslScram dengan string broker bootstrap yang Anda ambil di langkah 1 topik ini.

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

9. Untuk mengkonsumsi dari topik yang Anda buat, jalankan perintah berikut di mesin klien Anda. Ganti BootstrapBrokerStringSaslScram dengan string broker bootstrap yang Anda peroleh di langkah 1 topik ini.

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

Memecahkan masalah koneksi

Saat menjalankan perintah klien Kafka, Anda mungkin mengalami kesalahan memori heap Java, terutama saat bekerja dengan topik atau kumpulan data besar. Kesalahan ini terjadi karena alat Kafka berjalan sebagai aplikasi Java dengan pengaturan memori default yang mungkin tidak cukup untuk beban kerja Anda.

Untuk mengatasi Out of Memory Java Heap kesalahan, Anda dapat meningkatkan ukuran heap Java dengan memodifikasi variabel KAFKA_OPTS lingkungan untuk menyertakan pengaturan memori.

Contoh berikut menetapkan ukuran heap maksimum untuk 1GB ()-Xmx1G. Anda dapat menyesuaikan nilai ini berdasarkan memori dan persyaratan sistem yang tersedia.

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

Untuk mengkonsumsi topik besar, pertimbangkan untuk menggunakan parameter berbasis waktu atau berbasis offset alih-alih membatasi penggunaan --from-beginning memori:

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