@User yes, I think this is a bug. I guess the solution is to not honour the shardIteratorType in config if a valid offset to consume the records is avaliable.

It actually does work as the moment it consumes the latest message the iterator age drops to 0.I am not aware how the iterage age is supposed to reflect when the shard iterator is AT_SEQUENCE_NUMBER.

@User @User our GetIterator age metric in cloudwatch for the kinesis stream went out of sync ever since we made a Pinot 9.1 release fix last month.Any ideas what caused it ?

<!here> We have noticed an increased in read throughput in the past few months and wanted to increase the number of shards.Wanted to know whether you recommend doubling the provisioned shards or using On demand provisioning.I was trying to find documentation on on demand support but did not find ? Should we just double or quadruple the shards and will that need restart or does pinot support on demand provisioning

it will not need restart. you can just increase based on your increased event rate. pinot will automatically detect once all parent shards are read

so I can use On demand provisioning in kinesis ? I am using 0.9.1

By on demand provisioning if you mean auto scaling from kinesis side, then yes

@Seunghyun @Mayank for fyi

Thanks @Neha Pawar will try it out

Hello Team, I'm pushing data from spark streaming into Kinesis streams and Pinot is ingesting the data from kinesis. In order to know the latency from spark to pinot I'm appending an epoch timestamp before writing to kinesis stream and in the Pinot's table config I'm adding another field with transformation function now() When I query the table and check the latency using the difference between spark write time and pinot's row creation time, I get the average latency around 9-10 seconds So, I want to know if we can reduce the latency here by making changes to the configurations table config:

{
  "tableName": "backend_pre_processed",
  "tableType": "REALTIME",
  "segmentsConfig": {
    "schemaName": "backend_pre_processed",
    "retentionTimeUnit": "DAYS",
    "retentionTimeValue": "90",
    "replication": "1",
    "replicasPerPartition": "1",
    "timeColumnName": "spark_write_timestamp",
    "minimizeDataMovement": false
  },
  "tenants": {
    "broker": "DefaultTenant",
    "server": "DefaultTenant",
    "tagOverrideConfig": {}
  },
  "tableIndexConfig": {
    "invertedIndexColumns": [],
    "noDictionaryColumns": [],
    "autoGeneratedInvertedIndex": false,
    "createInvertedIndexDuringSegmentGeneration": false,
    "sortedColumn": [],
    "bloomFilterColumns": [],
    "loadMode": "MMAP",
    "streamConfigs": {
      "streamType": "kinesis",
      "stream.kinesis.topic.name": "backend-processed-events",
      "stream.kinesis.consumer.type": "lowlevel",
      "stream.kinesis.decoder.class.name": "org.apache.pinot.plugin.stream.kafka.KafkaJSONMessageDecoder",
      "stream.kinesis.consumer.factory.class.name": "org.apache.pinot.plugin.stream.kinesis.KinesisConsumerFactory",
      "realtime.segment.flush.threshold.time": "7d",
      "realtime.segment.flush.threshold.rows": "1000000",
      "region": "us-east-1",
      "maxRecordsToFetch": "3",
      "shardIteratorType": "LATEST",
      "stream.kinesis.fetch.timeout.millis": "30000"
    },
    "onHeapDictionaryColumns": [],
    "varLengthDictionaryColumns": [],
    "enableDefaultStarTree": false,
    "enableDynamicStarTreeCreation": false,
    "aggregateMetrics": false,
    "nullHandlingEnabled": false,
    "optimizeDictionaryForMetrics": false,
    "noDictionarySizeRatioThreshold": 0,
    "rangeIndexColumns": [],
    "rangeIndexVersion": 2
  },
  "metadata": {},
  "quota": {},
  "routing": {},
  "query": {},
  "ingestionConfig": {
    "segmentTimeValueCheck": true,
    "transformConfigs": [
      {
        "columnName": "spark_write_timestamp_epoch",
        "transformFunction": "FromDateTime(spark_write_time, 'yyyy-MM-dd''T''HH:mm:ss.SSS''Z')"
      },
      {
        "columnName": "spark_write_timestamp",
        "transformFunction": "TRIM(spark_write_time)"
      },
      {
        "columnName": "row_created_time",
        "transformFunction": "now()"
      },
      {
        "columnName": "m_event",
        "transformFunction": "TRIM(event)"
      }
    ],
    "continueOnError": false,
    "rowTimeValueCheck": false
  },
  "isDimTable": false
}

Also I want to know if this is the right way to test the ingestion time.