本地kafak集群需要讀取aws-kafka的topic數(shù)據(jù)，可以參考aws官網(wǎng)指引速梗，完成數(shù)據(jù)接入肮塞。

下面摘取官網(wǎng)說(shuō)明：
可以使用登錄憑證來(lái)控制對(duì) Amazon MSK 集群的訪問(wèn)權(quán)限，這些憑證使用存儲(chǔ)和保護(hù) AWS Secrets Manager姻锁。將用戶憑證存儲(chǔ)在 Secrets Manager 中可以減少集群身份驗(yàn)證的開(kāi)銷(xiāo)枕赵，例如審計(jì)、更新和輪換憑證位隶。Secrets Manager 還能夠跨集群共享用戶憑證拷窜。

工作方式

Amazon 的登錄憑證身份驗(yàn)證MSK使用SASL/SCRAM（簡(jiǎn)單身份驗(yàn)證和安全層/Salted Challenge 響應(yīng)機(jī)制）身份驗(yàn)證。要為群集設(shè)置登錄憑據(jù)身份驗(yàn)證，請(qǐng)?jiān)谥袆?chuàng)建一個(gè) Secret 資源 并將登錄憑據(jù)與該密鑰相關(guān)聯(lián)篮昧。
SCRAM使用安全的哈希算法赋荆，并且不在客戶端和服務(wù)器之間傳輸純文本登錄憑證。

當(dāng)您為集群設(shè)置SASL/SCRAM身份驗(yàn)證時(shí)恋谭，Amazon MSK 會(huì)為客戶端和代理之間的所有流量啟用TLS加密糠睡。

為 Amazon MSK 集群設(shè)置SASL/SCRAM身份驗(yàn)證

在為 Amazon MSK 集群創(chuàng)建密鑰時(shí)挽鞠，請(qǐng)注意以下要求：
（1）為密鑰類(lèi)型選擇其他類(lèi)型的密API鑰（例如密鑰）疚颊。
（2）機(jī)密名稱(chēng)必須以前綴 Amazon MSK _ 開(kāi)頭。
（3）必須使用現(xiàn)有的自定義設(shè)置 AWS KMS 鍵入或創(chuàng)建新的自定義 AWS KMS 你的秘密的密鑰信认。Secrets Manager 使用默認(rèn)值 AWS KMS 默認(rèn)情況下材义，密鑰是密鑰。
使用默認(rèn)值創(chuàng)建的密鑰 AWS KMS 密鑰不能用于 Amazon MSK 集群嫁赏。
（4）您的登錄憑證數(shù)據(jù)必須采用以下格式其掂，才能使用明文選項(xiàng)輸入鍵值對(duì)。

{
  "username": "alice",
  "password": "alice-secret"
}

（5）記錄您的密鑰的ARN（Amazon 資源名稱(chēng)）值潦蝇。
不能將 Secrets Manager 密鑰與超出 調(diào)整集群的大锌畎尽：每個(gè)代理的分區(qū)數(shù)量 中所述限制的集群關(guān)聯(lián)。
（6）如果您將 AWS CLI 要?jiǎng)?chuàng)建密鑰攘乒，請(qǐng)ARN為kms-key-id參數(shù)指定密鑰 ID 或贤牛。不要指定別名。
（7）要將密鑰與您的集群關(guān)聯(lián)则酝，請(qǐng)使用 Amazon MSK 控制臺(tái)或 BatchAssociateScramSecret操作殉簸。

以下BatchAssociateScramSecret操作的示例JSON輸入將密鑰與集群相關(guān)聯(lián)：
{
  "clusterArn" : "arn:aws:kafka:us-west-2:0123456789019:cluster/SalesCluster/abcd1234-abcd-cafe-abab-9876543210ab-4",          
  "secretArnList": [
    "arn:aws:secretsmanager:us-west-2:0123456789019:secret:AmazonMSK_MyClusterSecret"
  ]
}

使用登錄憑證連接到集群

在創(chuàng)建密鑰并將其與集群關(guān)聯(lián)后，您便可以將客戶端連接到集群沽讹。以下示例步驟演示如何將客戶端連接到使用SASL/SCRAM身份驗(yàn)證的集群，以及如何使用示例主題。

1.在裝有... 的計(jì)算機(jī)上運(yùn)行以下命令 AWS CLI已安裝嵌言，正在更換 clusterARN 與您的ARN集群的怖亭。

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

2.要?jiǎng)?chuàng)建示例主題，請(qǐng)運(yùn)行以下命令挚瘟，替換 BootstrapServerString 使用您在上一步中獲得的代理端點(diǎn)之一叹谁。

<path-to-your-kafka-installation>/bin/kafka-topics.sh --create --bootstrap-server BootstrapServerString --replication-factor 3 --partitions 1 --topic ExampleTopicName

3.在您的客戶端計(jì)算機(jī)上，創(chuàng)建一個(gè)包含存儲(chǔ)在您的密鑰中的用戶憑據(jù)的JAAS配置文件刽沾。例如本慕，對(duì)于用戶 alice，使用以下內(nèi)容創(chuàng)建一個(gè)名為 users_jaas.conf 的文件侧漓。

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

4.使用以下命令將您的JAAS配置文件導(dǎo)出為KAFKA_OPTS環(huán)境參數(shù)锅尘。

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

5.在 ./tmp 目錄中創(chuàng)建一個(gè)名為 kafka.client.truststore.jks 的文件。
6.使用以下命令將JDK密鑰庫(kù)文件從您的JVMcacerts文件夾復(fù)制到您在上一步中創(chuàng)建kafka.client.truststore.jks的文件中。Replace（替換） JDKFolder 使用您的實(shí)例上JDK文件夾的名稱(chēng)藤违。例如浪腐，您的JDK文件夾可能被命名為java-1.8.0-openjdk-1.8.0.201.b09-0.amzn2.x86_64。

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

7.在 Apache Kafka 安裝的 bin 目錄中顿乒，創(chuàng)建一個(gè)名為 client_sasl.properties 的客戶端屬性文件议街，其中包含以下內(nèi)容。此文件定義了SASL機(jī)制和協(xié)議璧榄。

#JKS文件很重要
security.protocol=SASL_SSL
sasl.mechanism=SCRAM-SHA-512
ssl.truststore.location=<path-to-keystore-file>/kafka.client.truststore.jks

8.使用以下命令檢索引導(dǎo)代理字符串特漩。Replace（替換） ClusterArn 使用您的集群的 Amazon 資源名稱(chēng) (ARN)：

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

根據(jù)命令的JSON結(jié)果，保存與名為的字符串關(guān)聯(lián)的值BootstrapBrokerStringSaslScram骨杂。

9.要生成您創(chuàng)建的示例主題涂身，請(qǐng)?jiān)诳蛻舳擞?jì)算機(jī)上運(yùn)行以下命令。Replace（替換） BootstrapBrokerStringSaslScram 使用您在上一步中檢索到的值搓蚪。

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

10.要使用您創(chuàng)建的主題蛤售，在您的客戶端計(jì)算機(jī)上運(yùn)行以下命令。Replace（替換） BootstrapBrokerStringSaslScram 使用你之前獲得的價(jià)值妒潭。

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

使用用戶

創(chuàng)建用戶：您在密鑰中以鍵值對(duì)的形式創(chuàng)建用戶悴能。在 Secrets Manager 控制臺(tái)中使用明文選項(xiàng)時(shí)，應(yīng)按以下格式指定登錄憑證數(shù)據(jù)雳灾。

{
  "username": "alice",
  "password": "alice-secret"
}

撤消用戶訪問(wèn)權(quán)限：要撤消用戶訪問(wèn)集群的證書(shū)漠酿，建議先在集群ACL上刪除或強(qiáng)制執(zhí)行，然后取消關(guān)聯(lián)該密鑰佑女。這是因?yàn)椋?br> 移除用戶并不能關(guān)閉現(xiàn)有連接记靡。
對(duì)密鑰的更改最多需要 10 分鐘才能傳播。

JKS文件

jks是一種數(shù)字證書(shū)和密鑰管理文件团驱。

jks文件是Java KeyStore文件的擴(kuò)展名摸吠，主要用于存儲(chǔ)私鑰和公鑰證書(shū)。它通常用于保護(hù)應(yīng)用程序的安全連接嚎花，特別是在客戶端與服務(wù)器之間進(jìn)行安全通信時(shí)寸痢。這些文件有助于驗(yàn)證服務(wù)器身份，確保安全的數(shù)據(jù)傳輸紊选。
此外啼止，jks文件還可以用于存儲(chǔ)其他類(lèi)型的密鑰和證書(shū)信息，如身份驗(yàn)證和授權(quán)相關(guān)的數(shù)據(jù)兵罢。

一個(gè)jks文件包含了一個(gè)或多個(gè)密鑰條目献烦。每個(gè)條目都與特定的別名相關(guān)聯(lián)，并通過(guò)加密方式保護(hù)其安全卖词。文件內(nèi)還包含關(guān)于如何解密和保護(hù)這些密鑰的元信息巩那。
通過(guò)這種方式，只有具備正確權(quán)限和知識(shí)的用戶才能訪問(wèn)和管理這些密鑰和證書(shū)。這有助于維護(hù)數(shù)字安全和保密性即横。這種文件的最大特點(diǎn)是其強(qiáng)大的加密功能噪生，確保數(shù)據(jù)的安全性和完整性。

在實(shí)際應(yīng)用中东囚，jks文件通常用于Web服務(wù)器跺嗽、客戶端應(yīng)用程序和其他需要加密保護(hù)的系統(tǒng)中。這些系統(tǒng)依賴(lài)于jks文件來(lái)驗(yàn)證和管理用戶身份页藻、加密數(shù)據(jù)以及在分布式系統(tǒng)中處理各種安全操作桨嫁。
在企業(yè)環(huán)境中，特別是在采用公鑰基礎(chǔ)設(shè)施的大型系統(tǒng)中惕橙，jks文件的運(yùn)用十分常見(jiàn)且關(guān)鍵瞧甩。它們是信息安全和數(shù)據(jù)保密的關(guān)鍵組成部分钉跷。開(kāi)發(fā)人員和管理員通過(guò)正確生成弥鹦、存儲(chǔ)和使用這些文件，來(lái)確保網(wǎng)絡(luò)的安全運(yùn)行爷辙。
隨著信息技術(shù)的不斷進(jìn)步和網(wǎng)絡(luò)安全需求的增加彬坏，jks文件的應(yīng)用前景將越來(lái)越廣泛。同時(shí)還需要定期進(jìn)行安全性評(píng)估和備份以確保數(shù)據(jù)安全性和可用性膝晾。