Hi team I am using Kafka emitter, and we have setup authentication for our Kafka client environment and schema registry (local) is pointing to https And we write our action pipeline from which we passing our configuration. This my emitter: For Kafka it’s able to authenticate and connect to client Kafka but for schema registry it is showing Ssl certificate error. Please refer the action config that we have as well

I am afraid I don’t fully understand the context here. Let me see if I can try to focus in on the problem, there are some configuration snippets. First I am guessing you mean the python emitter, not a java one, based on the first screenshot. So this doc on the python emitter. That seems to be created using that other data in the second screenshot. What is the error specifically? Is it an invalid cert or a protocol mismatch? In the schema registry url, did you remove the protocol? Why is there a url path? When you say schema registry

local

what do you mean by that? Do you mean using GMS as the schema registry with an

INTERNAL

configuration? If using the INTERNAL configuration, how did you implement the SSL encryption? What is doing the decoding and what is managing the certs there?

Yes we are using python emitter and in kafka configuration we are have setup basic authentication for kafka ( for external communication) and in producer config we are providing required configuration for this in order to connect to Kafka deployed in other cluster. For schema registry we haven’t setup any authentication it’s only on https ( using internal schema registry gms) and in schema registry config we are passing ssl.ca.location but it’s unable to verify certs and it’s working when the schema registry url is on http. Error : Http local issueers unable to verify certs And all these configuration coming from my action Pipeline config section.

Is this a custom action? What do you do with that config in your action?

We have custom action that will convert MCL into MCP and we are using Kafka emitter to push those MCP into client environment which is being replicated there via mirror maker. In the action config we are passing the required keys like bootstrap url, ca location , username and password for basic auth. And we are able to connect to Kafka of client environment but with schema registry( internal gms) it’s not working on https( ssl certificate verification failed: local issuers) but working on http And we have the provided configuration as shown in above picture

Based on the code we pass in everything to the Confluent’s Schema Registry client whatever you set in

schema_registry_config

-> https://github.com/datahub-project/datahub/blob/ddcd5109dcbe01aac28347cf34221d65cb5faa30/metadata-ingestion/src/datahub/emitter/kafka_emitter.py#L65

I tried to understand and provide the values like this now. Is it a correct way of passing ? @dazzling-judge-80093

This looks fine to me

While running I am getting x509 certificate signed by unknown authority error

Can you check if the generated config looks ok? It is hard to see from this code snippet.

Does Kafka and schema requires different certificates ?

It depends on your setup

The values are coming from action config file and I have checked the values I am getting this error : X509: certificate signed by unknown authority

But here the gms’ healthcheck threw the certificate issue and not the action, am I right?

Right

I think this line throws the error -> https://github.com/acryldata/datahub-actions/blob/fa7b2d5fa6c8e3d6ea3b02979652425469ddf24a/docker/datahub-actions/Dockerfile#L49

Hi tamas The above issue is fix but after adding the config into schema registry client like this My python code is giving this error : str object is not a mapping Can you please have a look on this picture ? And check it’s a correct way of providing ?

@creamy-van-28626, this issue happens if your variable is not a dict but a string you pass. Please, can you check/print

self.action_config.ssl_ca_istio_location

?

I am Using only Kafka emitter and in that I am providing all the configuration But only for schema registry certificate verification I am getting error :

is it intentional you are passing a different ca for producer than for schema_registry?

If we pass the same also same error

Do you pass the ca file to the curl command?

Yes

The config seems to be ok. Do you pass the same ca file location which you used in your curl command?

Curl —cacert etc/ssl/certs/ca.cry (url) I am using this command and getting response

I checked the code, and I’m confident the set properties are passed to the Schema Registry client. Here is my test recipe:

source:
  type: "kafka"
  config:
    connection:
      bootstrap: "localhost:9092"
      #consumer_config:
      #  security.protocol: "SASL_SSL"
      #  sasl.mechanism: "PLAIN"
      #  sasl.username: "kafka_key"
      #  sasl.password: "kafka_secret"
      schema_registry_url: "<http://localhost:8081>"
      schema_registry_config:
        # See <https://docs.confluent.io/platform/current/clients/confluent-kafka-python/html/index.html#confluent_kafka.schema_registry.SchemaRegistryClient>
        # Most of these config keys are optional.
        ssl.ca.location: "certificate_ca.pem"
        ssl.certificate.location: "certificate.pem"
        ssl.key.location: "private_key.pem"
        ssl.key.password: "XXXXXXX"

I am using Kafka emitter only and I am setting all these configuration like this

For the emitter the config looks like this:

sink:
  type: "datahub-kafka"
  config:
    connection:
      bootstrap: localhost:9092
      schema_registry_url: <http://localhost:8081>
      schema_registry_config:
        # See <https://docs.confluent.io/platform/current/clients/confluent-kafka-python/html/index.html#confluent_kafka.schema_registry.SchemaRegistryClient>
        # Most of these config keys are optional.
        #ssl.ca.location: "certificate_ca.pem"
        ssl.certificate.location: "certificate.pem"
        ssl.key.location: "private_key.pem"

I have this python code for Kafka emitter and all the configuration are passed from action pipeline

As this is a custom code, I would debug your code if it passes properly the parameters you set. Unfortunately, I can only test/check what is in our codebase and there the Kafka Emitter should pass the property if it is passed properly

Let me share you

Can’t you debug your code in standalone mode without the action framework?

Yeah debug that it’s working fine

or log from your code the config you pass to the Kafka Emitter and check from the Action Framework logs

Setting ssl.ca.location to false also not working