👋 Hi everyone! I am new to apache pinot, I am using it for realtime data ingestion from kafka topic, we are using confluent kafka and schema registry and avro schema. I am able to connect kafka topic as my table is successfully created and its in healthy state, but query is not showing any records. how can we check that it has some issues in consuming side. From swagger debug table api also I cannot see any errors.

Hi 🙂 Have you check server logs for any issues ?

where to check for server logs

It depends on the way you launch the server.

we launched it through helm

Did you get any error ?

Check the table debug api in swagger to see if it bubbles up any problems

@User no errors from table debug api,

[
  {
    "tableName": "tscalrecord15_REALTIME",
    "numSegments": 2,
    "numServers": 2,
    "numBrokers": 1,
    "segmentDebugInfos": [],
    "serverDebugInfos": [
      {
        "serverName": "Server_pinot-server-0.pinot-server-headless.pinot.svc.cluster.local_8098",
        "numMessages": 0,
        "errors": 0
      },
      {
        "serverName": "Server_pinot-server-1.pinot-server-headless.pinot.svc.cluster.local_8098",
        "numMessages": 0,
        "errors": 0
      }
    ],
    "brokerDebugInfos": [],
    "tableSize": {
      "reportedSize": "0 bytes",
      "estimatedSize": "0 bytes"
    },
    "ingestionStatus": {
      "ingestionState": "HEALTHY",
      "errorMessage": ""
    }
  }
]

cc: @User

@User error is [Consumer clientId=f83d920f-6e8d-465b-bd83-2552132f241e2, groupId=tscalrecord7_REALTIME_1647430665510_0] Bootstrap broker XXXXXX:9092 (id: -1 rack: null) disconnected as kafka consumer config showed in server logs is not taking sasl properties what we provided in table config. Below is kafka consumer config from server logs : If you check for below properties they are not as provided in table config.

sasl.mechanism = GSSAPI
        security.protocol = PLAINTEXT

for topic tSCalibrationRecord_pinot ConsumerConfig values:

allow.auto.create.topics = true
        <http://auto.commit.interval.ms|auto.commit.interval.ms> = 5000
        auto.offset.reset = latest
        bootstrap.servers = [XXXX]
        check.crcs = true
        client.dns.lookup = use_all_dns_ips
        client.id = consumer-tscalrecord14_REALTIME_1647432834089_0-10
        client.rack =
        <http://connections.max.idle.ms|connections.max.idle.ms> = 540000
        <http://default.api.timeout.ms|default.api.timeout.ms> = 60000
        enable.auto.commit = true
        exclude.internal.topics = true
        fetch.max.bytes = 52428800
        <http://fetch.max.wait.ms|fetch.max.wait.ms> = 500
        fetch.min.bytes = 1
        group.id = tscalrecord14_REALTIME_1647432834089_0
        group.instance.id = null
        <http://heartbeat.interval.ms|heartbeat.interval.ms> = 3000
        interceptor.classes = []
        internal.leave.group.on.close = true
        internal.throw.on.fetch.stable.offset.unsupported = false
        isolation.level = read_uncommitted
        key.deserializer = class org.apache.kafka.common.serialization.StringDeserializer
        max.partition.fetch.bytes = 1048576
        <http://max.poll.interval.ms|max.poll.interval.ms> = 300000
        max.poll.records = 500
        <http://metadata.max.age.ms|metadata.max.age.ms> = 300000
        metric.reporters = []
        metrics.num.samples = 2
        metrics.recording.level = INFO
        <http://metrics.sample.window.ms|metrics.sample.window.ms> = 30000
        partition.assignment.strategy = [class org.apache.kafka.clients.consumer.RangeAssignor]
        receive.buffer.bytes = 65536
        <http://reconnect.backoff.max.ms|reconnect.backoff.max.ms> = 1000
        <http://reconnect.backoff.ms|reconnect.backoff.ms> = 50
        <http://request.timeout.ms|request.timeout.ms> = 30000
        <http://retry.backoff.ms|retry.backoff.ms> = 100
        sasl.client.callback.handler.class = null
        sasl.jaas.config = null
        sasl.kerberos.kinit.cmd = /usr/bin/kinit
        sasl.kerberos.min.time.before.relogin = 60000
        sasl.kerberos.service.name = null
        sasl.kerberos.ticket.renew.jitter = 0.05
        sasl.kerberos.ticket.renew.window.factor = 0.8
        sasl.login.callback.handler.class = null
        sasl.login.class = null
        sasl.login.refresh.buffer.seconds = 300
        sasl.login.refresh.min.period.seconds = 60
        sasl.login.refresh.window.factor = 0.8
        sasl.login.refresh.window.jitter = 0.05
        sasl.mechanism = GSSAPI
        security.protocol = PLAINTEXT
        security.providers = null
        send.buffer.bytes = 131072
        <http://session.timeout.ms|session.timeout.ms> = 10000
        <http://socket.connection.setup.timeout.max.ms|socket.connection.setup.timeout.max.ms> = 30000
        <http://socket.connection.setup.timeout.ms|socket.connection.setup.timeout.ms> = 10000
        ssl.cipher.suites = null
        ssl.enabled.protocols = [TLSv1.2, TLSv1.3]
        ssl.endpoint.identification.algorithm = https
        ssl.engine.factory.class = null
        ssl.key.password = null
        ssl.keymanager.algorithm = SunX509
        ssl.keystore.certificate.chain = null
        ssl.keystore.key = null
        ssl.keystore.location = null
        ssl.keystore.password = null
        ssl.keystore.type = JKS
        ssl.protocol = TLSv1.3
        ssl.provider = null
        ssl.secure.random.implementation = null
        ssl.trustmanager.algorithm = PKIX
        ssl.truststore.certificates = null
        ssl.truststore.location = null
        ssl.truststore.password = null
        ssl.truststore.type = JKS
        value.deserializer = class org.apache.kafka.common.serialization.BytesDeserializer

The configuration 'auto.isolation.level' was supplied but isn't a known config. The configuration 'auto.commit.enable' was supplied but isn't a known config. Kafka version: 2.8.1

Just a spot check, I see you’re using high level consumer. How about try low level consumer? And check out the config example noted here: https://docs.pinot.apache.org/basics/data-import/pinot-stream-ingestion/import-from-apache-kafka#some-more-kafka-ingestion-configs which says ‘Use Kafka Partition(Low) Level Consumer with SSL’.

@User thank you for checking it quickly, I have tried with above link also, we are using confluent kafka , we are not using key store and other ssl key path, using basic auth(username and password) we are providing SASL_SSL as protocol. Low level consumer is always giving time out.

cc @User as I see you’re helping on another SASL_SSL related issue and in case you could spot any pattern here.

security protocol and other related configs need to be given directly, without the stream.kafka prefix: https://docs.pinot.apache.org/basics/data-import/pinot-stream-ingestion/import-from-apache-kafka#ingesting-streaming-data

Thanks for the heads up!

hi @User I am able to resolve this , will share table config in a while, you can take a look at it. many many thanks pinot team to take this up quickly and for supporting: @User, @User @User, @User

that’s great! yes please, we’ll put streamConfigs from your config as a reference for sasl_ssl in the docs, thank you 🙂

Thanks @User

hi @User, @User, Please find the below working config for connecting sasl_ssl :

{
  "tableName": "calrecord",
  "tableType": "REALTIME",
  "segmentsConfig": {
    "schemaName": "calrec",
    "timeColumnName": "fixTimestamp",
    "allowNullTimeValue": false,
    "replication": "1",
    "replicasPerPartition": "1",
    "completionConfig": {
      "completionMode": "DOWNLOAD"
    }
  },
  "tenants": {
    "broker": "DefaultTenant",
    "server": "DefaultTenant",
    "tagOverrideConfig": {}
  },
  "tableIndexConfig": {
    "invertedIndexColumns": [],
    "rangeIndexColumns": [],
    "rangeIndexVersion": 1,
    "autoGeneratedInvertedIndex": false,
    "createInvertedIndexDuringSegmentGeneration": false,
    "sortedColumn": [],
    "bloomFilterColumns": [],
    "loadMode": "MMAP",
    "streamConfigs": {
      "streamType": "kafka",
      "stream.kafka.topic.name": "tscrpinot",
      "stream.kafka.broker.list": "xxxx:9092",
      "stream.kafka.consumer.type": "lowlevel",
      "stream.kafka.consumer.prop.auto.offset.reset": "largest",
      "stream.kafka.schema.registry.url": "<https://xxx.cloud>",
      "stream.kafka.consumer.factory.class.name": "org.apache.pinot.plugin.stream.kafka20.KafkaConsumerFactory",
      "stream.kafka.decoder.prop.schema.registry.rest.url": "<https://xxx.cloud>",
      "sasl.mechanism": "PLAIN",
      "stream.kafka.decoder.prop.basic.auth.credentials.source": "USER_INFO",
      "<http://stream.kafka.decoder.prop.schema.registry.basic.auth.user.info|stream.kafka.decoder.prop.schema.registry.basic.auth.user.info>": "schema_registry_username:schema_registry_password",
      "security.protocol": "SASL_SSL",
      "sasl.jaas.config": "org.apache.kafka.common.security.scram.ScramLoginModule required username=\"kafkausername\" password=\"kafkapassword\";",
      "stream.kafka.decoder.class.name": "org.apache.pinot.plugin.inputformat.avro.confluent.KafkaConfluentSchemaRegistryAvroMessageDecoder",
      "realtime.segment.flush.threshold.rows": "0",
      "realtime.segment.flush.threshold.size": "0",
      "realtime.segment.flush.threshold.time": "24h",
      "realtime.segment.flush.autotune.initialRows": "3000000",
      "realtime.segment.flush.threshold.segment.size": "500M"
    },
    "onHeapDictionaryColumns": [],
    "varLengthDictionaryColumns": [],
    "enableDefaultStarTree": false,
    "enableDynamicStarTreeCreation": false,
    "aggregateMetrics": false,
    "nullHandlingEnabled": false
  },
  "metadata": {},
  "quota": {},
  "routing": {
    "instanceSelectorType": "strictReplicaGroup"
  },
  "query": {},
  "ingestionConfig": {},
  "isDimTable": false
}

thank you, added here: https://docs.pinot.apache.org/basics/data-import/pinot-stream-ingestion/import-from-apache-kafka#use-kafka-level-consumer-with-sasl_ssl

Thanks @User @User