# Issue's in configuring kafka input plugin with TLS

**URL:** <https://discuss.elastic.co/t/issues-in-configuring-kafka-input-plugin-with-tls/337761>\
**Category:** Logstash\
**Created:** [July 6, 2023, 7:50am UTC](https://discuss.elastic.co/t/issues-in-configuring-kafka-input-plugin-with-tls/337761 "2023-07-06T07:50:34Z")\
**Posts on this page:** 2\
**Page:** 1

<div class="post-metadata">

**Author:** ![girish.ms](https://avatars.discourse-cdn.com/v4/letter/g/ec9cab/32.png) [@girish.ms](https://discuss.elastic.co/u/girish.ms)\
**Post date:** [July 6, 2023, 7:50am UTC](https://discuss.elastic.co/t/issues-in-configuring-kafka-input-plugin-with-tls/337761/1 "2023-07-06T07:50:34Z")

</div>

Description of the problem:

I'm having trouble integrating Kafka with Logstash, and Kafka is configured with TLS.  
I am getting the following exceptions when trying to provide PKCS12 format TLS certificates in the Kafka input plugin.

1. Logstash version - **8.8.1**
2. Logstash installation source - **Bitnami Logstash helm chart**
3. How is Logstash being run - **Kubernetes stateful-set**
4. Kafka Input plugin - [Kafka input plugin | Logstash Reference [8.8] | Elastic](https://www.elastic.co/guide/en/logstash/current/plugins-inputs-kafka.html)

Error:

``  
Unable to create Kafka consumer from given configuration {:kafka\_error\_message=\>#\<Java::OrgApacheKafkaCommon::KafkaException: **Failed to construct kafka consumer\>, :cause=\>#\<Java::OrgApacheKafkaCommon::KafkaException: Failed to load SSL keystore /usr/share/logstash/config/certs/ample-user/p12 of type PKCS12\>}**  
[2023-07-05T16:16:33,533][ERROR][logstash.javapipeline][main][0a81a40d970eb2a3701645643b7b64f3e3bed2443d45da944a470013bf8ce0ef] **A plugin had an unrecoverable error. Will restart this plugin.**

```auto

Helm values:

```

```
extraEnvVars:
  - name: KAFKA_HOST
    value: "kafka-kafka-bootstrap"
  - name: KAFKA_PORT
    value: "9093"

extraVolumeMounts:
  - name: kafka-cluster-certs
    mountPath: /usr/share/logstash/config/certs/kafka-cluster
    readOnly: true
  - name: ample-user
    mountPath: /usr/share/logstash/config/certs/ample-user
    readOnly: true

extraVolumes: 
  - name: kafka-cluster-certs
    secret:
      secretName: kafka-cluster-ca-cert
      items:
      - key: ca.p12
        path: p12
      - key: ca.password
        path: password
  - name: ample-user
    secret:
      secretName: ample-user
      items:
      - key: user.p12
        path: p12
      - key: user.password
        path: password

input: |-
  kafka {
    bootstrap_servers => ["${KAFKA_HOST}:${KAFKA_PORT}"]
    topics => ["fluent-log-vault"]
    type => "vault"
    security_protocol => "SSL"
    ssl_truststore_type => "PKCS12"
    ssl_keystore_type => "PKCS12"
    ssl_truststore_location => "/usr/share/logstash/config/certs/kafka-cluster/p12"
    ssl_truststore_password => "/usr/share/logstash/config/certs/password"
    ssl_keystore_location => "/usr/share/logstash/config/certs/ample-user/p12"
    ssl_keystore_password => "/usr/share/logstash/config/certs/ample-user/password"
  }

```

```auto

Detailed Logs:

```

[2023-07-05T16:16:33,529][INFO][org.apache.kafka.common.metrics.Metrics][main][0a81a40d970eb2a3701645643b7b64f3e3bed2443d45da944a470013bf8ce0ef] Metrics scheduler closed  
[2023-07-05T16:16:33,530][INFO][org.apache.kafka.common.metrics.Metrics][main][0a81a40d970eb2a3701645643b7b64f3e3bed2443d45da944a470013bf8ce0ef] Closing reporter org.apache.kafka.common.metrics.JmxReporter  
[2023-07-05T16:16:33,530][INFO][org.apache.kafka.common.metrics.Metrics][main][0a81a40d970eb2a3701645643b7b64f3e3bed2443d45da944a470013bf8ce0ef] Metrics reporters closed  
[2023-07-05T16:16:33,530][INFO][org.apache.kafka.common.utils.AppInfoParser][main][0a81a40d970eb2a3701645643b7b64f3e3bed2443d45da944a470013bf8ce0ef] App info kafka.consumer for logstash-0 unregistered  
[2023-07-05T16:16:33,531][ERROR][logstash.inputs.kafka][main][0a81a40d970eb2a3701645643b7b64f3e3bed2443d45da944a470013bf8ce0ef] Unable to create Kafka consumer from given configuration {:kafka\_error\_message=\>#\<Java::OrgApacheKafkaCommon::KafkaException: Failed to construct kafka consumer\>, :cause=\>#\<Java::OrgApacheKafkaCommon::KafkaException: Failed to load SSL keystore /usr/share/logstash/config/certs/ample-user/p12 of type PKCS12\>}  
[2023-07-05T16:16:33,533][ERROR][logstash.javapipeline][main][0a81a40d970eb2a3701645643b7b64f3e3bed2443d45da944a470013bf8ce0ef] A plugin had an unrecoverable error. Will restart this plugin.  
Pipeline\_id:main  
Plugin: \<LogStash::Inputs::Kafka ssl\_keystore\_location=\>"/usr/share/logstash/config/certs/ample-user/p12", ssl\_keystore\_password=\>, topics=\>["fluent-log-vault"], ssl\_truststore\_location=\>"/usr/share/logstash/config/certs/kafka-cluster/p12", ssl\_truststore\_password=\>, security\_protocol=\>"SSL", id=\>"0a81a40d970eb2a3701645643b7b64f3e3bed2443d45da944a470013bf8ce0ef", type=\>"vault", ssl\_truststore\_type=\>"PKCS12", bootstrap\_servers=\>"kafka-kafka-bootstrap:9093", ssl\_keystore\_type=\>"PKCS12", codec=\>\<LogStash::Codecs::Plain id=\>"plain\_c97407f8-f507-4c16-a8a6-8cc822b8e0eb", enable\_metric=\>true, charset=\>"UTF-8"\>, enable\_metric=\>true, connections\_max\_idle\_ms=\>540000, metadata\_max\_age\_ms=\>300000, request\_timeout\_ms=\>40000, schema\_registry\_ssl\_keystore\_type=\>"jks", schema\_registry\_ssl\_truststore\_type=\>"jks", schema\_registry\_validation=\>"auto", auto\_commit\_interval\_ms=\>5000, check\_crcs=\>true, client\_dns\_lookup=\>"use\_all\_dns\_ips", client\_id=\>"logstash", consumer\_threads=\>1, enable\_auto\_commit=\>true, fetch\_max\_bytes=\>52428800, fetch\_max\_wait\_ms=\>500, group\_id=\>"logstash", heartbeat\_interval\_ms=\>3000, isolation\_level=\>"read\_uncommitted", key\_deserializer\_class=\>"org.apache.kafka.common.serialization.StringDeserializer", max\_poll\_interval\_ms=\>300000, max\_partition\_fetch\_bytes=\>1048576, max\_poll\_records=\>500, receive\_buffer\_bytes=\>32768, reconnect\_backoff\_ms=\>50, retry\_backoff\_ms=\>100, send\_buffer\_bytes=\>131072, session\_timeout\_ms=\>10000, value\_deserializer\_class=\>"org.apache.kafka.common.serialization.StringDeserializer", poll\_timeout\_ms=\>100, ssl\_endpoint\_identification\_algorithm=\>"https", sasl\_mechanism=\>"GSSAPI", decorate\_events=\>"none"\>  
Error: Failed to construct kafka consumer  
Exception: Java::OrgApacheKafkaCommon::KafkaException  
Stack: org.apache.kafka.clients.consumer.KafkaConsumer.(org/apache/kafka/clients/consumer/KafkaConsumer.java:830)  
org.apache.kafka.clients.consumer.KafkaConsumer.(org/apache/kafka/clients/consumer/KafkaConsumer.java:666)  
org.apache.kafka.clients.consumer.KafkaConsumer.(org/apache/kafka/clients/consumer/KafkaConsumer.java:647)  
org.apache.kafka.clients.consumer.KafkaConsumer.(org/apache/kafka/clients/consumer/KafkaConsumer.java:627)  
jdk.internal.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)  
jdk.internal.reflect.NativeConstructorAccessorImpl.newInstance(jdk/internal/reflect/NativeConstructorAccessorImpl.java:77)  
jdk.internal.reflect.DelegatingConstructorAccessorImpl.newInstance(jdk/internal/reflect/DelegatingConstructorAccessorImpl.java:45)  
java.lang.reflect.Constructor.newInstanceWithCaller(java/lang/reflect/Constructor.java:499)  
java.lang.reflect.Constructor.newInstance(java/lang/reflect/Constructor.java:480)  
org.jruby.javasupport.JavaConstructor.newInstanceDirect(org/jruby/javasupport/JavaConstructor.java:269)  
org.jruby.RubyClass.new(org/jruby/RubyClass.java:890)  
org.jruby.RubyClass$INVOKER$i$newInstance.call(org/jruby/RubyClass$INVOKER$i$newInstance.gen)  
RUBY.create\_consumer(/opt/bitnami/logstash/vendor/bundle/jruby/2.6.0/gems/logstash-integration-kafka-11.2.1-java/lib/logstash/inputs/kafka.rb:477)  
RUBY.run(/opt/bitnami/logstash/vendor/bundle/jruby/2.6.0/gems/logstash-integration-kafka-11.2.1-java/lib/logstash/inputs/kafka.rb:297)  
org.jruby.RubyEnumerable$18.call(org/jruby/RubyEnumerable.java:815)  
org.jruby.RubyEnumerator$1.call(org/jruby/RubyEnumerator.java:400)  
org.jruby.RubyFixnum.times(org/jruby/RubyFixnum.java:308)  
org.jruby.RubyInteger$INVOKER$i$0$0$times.call(org/jruby/RubyInteger$INVOKER$i$0$0$times.gen)  
org.jruby.RubyClass.finvokeWithRefinements(org/jruby/RubyClass.java:514)  
org.jruby.RubyClass.finvoke(org/jruby/RubyClass.java:502)  
org.jruby.RubyBasicObject.callMethod(org/jruby/RubyBasicObject.java:387)  
org.jruby.RubyEnumerator. **each** (org/jruby/RubyEnumerator.java:396)  
org.jruby.RubyEnumerator.each(org/jruby/RubyEnumerator.java:392)  
org.jruby.RubyEnumerator$INVOKER$i$each.call(org/jruby/RubyEnumerator$INVOKER$i$each.gen)  
org.jruby.RubyEnumerable.collectCommon(org/jruby/RubyEnumerable.java:807)  
org.jruby.RubyEnumerable.map(org/jruby/RubyEnumerable.java:799)  
org.jruby.RubyEnumerable$INVOKER$s$0$0$map.call(org/jruby/RubyEnumerable$INVOKER$s$0$0$map.gen)  
RUBY.run(/opt/bitnami/logstash/vendor/bundle/jruby/2.6.0/gems/logstash-integration-kafka-11.2.1-java/lib/logstash/inputs/kafka.rb:295)  
RUBY.inputworker(/opt/bitnami/logstash/logstash-core/lib/logstash/java\_pipeline.rb:414)  
RUBY.start\_input(/opt/bitnami/logstash/logstash-core/lib/logstash/java\_pipeline.rb:405)  
org.jruby.RubyProc.call(org/jruby/RubyProc.java:309)  
java.lang.Thread.run(java/lang/Thread.java:833)

```auto

Is there any way to solve this problem?
```

---

<div class="post-metadata">

**Author:** ![system](https://us1.discourse-cdn.com/elastic/original/3X/1/a/1ac57faf039f6b580b3f104ef42a2a89e41014de.png) [@system](https://discuss.elastic.co/u/system)\
**Post date:** [August 3, 2023, 7:51am UTC](https://discuss.elastic.co/t/issues-in-configuring-kafka-input-plugin-with-tls/337761/2 "2023-08-03T07:51:01Z")

</div>

This topic was automatically closed 28 days after the last reply. New replies are no longer allowed.