我有一个python kafka,它可以工作,代码如下:
class TokenProvider(object):
def __init__(self,client_id,client_secret):
self.client_id = client_id
self.client_secret = client_secret
def token(self):
token_url = 'https://test.com/protocol/openid-connect/token'
client = BackendApplicationClient(client_id=self.client_id)
oauth = OAuth2Session(client=client)
token_json = oauth.fetch_token(token_url=token_url, client_id=self.client_id, client_secret=self.client_secret)
token = token_json['access_token']
#print(token)
return token
consumer = KafkaConsumer(
group_id=None,
bootstrap_servers=['test.com:9094'],
security_protocol='SASL_SSL',
sasl_mechanism='OAUTHBEARER',
sasl_oauth_token_provider=TokenProvider(client_id,client_secret),
ssl_check_hostname=False,
ssl_context=create_ssl_context(),
auto_offset_reset=offset,
enable_auto_commit=False,
value_deserializer=lambda m: decode(m)
)
consumer.subscribe(topics=['test.stream'])
我的合流python如下所示,我得到了这个错误
cimpl.KafkaException: KafkaError{code=_INVALID_ARG,val=-186,str="Property "oauthbearer_token_refresh_cb" must be set through dedicated .._set_..() function"}
c = Consumer({
'bootstrap.servers': 'test.com:9094',
'sasl.mechanism': 'OAUTHBEARER',
'security.protocol': 'SASL_SSL',
'oauthbearer_token_refresh_cb': TokenProvider(client_id,client_secret),
'group.id': str(uuid.uuid1()),
'auto.offset.reset': 'earliest'
})
c.subscribe(['test.stream'])
那么,我如何让融合的卡夫卡发挥作用呢?我似乎对使用oauthbearer和SASL的oauthbearer\u令牌\u刷新\u cb有问题
本质上,我使用jwt令牌进行身份验证
从源代码上看,python和go客户端还不支持OAuthBear
根据位于https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md的文档,必须使用
rd_kafka_conf_set_oauthbearer_token_refresh_cb()
设置oauthbearer_token_refresh_cb
选项。但是请注意,您正试图将其设置为TokenProvider
实例,该实例不可调用,因此您可能希望传递TokenProvider(...).token
相关问题 更多 >
编程相关推荐