Realtime Ingestion Using Kafka with SSL | Error: Failed to construct kafka consumer

Trying to create a table. I have restarted this project again after 1 year but having the same issue I'm sure I'm doing something wrong, need I really appreciate if someone helps me to setup this up. my org is not using slack so not possible to post the issue on slack community. here is my old issue https://github.com/apache/pinot/issues/8041

> 2023/02/03 11:52:20.696 INFO [AddTableCommand] [main] {"code":500,"error":"org.apache.pinot.shaded.org.apache.kafka.common.KafkaException: Failed to construct kafka consumer"}

```
$bin/pinot-admin.sh AddTable -schemaFile /opt/pinot/volume_app/dirrtl/dirrtl-schema.json -tableConfigFile /opt/pinot/volume_app/dirrtl/dirrtl-table-realtime.json -exec#

[0.001s][warning][gc] -Xloggc is deprecated. Will use -Xlog:gc:gc-pinot-controller.log instead.
SLF4J: Class path contains multiple SLF4J bindings.
SLF4J: Found binding in [jar:file:/opt/pinot/lib/pinot-all-0.12.0-jar-with-dependencies.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: Found binding in [jar:file:/opt/pinot/plugins/pinot-file-system/pinot-s3/pinot-s3-0.12.0-shaded.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: Found binding in [jar:file:/opt/pinot/plugins/pinot-metrics/pinot-yammer/pinot-yammer-0.12.0-shaded.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: Found binding in [jar:file:/opt/pinot/plugins/pinot-metrics/pinot-dropwizard/pinot-dropwizard-0.12.0-shaded.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: Found binding in [jar:file:/opt/pinot/plugins/pinot-environment/pinot-azure/pinot-azure-0.12.0-shaded.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: Found binding in [jar:file:/opt/pinot/plugins/pinot-input-format/pinot-parquet/pinot-parquet-0.12.0-shaded.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: See http://www.slf4j.org/codes.html#multiple_bindings for an explanation.
SLF4J: Actual binding is of type [org.apache.logging.slf4j.Log4jLoggerFactory]
WARNING: sun.reflect.Reflection.getCallerClass is not supported. This will impact performance.
WARNING: An illegal reflective access operation has occurred
WARNING: Illegal reflective access by org.codehaus.groovy.reflection.CachedClass (file:/opt/pinot/lib/pinot-all-0.12.0-jar-with-dependencies.jar) to method java.lang.Object.finalize()
WARNING: Please consider reporting this to the maintainers of org.codehaus.groovy.reflection.CachedClass
WARNING: Use --illegal-access=warn to enable warnings of further illegal reflective access operations
WARNING: All illegal access operations will be denied in a future release
2023/02/03 11:52:19.942 INFO [AddTableCommand] [main] Executing command: AddTable -tableConfigFile /opt/pinot/volume_app/dirrtl/dirrtl-table-realtime.json -offlineTableConfigFile null -realtimeTableConfigFilenull -schemaFile /opt/pinot/volume_app/dirrtl/dirrtl-schema.json -controllerProtocol http -controllerHost 172.19.0.3 -controllerPort 9000 -user null -password [hidden] -exec
2023/02/03 11:52:20.696 INFO [AddTableCommand] [main] {"code":500,"error":"org.apache.pinot.shaded.org.apache.kafka.common.KafkaException: Failed to construct kafka consumer"}

```

**Table JSON**

```
{
"tableName": "schema-dirrtl",
"tableType": "REALTIME",
"tenants": {
"broker": "DefaultTenant",
"server": "DefaultTenant",
"tagOverrideConfig": {}
},
"segmentsConfig": {
"schemaName": "schema-dirrtl",
"timeColumnName": "UPDT_TMSTM",
"replication": "1",
"replicasPerPartition": "1",
"retentionTimeUnit": null,
"retentionTimeValue": null,
"completionConfig": null,
"crypterClassName": null,
"peerSegmentDownloadScheme": null
},
"tableIndexConfig": {
"loadMode": "MMAP",
"invertedIndexColumns": [],
"createInvertedIndexDuringSegmentGeneration": false,
"rangeIndexColumns": [],
"sortedColumn": [],
"bloomFilterColumns": [],
"bloomFilterConfigs": null,
"noDictionaryColumns": [],
"onHeapDictionaryColumns": [],
"varLengthDictionaryColumns": [],
"enableDefaultStarTree": false,
"starTreeIndexConfigs": null,
"enableDynamicStarTreeCreation": false,
"segmentPartitionConfig": null,
"columnMinMaxValueGeneratorMode": null,
"aggregateMetrics": false,
"nullHandlingEnabled": false,
"streamConfigs": {
"streamType": "kafka",
"stream.kafka.topic.name": "mytopic",
"stream.kafka.broker.list": "kafka-2-client.prod1.gke.kaas-prod-us.gcp.extscloud.com:16000",
"stream.kafka.consumer.type": "lowlevel",
"stream.kafka.consumer.prop.auto.offset.reset": "smallest",
"stream.kafka.consumer.factory.class.name": "org.apache.pinot.plugin.stream.kafka20.KafkaConsumerFactory",
"stream.kafka.decoder.class.name": "org.apache.pinot.plugin.stream.kafka.KafkaAvroMessageDecoder",
"realtime.segment.flush.threshold.rows": "0",
"realtime.segment.flush.threshold.time": "24h",
"realtime.segment.flush.threshold.segment.size": "100M",
"security.protocol": "SSL",
"sasl.mechanism": "PLAN",
"ssl.keystore.location": "/opt/pinot/volume_app/dirrtl/keystore.jks",
"ssl.keystore.password": "**********",
"ssl.truststore.location": "/opt/pinot/volume_app/dirrtl/truststore.jks",
"ssl.truststore.password": "**********",
"ssl.key.password": "****************",
"ssl.client.auth": "required",
"stream.kafka.decoder.prop.schema.registry.rest.url": "http://kaas-test-schema-registry-a.company.com/ "
}
},
"metadata": {},
"ingestionConfig": {
"filterConfig": null,
"transformConfigs": []
},
"quota": {
"storage": null,
"maxQueriesPerSecond": null
},
"task": null,
"routing": {
"segmentPrunerTypes": null,
"instanceSelectorType": null
},
"query": {
"timeoutMs": null
},
"fieldConfigList": null,
"upsertConfig": null,
"tierConfigs": null
}

```

Start with the AddTable command and the streamConfigs in the supplied realtime table JSON, then reproduce the Kafka SSL setup while collecting the full consumer exception beyond the summary error. Compare the SSL, SASL, broker, keystore, truststore, and schema-registry settings with Pinot's Kafka ingestion documentation; done means AddTable succeeds and the realtime consumer connects to the topic.

Written by the indexing model from the issue text.