Hello Everyone.. i am trying to inject data from kafka topic to apache pinot but i didn't see any data loaded do i am missing anything in config related to avro ? Schema

{
  "schemaName": "test_schema",
  "dimensionFieldSpecs": [
    {
      "name": "client_id",
      "dataType": "STRING"
    },
    {
      "name": "master_property_id",
      "dataType": "INT"
    },
    {
      "name": "business_unit",
      "dataType": "STRING"
    },
    {
      "name": "error_info_str",
      "dataType": "STRING"
    }
  ],
  "dateTimeFieldSpecs": [
    {
      "name": "timestamp",
      "dataType": "LONG",
      "format": "1:MILLISECONDS:EPOCH",
      "granularity": "1:MILLISECONDS"
    }
  ]
}

Table:

{
  "REALTIME": {
    "tableName": "test_schema_REALTIME",
    "tableType": "REALTIME",
    "segmentsConfig": {
      "schemaName": "test_schema",
      "replication": "1",
      "replicasPerPartition": "1",
      "timeColumnName": "timestamp"
    },
    "tenants": {
      "broker": "DefaultTenant",
      "server": "DefaultTenant",
      "tagOverrideConfig": {}
    },
    "tableIndexConfig": {
      "bloomFilterColumns": [],
      "noDictionaryColumns": [],
      "onHeapDictionaryColumns": [],
      "varLengthDictionaryColumns": [],
      "enableDefaultStarTree": false,
      "enableDynamicStarTreeCreation": false,
      "aggregateMetrics": false,
      "nullHandlingEnabled": false,
      "invertedIndexColumns": [],
      "rangeIndexColumns": [],
      "autoGeneratedInvertedIndex": false,
      "createInvertedIndexDuringSegmentGeneration": false,
      "sortedColumn": [],
      "loadMode": "MMAP",
      "streamConfigs": {
        "streamType": "kafka",
        "stream.kafka.topic.name": "TestTopic",
        "stream.kafka.broker.list": "localhost:9092",
        "stream.kafka.consumer.type": "lowlevel",
        "stream.kafka.consumer.prop.auto.offset.reset": "smallest",
        "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",
        "schema.registry.url": "<http://localhost:8081>",
        "realtime.segment.flush.threshold.rows": "0",
        "realtime.segment.flush.threshold.time": "24h",
        "realtime.segment.flush.segment.size": "100M"
      }
    },
    "metadata": {},
    "quota": {},
    "routing": {},
    "query": {},
    "ingestionConfig": {
      "transformConfigs": [
        {
          "columnName": "error_info_str",
          "transformFunction": "json_format(error_info)"
        }
      ]
    },
    "isDimTable": false
  }
}

Kafka Avro Schema:

{
  "type": "record",
  "name": "TestRecord",
  "namespace": "com.test.ns",
  "fields": [
    {
      "name": "client_id",
      "type": [
        "null",
        "string"
      ]
    },
    {
      "name": "master_property_id",
      "type": "int"
    },
    {
      "name": "business_unit",
      "type": [
        "null",
        "string"
      ]
    },
    {
      "name": "error_info",
      "type": {
        "type": "record",
        "name": "ErrorInfo",
        "fields": [
          {
            "name": "code",
            "type": [
              "null",
              "string"
            ]
          },
          {
            "name": "description",
            "type": [
              "null",
              "string"
            ]
          }
        ]
      }
    },
    {
      "name": "timestamp",
      "type": [
        "null",
        "long"
      ],
      "default": null
    }
  ]
}

is there anything else in the log?

its working now i missed one configuration

"stream.kafka.decoder.prop.schema.registry.rest.url":

Cool

@User just a quick point correct me if i am wrong.. we have to add one timestamp field in schema that when creating realtime table that timestamp will be use by retention of data etc..? in my case lets say i don't have any timestamp in kafka avro schema but i added a field as

timestamp

in apache pinot schema will it pick a default value automatically?

No.. you can probably use a udf to set it to now()

i was testing the same scenario and realised it take some negative value not sure what it indicate in time of timestamp

-9223372036854776000

now() do i need to use this in transform function?

Yes

its throwing error if i add now() in transform function for timestamp. Invalid table config: test_schema_now_REALTIME. Invalid transform function

'now()'

for column '`timestamp`'