Getting error in "refreshIngestionResources" - Could not initialize class com.azure.core.http.policy.HttpPolicyProviders

[Srivathsan97]

We are testing the usage of the Kusto connector to connect Confluent with Kusto.

We downloaded the module and uploaded the plugin/module into Confluent for using it.
After filling in the necessary details, we were able to create the connector. The connector is "RUNNING," but it couldn't connect to the Kusto due to the error shown on the screenshots below.

(FYI: The Screenshots were taken from Confluent Cloud UI)

Error:

`_Error in refreshIngestionResources: Could not initialize class com.azure.core.http.policy.HttpPolicyProviders
Could not initialize class com.azure.core.http.policy.HttpPolicyProviders

Hide stacktrace

com.microsoft.azure.kusto.ingest.exceptions.IngestionClientException: Could not initialize class com.azure.core.http.policy.HttpPolicyProviders
at com.microsoft.azure.kusto.ingest.ResourceManager$RefreshIngestionResourcesTask.refreshIngestionResources(ResourceManager.java:174)
at com.microsoft.azure.kusto.ingest.ResourceManager$RefreshIngestionResourcesTask.lambda$run$0(ResourceManager.java:136)
at com.microsoft.azure.kusto.data.instrumentation.MonitoredActivity.invoke(MonitoredActivity.java:33)
at com.microsoft.azure.kusto.data.instrumentation.MonitoredActivity.invoke(MonitoredActivity.java:26)
at com.microsoft.azure.kusto.ingest.ResourceManager$RefreshIngestionResourcesTask.run(ResourceManager.java:135)
at java.base/java.util.TimerThread.mainLoop(Timer.java:566)
at java.base/java.util.TimerThread.run(Timer.java:516)
Caused by: java.lang.NoClassDefFoundError: Could not initialize class com.azure.core.http.policy.HttpPolicyProviders
at com.azure.storage.blob.implementation.util.BuilderHelper.buildPipeline(BuilderHelper.java:103)
at com.azure.storage.blob.BlobContainerClientBuilder.buildAsyncClient(BlobContainerClientBuilder.java:163)
at com.azure.storage.blob.BlobContainerClientBuilder.buildClient(BlobContainerClientBuilder.java:124)
at com.microsoft.azure.kusto.ingest.resources.ContainerWithSas.(ContainerWithSas.java:24)
at com.microsoft.azure.kusto.ingest.ResourceManager$RefreshIngestionResourcesTask.addIngestionResource(ResourceManager.java:185)
at com.microsoft.azure.kusto.ingest.ResourceManager$RefreshIngestionResourcesTask.refreshIngestionResources(ResourceManager.java:161)
... 6 more
Caused by: java.lang.ExceptionInInitializerError: Exception java.util.ServiceConfigurationError: com.azure.core.http.policy.AfterRetryPolicyProvider: io.opentelemetry.javaagent.instrumentation.azurecore.v1_19.shaded.com.azure.core.tracing.opentelemetry.OpenTelemetryHttpPolicy not a subtype [in thread "Timer-12"]
at java.base/java.util.ServiceLoader.fail(ServiceLoader.java:593)
at java.base/java.util.ServiceLoader$LazyClassPathLookupIterator.hasNextService(ServiceLoader.java:1244)
at java.base/java.util.ServiceLoader$LazyClassPathLookupIterator.hasNext(ServiceLoader.java:1273)
at java.base/java.util.ServiceLoader$2.hasNext(ServiceLoader.java:1309)
at java.base/java.util.ServiceLoader$3.hasNext(ServiceLoader.java:1393)
at java.base/java.lang.Iterable.forEach(Iterable.java:74)
at com.azure.core.http.policy.HttpPolicyProviders.(HttpPolicyProviders.java:61)
... 12 more_`

I saw that there were changes made to the parent modules in the following repo: Azure/azure-kusto-java@d73eeb3

Could that have caused any issue here?

Can anyone please work on this as it is hindering out testing?

[ag-ramachandran]

Hi @Srivathsan97

What version are you testing with. How are you installing the connector ? Do you use the confluent hub store to install the connector ? .

Can you check if you have used the uber jar in case you are uploading to the plugin path in the Connect runtime ?

[Srivathsan97]

Hi @Srivathsan97

What version are you testing with. How are you installing the connector ? Do you use the confluent hub store to install the connector ? .

Can you check if you have used the uber jar in case you are uploading to the plugin path in the Connect runtime ?

@ag-ramachandran :

1. I am testing with the latest version of the connector (4.1.2)

2. The connector plugin was downloaded from Confluent's website
   NOTE: The Zip file is the same that is available for download in the repo

3. Yes, I use the confluent hub store only (refer the above image)

4. I shall try this suggestion as well.

Will keep you posted on this.

[Srivathsan97]

@ag-ramachandran Any analysis from your end for the issue? I will be working on deploying a local KConnect to test this out. When I previously tried, I got the same logs/errors and the task failed in Kconnect as well.

I will try further and get back by this week, but do we have an update for the confluent usage?

[ag-ramachandran]

@Srivathsan97 , local tests failed to replicate the issue you face. Was waiting on the confirmation tests you were to try. I will have a look at it later tghis week as well

[ag-ramachandran]

@Srivathsan97 , Just looked at ingestion telemetry to see if 4.1.2 seems to have an issue

Saw about 30 distinct hosts in the last 2 hours using this version too, so I will have a look at this only tomorrow or so when I can try a full replication

[Srivathsan97]

@ag-ramachandran :
Thanks for the reply.

Additionally, please find the config I used:
"config = {
"name": "XXXXXXXXX",
"connector.class": "com.microsoft.azure.kusto.kafka.connect.sink.KustoSinkConnector",
"value.converter": "org.apache.kafka.connect.storage.StringConverter",
"key.converter": "org.apache.kafka.connect.storage.StringConverter",
"kafka.auth.mode" : "KAFKA_API_KEY",
"kafka.api.key" : "XXXXXXXX",
"kafka.api.secret" : "XXXXXXXX",
"tasks.max" : 1,
"topics": "topic1,topic_2_json"
"kusto.ingestion.url": "XXXXXXXX"
"kusto.query.url": "XXXXXXXX"
"aad.auth.appid": "XXXXXXXX"
"aad.auth.appkey": "XXXXXXXX"
"aad.auth.authority": "XXXXXXXX"
"kusto.tables.topics.mapping": "[{'topic': 'topic1','db': 'testdb', 'table': 'test_table_1','format': 'avro', 'mapping':'AvroMapping'},{'topic': 'topic_2_json','db': 'testdb', 'table': 'test_table_2','format': 'json', 'mapping':'JsonMapping'}]"
}"

Also, following are additional stack traces:

"
`

throwIfFatal detected a jvm fatal exception, which is thrown and logged below:Could not initialize class com.azure.core.http.policy.HttpPolicyProvidersHide stacktracejava.lang.NoClassDefFoundError: Could not initialize class com.azure.core.http.policy.HttpPolicyProviders at com.azure.identity.implementation.IdentityClientBase.setupPipeline(IdentityClientBase.java:864) at com.azure.identity.implementation.IdentityClientBase.getPipeline(IdentityClientBase.java:897) at com.azure.identity.implementation.IdentityClientBase.initializeHttpPipelineAdapter(IdentityClientBase.java:878) at com.azure.identity.implementation.IdentityClientBase.getManagedIdentityConfidentialClient(IdentityClientBase.java:448) at com.azure.identity.implementation.IdentityClient.lambda$getManagedIdentityConfidentialClientApplication$5(IdentityClient.java:155) at reactor.core.publisher.MonoDefer.subscribe(MonoDefer.java:45) at reactor.core.publisher.InternalMonoOperator.subscribe(InternalMonoOperator.java:64) at reactor.core.publisher.MonoCacheTime.subscribeOrReturn(MonoCacheTime.java:143) at reactor.core.publisher.Mono.subscribe(Mono.java:4475) at reactor.core.publisher.Mono.block(Mono.java:1741) at com.microsoft.azure.kusto.data.auth.ManagedIdentityTokenProvider.acquireAccessTokenImpl(ManagedIdentityTokenProvider.java:44) at com.microsoft.azure.kusto.data.instrumentation.MonitoredActivity.invoke(MonitoredActivity.java:33) at com.microsoft.azure.kusto.data.auth.TokenProviderBase.acquireAccessToken(TokenProviderBase.java:32) at com.microsoft.azure.kusto.data.ClientImpl.generateIngestAndCommandHeaders(ClientImpl.java:405) at com.microsoft.azure.kusto.data.ClientImpl.executeToJsonResult(ClientImpl.java:213) at com.microsoft.azure.kusto.data.ClientImpl.executeImpl(ClientImpl.java:173) at com.microsoft.azure.kusto.data.ClientImpl.lambda$execute$0(ClientImpl.java:122) at com.microsoft.azure.kusto.data.instrumentation.MonitoredActivity.invoke(MonitoredActivity.java:33) at com.microsoft.azure.kusto.data.ClientImpl.execute(ClientImpl.java:121) at com.microsoft.azure.kusto.data.ClientImpl.execute(ClientImpl.java:116) at com.microsoft.azure.kusto.data.ClientImpl.execute(ClientImpl.java:111) at com.microsoft.azure.kusto.data.ClientImpl.execute(ClientImpl.java:106) at com.microsoft.azure.kusto.ingest.ResourceManager$RefreshIngestionResourcesTask.lambda$refreshIngestionResources$e961bae5$1(ResourceManager.java:153) at io.github.resilience4j.retry.Retry.lambda$decorateCheckedSupplier$3f69f149$1(Retry.java:137) at com.microsoft.azure.kusto.ingest.ResourceManager$RefreshIngestionResourcesTask.refreshIngestionResources(ResourceManager.java:154) at com.microsoft.azure.kusto.ingest.ResourceManager$RefreshIngestionResourcesTask.lambda$run$0(ResourceManager.java:136) at com.microsoft.azure.kusto.data.instrumentation.MonitoredActivity.invoke(MonitoredActivity.java:33) at com.microsoft.azure.kusto.data.instrumentation.MonitoredActivity.invoke(MonitoredActivity.java:26) at com.microsoft.azure.kusto.ingest.ResourceManager$RefreshIngestionResourcesTask.run(ResourceManager.java:135) at java.base/java.util.TimerThread.mainLoop(Timer.java:566) at java.base/java.util.TimerThread.run(Timer.java:516) Caused by: java.lang.ExceptionInInitializerError: Exception java.util.ServiceConfigurationError: com.azure.core.http.policy.AfterRetryPolicyProvider: io.opentelemetry.javaagent.instrumentation.azurecore.v1_19.shaded.com.azure.core.tracing.opentelemetry.OpenTelemetryHttpPolicy not a subtype [in thread "Timer-3"] at java.base/java.util.ServiceLoader.fail(ServiceLoader.java:593) at java.base/java.util.ServiceLoader$LazyClassPathLookupIterator.hasNextService(ServiceLoader.java:1244) at java.base/java.util.ServiceLoader$LazyClassPathLookupIterator.hasNext(ServiceLoader.java:1273) at java.base/java.util.ServiceLoader$2.hasNext(ServiceLoader.java:1309) at java.base/java.util.ServiceLoader$3.hasNext(ServiceLoader.java:1393) at java.base/java.lang.Iterable.forEach(Iterable.java:74) at com.azure.core.http.policy.HttpPolicyProviders.(HttpPolicyProviders.java:61) at com.azure.storage.blob.implementation.util.BuilderHelper.buildPipeline(BuilderHelper.java:103) at com.azure.storage.blob.BlobContainerClientBuilder.buildAsyncClient(BlobContainerClientBuilder.java:163) at com.azure.storage.blob.BlobContainerClientBuilder.buildClient(BlobContainerClientBuilder.java:124) at com.microsoft.azure.kusto.ingest.resources.ContainerWithSas.(ContainerWithSas.java:24) at com.microsoft.azure.kusto.ingest.ResourceManager$RefreshIngestionResourcesTask.addIngestionResource(ResourceManager.java:185) at com.microsoft.azure.kusto.ingest.ResourceManager$RefreshIngestionResourcesTask.refreshIngestionResources(ResourceManager.java:161) ... 6 more -- ` "

Hope this helps.
Let me know if you need more details from my end.

Regards

[ag-ramachandran]

Thanks for the detail

2 points @Srivathsan97

a) Are you using some other azure connectors as well in the Kafka Connect instance ( to clear any issues on classpath)

b) Were you able to try with Uber jars.

It will take a couple of days for me to set up the right infra for me to test a similar set up. Will keep this ticket updated in case of a replication

[Srivathsan97]

@ag-ramachandran : I have tried using the Uber jars in KConnect in Confluent Platform and data moved into the table in Kusto.
There was one other thing we noticed:

1. There are 2 topics and tables being mapped.
2. Data was moving into Table 2 (topic 2 was a JSON data).
3. Data was not moving into Table 1 (topic 2 was AVRO data).
4. Trying to access the logs to find the issue, I will keep you posted on the same.

Also, raised a ticket with the confluent to look into the same.

[ag-ramachandran]

Sure @Srivathsan97 , email is on the profile in case you want to have a working session to look at the issue

[Srivathsan97]

@ag-ramachandran : The data flow via local KConnect runtime, using Confluent Platform, is now working and we can see the data flow for both the tables.

But in confluent cloud, we still don't have a solution and the confluent team are looking into it.
So if this ticket needs to be closed, it can be, or can wait until Confluent finds a solution.

[ag-ramachandran]

Hello @Srivathsan97 , we can have a look on Monday if it works for you. My email is in the GH profile - ramacg at my org, in case you want to discuss on call. I work IST