https://pinot.apache.org/ logo
Join Slack
Powered by
# pinot-dev
  • l
    do I need to supply any other config than that one here?
    Copy code
    "streamType": "kafka",
      "stream.kafka.consumer.type": "lowlevel",
     "stream.kafka.topic.name": "src-app-users-json",
     "stream.kafka.decoder.class.name": "org.apache.pinot.plugin.stream.kafka.KafkaJSONMessageDecoder",
     "stream.kafka.consumer.factory.class.name": "org.apache.pinot.plugin.stream.kafka20.KafkaConsumerFactory",
     "stream.kafka.broker.list": "127.0.0.1:9094",
     "realtime.segment.flush.threshold.time": "3600000",
     "realtime.segment.flush.threshold.size": "50000",
     "stream.kafka.consumer.prop.auto.offset.reset": "smallest"
  • n
    no, just change the hostname in this 
    "stream.kafka.broker.list": "127.0.0.1:9094",
    to whatever your kafka broker process is named as
  • l
    perfect checking that now, we're super close!
  • n
    here’s an example for you: https://docs.pinot.apache.org/basics/recipes/github-events-stream#create-a-kafka-topic . This table config uses “kafka:9092” because kafka process was called “kafka”
  • l
    Copy code
    2021/01/20 23:26:15.116 ERROR [PinotTableRestletResource] [grizzly-http-server-0] Failed to fetch the offset for topic: src-app-users-json, partition: 0 with criteria: OffsetCriteria{_offsetType=SMALLEST, _offsetString='smallest'}
java.lang.IllegalStateException: Failed to fetch the offset for topic: src-app-users-json, partition: 0 with criteria: OffsetCriteria{_offsetType=SMALLEST, _offsetString='smallest'}
	at org.apache.pinot.controller.helix.core.realtime.PinotLLCRealtimeSegmentManager.getPartitionOffset(PinotLLCRealtimeSegmentManager.java:643) ~[pinot-all-0.7.0-SNAPSHOT-root@pinot-controller-0:/opt/pinot#
  • l
    in this case @User I'm using ClusterIP
    Copy code
    "stream.kafka.broker.list": "edh-kafka-0.ingestion.svc.cluster.local:9094"
  • n
    huh interesting. At least now it’s connecting to kafka.
  • n
    anything else in logs?
  • n
    can you try with “stream.kafka.consumer.prop.auto.offset.reset”: “largest” instead of smallest?
  • l
    actually I think it's my mistake!!! this is the correct address of kafka let me test now
    Copy code
    edh-kafka-brokers.ingestion.svc.Cluster.local:9092
  • l
    boom seems that is rolling! 🙂
    Copy code
    2021/01/20 23:30:03.665 INFO [AssignableInstanceManager] [HelixController-pipeline-task-pinot-(0592e5c6_TASK)] AssignableInstanceManager built AssignableInstances from scratch based on contexts in TaskDataCache due to Controller switch or ClusterConfig change.
2021/01/20 23:30:03.665 INFO [AssignableInstanceManager] [HelixController-pipeline-task-pinot-(0592e5c6_TASK)] Current quota capacity: {"Server_pinot-server-0.pinot-server-headless.datastore.svc.cluster.local_8098":{"TASK_EXEC_THREAD":{"DEFAULT":"0/40"}},"Controller_pinot-controller-0.pinot-controller-headless.datastore.svc.cluster.local_9000":{"TASK_EXEC_THREAD":{"DEFAULT":"0/40"}},"Minion_pinot-minion-0.pinot-minion-headless.datastore.svc.cluster.local_9514":{"TASK_EXEC_THREAD":{"DEFAULT":"0/40"}},"Broker_pinot-broker-0.pinot-broker-headless.datastore.svc.cluster.local_8099":{"TASK_EXEC_THREAD":{"DEFAULT":"0/40"}}}
2021/01/20 23:30:03.665 INFO [WorkflowControllerDataProvider] [HelixController-pipeline-task-pinot-(0592e5c6_TASK)] Event 0592e5c6_TASK : END: WorkflowControllerDataProvider.refresh() for cluster pinot, started at 1611185403620 took 45 for TASK pipeline
2021/01/20 23:30:03.665 INFO [Pipeline] [HelixController-pipeline-task-pinot-(0592e5c6_TASK)] END ReadClusterDataStage for TASK pipeline for cluster pinot. took: 45 ms for event 0592e5c6_TASK
2021/01/20 23:30:03.665 INFO [Pipeline] [HelixController-pipeline-task-pinot-(0592e5c6_TASK)] END ResourceComputationStage for TASK pipeline for cluster pinot. took: 0 ms for event 0592e5c6_TASK
2021/01/20 23:30:03.665 INFO [Pipeline] [HelixController-pipeline-task-pinot-(0592e5c6_TASK)] END ResourceValidationStage for TASK pipeline for cluster pinot. took: 0 ms for event 0592e5c6_TASK
2021/01/20 23:30:03.665 INFO [Pipeline] [HelixController-pipeline-task-pinot-(0592e5c6_TASK)] END CurrentStateComputationStage for TASK pipeline for cluster pinot. took: 0 ms for event 0592e5c6_TASK
2021/01/20 23:30:03.665 INFO [Pipeline] [HelixController-pipeline-task-pinot-(0592e5c6_TASK)] END TaskSchedulingStage for TASK pipeline for cluster pinot. took: 0 ms for event 0592e5c6_TASK
2021/01/20 23:30:03.665 INFO [TaskPersistDataStage] [HelixController-pipeline-task-pinot-(0592e5c6_TASK)] START TaskPersistDataStage.process()
2021/01/20 23:30:03.665 INFO [TaskPersistDataStage] [HelixController-pipeline-task-pinot-(0592e5c6_TASK)] END TaskPersistDataStage.process() for cluster pinot took 0 ms
2021/01/20 23:30:03.665 INFO [Pipeline] [HelixController-pipeline-task-pinot-(0592e5c6_TASK)] END TaskPersistDataStage for TASK pipeline for cluster pinot. took: 0 ms for event 0592e5c6_TASK
2021/01/20 23:30:03.666 INFO [AbstractAsyncBaseStage] [TaskJobPurgeWorker-pinot] START AsyncProcess: TASK::TaskGarbageCollectionStage
  • n
    great!
  • l
    woot woot @User kudos if was not you I've sure that I would spend countless hours on that, super appreciate your effort and patience
    🙂 1
  • l
    jackpot! 🙂
  • l
    last question now that you're here @User is easy to use avro? in this case work with avro topics integrated with Schema Registry
  • l
    is there any place on the docos where you can point me for this integration?
  • n
    yes it is easy to use avro. I believe many of the companies are using the AvroDecoders and schema registry setup
  • n
    let me look if we have a recipe or that
  • n
    cannot find an end to end integration recipe, but this sample table shows avro and schema related properties: https://docs.pinot.apache.org/basics/data-import/pinot-stream-ingestion/import-from-apache-kafka#some-more-kafka-ingestion-configs
  • n
    we can also connect you to folks who are using that
  • l
    @User I would love that, if you can forward a contact from someone using Schema Registry we will love it
  • n
    @User @User would you be able to help ^^ ?
  • a
    I get this error while running Quickstart.java (straight out of box). Has anything changed here?
    Copy code
    TotalProcessed time for event: MessageChange took: 18 ms
Exception in thread "main" java.lang.RuntimeException: Failed to create IngestionJobRunner instance for class - org.apache.pinot.plugin.ingestion.batch.standalone.SegmentGenerationJobRunner
	at org.apache.pinot.spi.ingestion.batch.IngestionJobLauncher.kickoffIngestionJob(IngestionJobLauncher.java:137)
	at org.apache.pinot.spi.ingestion.batch.IngestionJobLauncher.runIngestionJob(IngestionJobLauncher.java:113)
	at org.apache.pinot.tools.BootstrapTableTool.bootstrapOfflineTable(BootstrapTableTool.java:189)
	at org.apache.pinot.tools.BootstrapTableTool.execute(BootstrapTableTool.java:99)
	at org.apache.pinot.tools.admin.command.QuickstartRunner.bootstrapTable(QuickstartRunner.java:207)
	at org.apache.pinot.tools.Quickstart.execute(Quickstart.java:180)
	at org.apache.pinot.tools.Quickstart.main(Quickstart.java:223)
Caused by: java.lang.ClassNotFoundException: org.apache.pinot.plugin.ingestion.batch.standalone.SegmentGenerationJobRunner
	at java.net.URLClassLoader.findClass(URLClassLoader.java:381)
	at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
	at org.apache.pinot.spi.plugin.PluginClassLoader.loadClass(PluginClassLoader.java:80)
	at org.apache.pinot.spi.plugin.PluginManager.createInstance(PluginManager.java:293)
	at org.apache.pinot.spi.plugin.PluginManager.createInstance(PluginManager.java:264)
	at org.apache.pinot.spi.plugin.PluginManager.createInstance(PluginManager.java:245)
	at org.apache.pinot.spi.ingestion.batch.IngestionJobLauncher.kickoffIngestionJob(IngestionJobLauncher.java:135)
	... 6 more
  • a
    @User I am wondering if this ^^ is related to https://github.com/apache/incubator-pinot/pull/6469 ?
  • x
    shouldn’t be, where you running this quickstart ?
  • a
    In IntelliJ, right click and run
  • x
    are you running in pinot-distribution directory
  • x
    hmm
  • x
    are you on master branch?
  • a
    yes and without any changes.
1...212223...30Latest