Hi, I'm using Pinot 0.10.0 I try to setup a REALTIME table to stream Kafka topic messages to a Pinot Table. I'm not an expert on Pinot and it's the first time I setup a REALTIME table so be kind 😉 . When I create the table I see a NPE in the Controller log and the table is not created (in fact it seems to be partialy created as it's not shown in the UI but trying to re-create it fail with table already exists).

pinot-controller  | 2022/09/06 15:11:30.283 ERROR [PinotTableIdealStateBuilder] [grizzly-http-server-1] Could not get PartitionGroupMetadata for topic: my-topic of table: my-topic_REALTIME
pinot-controller  | java.lang.NullPointerException: null
pinot-controller  |     at org.apache.pinot.plugin.stream.kafka20.KafkaStreamMetadataProvider.fetchPartitionCount(KafkaStreamMetadataProvider.java:48) ~[pinot-kafka-2.0-0.10.0-shaded.jar:0.10.0-30c4635bfeee88f88aa9c9f63b93bcd4a650607f]
pinot-controller  |     at org.apache.pinot.spi.stream.StreamMetadataProvider.computePartitionGroupMetadata(StreamMetadataProvider.java:66) ~[pinot-all-0.10.0-jar-with-dependencies.jar:0.10.0-30c4635bfeee88f88aa9c9f63b93bcd4a650607f]

For me NPE is always suspicious (a bug). Here is my table config.

{
  "tableName": "my-topic",
  "tableType": "REALTIME",
  "segmentsConfig": {
    "timeColumnName": "timestamp",
    "timeType": "MILLISECONDS",
    "schemaName": "my-topic",
    "replicasPerPartition": "1"
  },
  "tenants": {},
  "tableIndexConfig": {
    "loadMode": "MMAP",
    "streamConfigs": {
      "streamType": "kafka",
      "stream.kafka.consumer.type": "LowLevel",
      "stream.kafka.topic.name": "my-topic",
      "stream.kafka.decoder.class.name": "org.apache.pinot.plugin.inputformat.avro.confluent.KafkaConfluentSchemaRegistryAvroMessageDecoder",
      "stream.kafka.consumer.factory.class.name": "org.apache.pinot.plugin.stream.kafka20.KafkaConsumerFactory",
      "stream.kafka.broker.list": "my-broker:9092",
      "stream.kafka.consumer.prop.auto.offset.reset": "smallest",
      "stream.kafka.consumer.prop.schema.registry.url": "<https://my-registry:8000>",
      "<http://stream.kafka.consumer.prop.basic.auth.user.info|stream.kafka.consumer.prop.basic.auth.user.info>": "<redacted>",
      "security.protocol": "SSL",
      "ssl.truststore.type": "PEM",
      "ssl.truststore.certificates": "<redacted>",
      "ssl.keystore.type": "PEM",
      "ssl.keystore.certificate.chain": "<redacted>",
      "ssl.keystore.key": "<redacted>"
    }
  },
  "metadata": {
    "customConfigs": {}
  }
}

Yes, NPE is definitely not a desired behavior @Haitao Zhang could you take a look? (PS - we are always kind here 😊).

Thanks, I fear the NPE hides the real issue here so I don't know how to debug that

The NPE is because it is unable to fetch the partition info from stream provider.

@Loïc Mathieu Just want to double confirm, does the topic exist? This might be the root cause for the NPE

Yes, it exists, I double checked it. And the user authenticated by the private key has access to it. And there is no SSL issue (there was previously so I know that now my SSL setup is correct).

The NPE is thrown by this line (https://github.com/apache/pinot/blob/master/pinot-plugins/pinot-stream-ingestion/p[…]he/pinot/plugin/stream/kafka20/KafkaStreamMetadataProvider.java, this is the latest master branch, the version you use has this on line 48). What I can think of is either

_consumer

is null or

partitionsFor

returns null. But _consumer is not null, otherwise, during initialization, this line (https://github.com/apache/pinot/blob/master/pinot-plugins/pinot-stream-ingestion/p[…]plugin/stream/kafka20/KafkaPartitionLevelConnectionHandler.java) throws exception. Therefore, I feel

partitionsFor

returns null.

@Haitao Zhang yes, I can write some simple code to try to understand what's going on

Thanks for sharing