Hello Team! I tried to ingest the data from multiple kafka topics to single table. Is this supported in Confluent kafka? I got below error when I followed this document:

2025/07/30 13:27:36.981 ERROR [RealtimeSegmentDataManager_order_events_test__10001__0__20250730T1123Z] [order_events_test__10001__0__20250730T1123Z] Exception while in work
org.apache.kafka.common.KafkaException: Failed to construct kafka consumer
	at org.apache.kafka.clients.consumer.KafkaConsumer.<init>(KafkaConsumer.java:823) ~[pinot-all-1.3.0-jar-with-dependencies.jar:1.3.0-c0023da298126af6a01b802a04b66da34ba16134]
	at org.apache.kafka.clients.consumer.KafkaConsumer.<init>(KafkaConsumer.java:665) ~[pinot-all-1.3.0-jar-with-dependencies.jar:1.3.0-c0023da298126af6a01b802a04b66da34ba16134]
	at org.apache.kafka.clients.consumer.KafkaConsumer.<init>(KafkaConsumer.java:646) ~[pinot-all-1.3.0-jar-with-dependencies.jar:1.3.0-c0023da298126af6a01b802a04b66da34ba16134]
	at org.apache.kafka.clients.consumer.KafkaConsumer.<init>(KafkaConsumer.java:626) ~[pinot-all-1.3.0-jar-with-dependencies.jar:1.3.0-c0023da298126af6a01b802a04b66da34ba16134]
	at org.apache.pinot.plugin.stream.kafka20.KafkaPartitionLevelConnectionHandler.lambda$createConsumer$0(KafkaPartitionLevelConnectionHandler.java:86) ~[pinot-all-1.3.0-jar-with-dependencies.jar:1.3.0-c0023da298126af6a01b802a04b66da34ba16134]
	at org.apache.pinot.plugin.stream.kafka20.KafkaPartitionLevelConnectionHandler.retry(KafkaPartitionLevelConnectionHandler.java:100) ~[pinot-all-1.3.0-jar-with-dependencies.jar:1.3.0-c0023da298126af6a01b802a04b66da34ba16134]
	at org.apache.pinot.plugin.stream.kafka20.KafkaPartitionLevelConnectionHandler.createConsumer(KafkaPartitionLevelConnectionHandler.java:86) ~[pinot-all-1.3.0-jar-with-dependencies.jar:1.3.0-c0023da298126af6a01b802a04b66da34ba16134]
	at org.apache.pinot.plugin.stream.kafka20.KafkaPartitionLevelConnectionHandler.<init>(KafkaPartitionLevelConnectionHandler.java:67) ~[pinot-all-1.3.0-jar-with-dependencies.jar:1.3.0-c0023da298126af6a01b802a04b66da34ba16134]
	at org.apache.pinot.plugin.stream.kafka20.KafkaPartitionLevelConsumer.<init>(KafkaPartitionLevelConsumer.java:52) ~[pinot-all-1.3.0-jar-with-dependencies.jar:1.3.0-c0023da298126af6a01b802a04b66da34ba16134]
	at org.apache.pinot.plugin.stream.kafka20.KafkaConsumerFactory.createPartitionGroupConsumer(KafkaConsumerFactory.java:51) ~[pinot-all-1.3.0-jar-with-dependencies.jar:1.3.0-c0023da298126af6a01b802a04b66da34ba16134]
	at org.apache.pinot.core.data.manager.realtime.RealtimeSegmentDataManager.recreateStreamConsumer(RealtimeSegmentDataManager.java:1830) ~[pinot-all-1.3.0-jar-with-dependencies.jar:1.3.0-c0023da298126af6a01b802a04b66da34ba16134]
	at org.apache.pinot.core.data.manager.realtime.RealtimeSegmentDataManager.consumeLoop(RealtimeSegmentDataManager.java:529) ~[pinot-all-1.3.0-jar-with-dependencies.jar:1.3.0-c0023da298126af6a01b802a04b66da34ba16134]
	at org.apache.pinot.core.data.manager.realtime.RealtimeSegmentDataManager$PartitionConsumer.run(RealtimeSegmentDataManager.java:765) ~[pinot-all-1.3.0-jar-with-dependencies.jar:1.3.0-c0023da298126af6a01b802a04b66da34ba16134]
	at java.base/java.lang.Thread.run(Thread.java:1583) [?:?]
Caused by: org.apache.kafka.common.config.ConfigException: No resolvable bootstrap urls given in bootstrap.servers
	at org.apache.kafka.clients.ClientUtils.parseAndValidateAddresses(ClientUtils.java:89) ~[pinot-all-1.3.0-jar-with-dependencies.jar:1.3.0-c0023da298126af6a01b802a04b66da34ba16134]
	at org.apache.kafka.clients.ClientUtils.parseAndValidateAddresses(ClientUtils.java:48) ~[pinot-all-1.3.0-jar-with-dependencies.jar:1.3.0-c0023da298126af6a01b802a04b66da34ba16134]
	at org.apache.kafka.clients.consumer.KafkaConsumer.<init>(KafkaConsumer.java:731) ~[pinot-all-1.3.0-jar-with-dependencies.jar:1.3.0-c0023da298126af6a01b802a04b66da34ba16134]
	... 13 more

Since I was able to ingest data from the first topic, I don't think the broker URL is the issue. Is there anything I might be missing?? I set table config like this:

"streamConfigMaps": [
  {
    "streamType": "kafka",
    "stream.kafka.consumer.prop.auto.offset.reset": "smallest",
    "stream.kafka.consumer.factory.class.name": "org.apache.pinot.plugin.stream.kafka20.KafkaConsumerFactory",
    "stream.kafka.broker.list": "{BROKER}:9092",
    "stream.kafka.decoder.class.name": "org.apache.pinot.plugin.inputformat.protobuf.KafkaConfluentSchemaRegistryProtoBufMessageDecoder",
    "stream.kafka.decoder.prop.schema.registry.rest.url": "https://{REGISTRY}}",
    "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>": "{USER_INFO}",
    "stream.kafka.consumer.type": "LOWLEVEL",
    "security.protocol": "SASL_SSL",
    "sasl.mechanism": "PLAIN",
    "sasl.jaas.config": "{SASL}",
    "realtime.segment.flush.threshold.rows": "500000",
    "realtime.segment.flush.autotune.initialRows": "500000",
    "stream.kafka.topic.name": "{TOPIC_NAME1}"
  },
  {
    "streamType": "kafka",
    "stream.kafka.consumer.prop.auto.offset.reset": "smallest",
    "stream.kafka.consumer.factory.class.name": "org.apache.pinot.plugin.stream.kafka20.KafkaConsumerFactory",
    "stream.kafka.broker.list": "{BROKER}:9092",
    "stream.kafka.decoder.class.name": "org.apache.pinot.plugin.inputformat.protobuf.KafkaConfluentSchemaRegistryProtoBufMessageDecoder",
    "stream.kafka.decoder.prop.schema.registry.rest.url": "https://{REGISTRY}}",
    "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>": "{USER_INFO}",
    "stream.kafka.consumer.type": "LOWLEVEL",
    "security.protocol": "SASL_SSL",
    "sasl.mechanism": "PLAIN",
    "sasl.jaas.config": "{SASL}",
    "realtime.segment.flush.threshold.rows": "500000",
    "realtime.segment.flush.autotune.initialRows": "500000",
    "stream.kafka.topic.name": "{TOPIC_NAME2}"
  }
],

@Chris Peck @Qiaochu Liu any idea ^^?

cc: @X G

which version of Pinot are you using?

version 1.3 @X G

Does pinot support multi topic data injection

Yes

There was one change we've added after 1.3 as a bugfix. Can you try cherry picking this? @박민지 https://github.com/apache/pinot/pull/15094/files

Ah, I installed it via homebrew, which gave me the binary version from this. I’ll download the source, cherry pick and build it. Thanks a lot! @X G