Hello everyone, Im struggling to create a Kafka stream ingestion with HLC consumer and a custom consumer group id. Anyone has worked on a similar case ?

Hi can you share you table config and schema. Please remove any secrets from it if there

Don’t worry, I’m using helm with env var config secret

{
  "tableName": "contactEvents",
  "tableType": "REALTIME",
  "segmentsConfig": {
    "timeColumnName": "created_at",
    "timeType": "DAYS",
    "retentionTimeUnit": "DAYS",
    "retentionTimeValue": "3650",
    "segmentPushType": "APPEND",
    "segmentAssignmentStrategy": "BalanceNumSegmentAssignmentStrategy",
    "schemaName": "contactEvents",
    "replication": "1",
    "replicasPerPartition": "1"
  },
  "tenants": {
    "broker": "DefaultTenant",
    "server": "DefaultTenant",
    "tagOverrideConfig": {}
  },
  "tableIndexConfig": {
    "loadMode": "MMAP",
    "streamConfigs": {
      "streamType": "kafka",
      "stream.kafka.consumer.type": "highlevel",
      "stream.kafka.topic.name": "united-pg.public.contact_events",
      "stream.kafka.consumer.factory.class.name": "org.apache.pinot.plugin.stream.kafka20.KafkaConsumerFactory",
      "stream.kafka.decoder.class.name": "org.apache.pinot.plugin.stream.kafka.KafkaJSONMessageDecoder",
      "stream.kafka.broker.list": "strimzi-kafka-bootstrap.strimzi.svc.cluster.local:9093",
      "stream.kafka.hlc.bootstrap.server": "strimzi-kafka-bootstrap.strimzi.svc.cluster.local:9093",
      "stream.kafka.hlc.zk.connect.string": "strimzi-zookeeper-client.strimzi.svc.cluster.local:2181",
      "stream.kafka.consumer.prop.auto.offset.reset": "earliest",
      "stream.kafka.hlc.group.id": "pinot",
      "security.protocol": "SSL",
      "ssl.truststore.type": "PKCS12",
      "ssl.keystore.type": "PKCS12",
      "ssl.truststore.location": "/ssl/ca.p12",
      "ssl.keystore.location": "/ssl/strimzi-user-pinot.p12",
      "ssl.truststore.password": "${SSL_TRUSTSTORE_PASSWORD}",
      "ssl.keystore.password": "${SSL_KEYSTORE_PASSWORD}"
    },
    "bloomFilterColumns": [
      "tenant_id",
      "contact_id",
      "content_id"
    ]
  },
  "metadata": {
    "customConfigs": {}
  }
}

Yes, that is the cause of the issue. Will it be possible for you to use

lowlevel

consumer? I am meanwhile looking for a fix. I guess just increasing the groupId to a bigger string should fix it but need to verify.

In my case, I really need a consumer group so I can’t use

lowlevel

consumer type right now 😕 I will try with a bigger group id (

apachepinot

instead and let you know)

The error seems to be in

PinotTableIdealStateBuilder

class. Working on alternative/fix.

why do you particularly need a consumer group? HLC has been deprecated.. here’s some reasons why we did that: https://www.confluent.io/resources/kafka-summit-2020/apache-pinot-case-study-building-distributed-analytics-systems-using-apache-kafka/

I need It to keep control on the offset and track the lag on my broker

there’s a separate effort ongoing to be able to track lag between kafka consumer and broker offset: https://github.com/apache/pinot/pull/8280

Because, If you specify a consumer group id, you need to have authorization for a user to access this group id and perform offset commit , read ,etc But if you specify a

null

group id, you will have no authorization error. After some search Kafka has change the behavior of group id

""

and group id

null

https://issues.apache.org/jira/browse/KAFKA-6774

can you use one of these methods? https://docs.pinot.apache.org/basics/data-import/pinot-stream-ingestion/import-from-apache-kafka#some-more-kafka-ingestion-configs

Yes I already did It and It’s working