https://datahubproject.io logo
Join Slack
Powered by
I am trying to ingest Kafka and getting this error...
# ingestion
r
I am trying to ingest Kafka and getting this error
Copy code
KafkaException: KafkaError{code=_INVALID_ARG,val=-186,str="Failed to create consumer: Invalid sasl.kerberos.kinit.cmd value: Property not available: "sasl.kerberos.keytab""}
Anyone faced this before? For context, we use Confluent Kafka
b
Are you running on windows by chance?
r
no, macOS
Any ideas or if anyone could share an example of a Kafka ingestion using confluent?
m
@red-pizza-28006 Could you give a full stackstrace and your recipe file? From the looks it seems like you are unable to connect to kafka which is setup to have kerberos auth in your case. Is that how your kafka is setup?
r
Sorry for the late reply. that is correct, this is a secured Kafka instance.
Copy code
File "/Users/ajaykumarmuppuri/opt/anaconda3/lib/python3.8/site-packages/datahub/ingestion/source/kafka.py", line 61, in __init__
    58   def __init__(self, config: KafkaSourceConfig, ctx: PipelineContext):
    59       super().__init__(ctx)
    60       self.source_config = config
--> 61       self.consumer = confluent_kafka.Consumer(
    62           {
    ..................................................
     self = # error calling repr and str
     config = KafkaSourceConfig(env='PROD', connection=KafkaConsumerConnectionConfig(bootstrap='kafka:9092', schema_registry_url='sr:8081', schema_registry_config={}, cons
               umer_config={'sasl.username': 'test1', 'sasl.password': 'test2', 'security.protocol': 'sasl_plaintext'}), topic_patterns=AllowDenyPattern(allow=['.*'], deny=['^_.*'], ign
               oreCase=True, alphabet='[A-Za-z0-9 _.-]'))
     KafkaSourceConfig = <class 'datahub.ingestion.source.kafka.KafkaSourceConfig'>
     ctx = <datahub.ingestion.api.common.PipelineContext object at 0x7fc9c1668280>
     PipelineContext = <class 'datahub.ingestion.api.common.PipelineContext'>
     self.source_config = KafkaSourceConfig(env='PROD', connection=KafkaConsumerConnectionConfig(bootstrap='kafka:9092', schema_registry_url='<https://sr:8081>', schema_registry_config={}, cons
                           umer_config={'sasl.username': 'test1', 'sasl.password': 'test2/', 'security.protocol': 'sasl_plaintext'}), topic_patterns=AllowDenyPattern(allow=['.*'], deny=['^_.*'], ign
                           oreCase=True, alphabet='[A-Za-z0-9 _.-]'))
     self.consumer = # AttributeError
          self = # error calling repr and str
     confluent_kafka.Consumer = <class 'cimpl.Consumer'>
    ..................................................

---- (full traceback above) ----
File "/Users/ajaykumarmuppuri/opt/anaconda3/lib/python3.8/site-packages/datahub/entrypoints.py", line 93, in main
    sys.exit(datahub(standalone_mode=False, **kwargs))
File "/Users/ajaykumarmuppuri/opt/anaconda3/lib/python3.8/site-packages/click/core.py", line 829, in __call__
    return self.main(*args, **kwargs)
File "/Users/ajaykumarmuppuri/opt/anaconda3/lib/python3.8/site-packages/click/core.py", line 782, in main
    rv = self.invoke(ctx)
File "/Users/ajaykumarmuppuri/opt/anaconda3/lib/python3.8/site-packages/click/core.py", line 1259, in invoke
    return _process_result(sub_ctx.command.invoke(sub_ctx))
File "/Users/ajaykumarmuppuri/opt/anaconda3/lib/python3.8/site-packages/click/core.py", line 1259, in invoke
    return _process_result(sub_ctx.command.invoke(sub_ctx))
File "/Users/ajaykumarmuppuri/opt/anaconda3/lib/python3.8/site-packages/click/core.py", line 1066, in invoke
    return ctx.invoke(self.callback, **ctx.params)
File "/Users/ajaykumarmuppuri/opt/anaconda3/lib/python3.8/site-packages/click/core.py", line 610, in invoke
    return callback(*args, **kwargs)
File "/Users/ajaykumarmuppuri/opt/anaconda3/lib/python3.8/site-packages/datahub/cli/ingest_cli.py", line 52, in run
    pipeline = Pipeline.create(pipeline_config)
File "/Users/ajaykumarmuppuri/opt/anaconda3/lib/python3.8/site-packages/datahub/ingestion/run/pipeline.py", line 136, in create
    return cls(config)
File "/Users/ajaykumarmuppuri/opt/anaconda3/lib/python3.8/site-packages/datahub/ingestion/run/pipeline.py", line 104, in __init__
    self.source: Source = source_class.create(
File "/Users/ajaykumarmuppuri/opt/anaconda3/lib/python3.8/site-packages/datahub/ingestion/source/kafka.py", line 76, in create
    return cls(config, ctx)
File "/Users/ajaykumarmuppuri/opt/anaconda3/lib/python3.8/site-packages/datahub/ingestion/source/kafka.py", line 61, in __init__
    self.consumer = confluent_kafka.Consumer(

KafkaException: KafkaError{code=_INVALID_ARG,val=-186,str="Failed to create consumer: Invalid sasl.kerberos.kinit.cmd value: Property not available: "sasl.kerberos.keytab""}
13 Views
Open in Slack
PreviousNext
Powered by