如何在 avro-console-producer 中发送授权标头（或）访问令牌答案

【问题标题】：How to send authorization header (or) access token in avro-console-producer如何在 avro-console-producer 中发送授权标头（或）访问令牌
【发布时间】：2021-07-28 16:12:06
【问题描述】：

我们有一个启用 SSL 的 kafka 代理，并且通过 Keycloak 访问模式注册表。从外部机器，我可以使用 kafka-console-producer 发送数据，下面是我的配置。

ssl.properties：

security.protocol=SASL_SSL
ssl.truststore.location=truststore.jks
ssl.truststore.password=password
sasl.mechanism=PLAIN

jaas.conf：

KafkaClient
{
org.apache.kafka.common.security.plain.PlainLoginModule required
username="<user-name>"
password="<password>";
};

export KAFKA_OPTS="-Djavax.net.ssl.trustStore=truststore.jks -Djavax.net.ssl.trustStorePassword=password -Djava.security.auth.login.config=jaas.conf"
 
./kafka-console-producer --bootstrap-server broker-url:<external_port> --topic sample.data --producer.config ssl.properties

你好样本数据发送

我可以看到他们使用消费者

现在，对于模式注册表，我需要获取令牌，如下所示：

curl  -k --data 'grant_type=password&client_id=schema-registry-client&username=username&password=password' https://<keycloakurl>/auth/realms/<namespace>/protocol/openid-connect/token

输出：

{"access_token":"<access_token>","expires_in":600,"refresh_expires_in":1800,"refresh_token":"<refresh_token>","token_type":"bearer","not-before-policy":0,"session_state":"4117e69c-afe9-43ae-9756-90b151f0b536","scope":"profile email"}

 curl -k  -H "Authorization: Bearer <access_token>"  
 https://<sc_url>/schemaregistry/subjects

输出：

 ["test.data-value"]

问题是，如何在 avro-console-producer 中使用 access_token ？我没有办法。

【问题讨论】：

标签： apache-kafka kafka-producer-api confluent-schema-registry

【解决方案1】：

基于注册表 REST 客户端的 on the source code，这些属性应该是您想要的（未经测试）

bearer.auth.credentials.source="STATIC_TOKEN"
bearer.auth.token="your token"

【讨论】：

不，这不起作用。我可以在某个时候分享快照。
澄清一下，第一个属性是一个字面量，这里只有第二个值需要改变。除此之外，尚不清楚您尝试使用的 Avro 生产者版本，因此这些设置可能与我链接到的源代码不匹配
您好，抱歉耽搁了。我正在使用 confluent-6.0.1 avro-producer。
我可以使用访问令牌获取注册主题的版本。但是以下命令失败： ./kafka-avro-console-producer --bootstrap-server --topic --producer.config ssl.properties --property value.schema.id=1 --property schema.registry.url="https:///schemaregistry/" --property bearer.auth.credentials.source="STATIC_TOKEN" --property bearer.auth.token="Authorization: Bearer " 和得到如下错误：
org.apache.kafka.common.errors.SerializationException：检索 id 1 的架构时出错原因：javax.net.ssl.SSLHandshakeException：sun.security.validator.ValidatorException：PKIX 路径构建失败： sun.security.provider.certpath.SunCertPathBuilderException：无法找到请求目标的有效证书路径