Hi @Mayank By default which strategy does upsert table follow? Attaching this payload which I am using`{` "REALTIME": { "tableName": "lmd_order_oe_REALTIME", "tableType": "REALTIME", "segmentsConfig": { "timeType": "MILLISECONDS", "segmentPushType": "APPEND", "timeColumnName": "updatedTime", "retentionTimeUnit": "DAYS", "retentionTimeValue": "30", "minimizeDataMovement": false, "schemaName": "lmd_order_oe", "replication": "2" }, "tenants": { "broker": "DefaultTenant", "server": "DefaultTenant" }, "tableIndexConfig": { "streamConfigs": { "streamType": "kafka", "stream.kafka.topic.name": "lmd_order_oe", "stream.kafka.broker.list": "", "stream.kafka.consumer.type": "lowlevel", "stream.kafka.consumer.prop.auto.offset.reset": "smallest", "stream.kafka.consumer.factory.class.name": "org.apache.pinot.plugin.stream.kafka20.KafkaConsumerFactory", "stream.kafka.decoder.class.name": "org.apache.pinot.plugin.stream.kafka.KafkaJSONMessageDecoder", "realtime.segment.flush.threshold.rows": "0", "realtime.segment.flush.threshold.time": "24h", "realtime.segment.flush.threshold.segment.size": "500M", "security.protocol": "SSL", "ssl.truststore.location": "", "ssl.keystore.location": “”, "ssl.truststore.password": "", "ssl.keystore.password": "", "ssl.key.password": "", "realtime.segment.serverUploadToDeepStore": "true", "stream.kafka.group.id": "pinot_lmd_order_oe", "group.id": "pinot_lmd_order_oe" }, "rangeIndexVersion": 2, "autoGeneratedInvertedIndex": false, "createInvertedIndexDuringSegmentGeneration": false, "loadMode": "MMAP", "enableDefaultStarTree": false, "enableDynamicStarTreeCreation": false, "aggregateMetrics": false, "nullHandlingEnabled": true, "columnMajorSegmentBuilderEnabled": true, "optimizeDictionary": false, "optimizeDictionaryForMetrics": false, "optimizeDictionaryType": false, "noDictionarySizeRatioThreshold": 0.85, "invertedIndexColumns": [], "noDictionaryColumns": [], "bloomFilterColumns": [], "onHeapDictionaryColumns": [], "rangeIndexColumns": [], "sortedColumn": [ "orderId" ], "varLengthDictionaryColumns": [] }, "metadata": {}, "quota": {}, "routing": { "segmentPrunerTypes": [ "partition" ], "instanceSelectorType": "strictReplicaGroup" }, "query": {}, "fieldConfigList": [ { "name": "scheduleStart", "encodingType": "DICTIONARY", "indexType": "TIMESTAMP", "indexTypes": [ "TIMESTAMP" ], "timestampConfig": { "granularities": [ "HOUR", "DAY", "WEEK" ] }, "indexes": null, "tierOverwrites": null }, { "name": "scheduleEnd", "encodingType": "DICTIONARY", "indexType": "TIMESTAMP", "indexTypes": [ "TIMESTAMP" ], "timestampConfig": { "granularities": [ "HOUR", "DAY", "WEEK" ] }, "indexes": null, "tierOverwrites": null }, { "name": "fmsStatus", "encodingType": "DICTIONARY", "indexTypes": [], "indexes": { "inverted": {} }, "tierOverwrites": null }, { "name": "srmDeliveryStatus", "encodingType": "DICTIONARY", "indexTypes": [], "indexes": { "inverted": {} }, "tierOverwrites": null }, { "name": "srmFulfillmentStatus", "encodingType": "DICTIONARY", "indexTypes": [], "indexes": { "inverted": {} }, "tierOverwrites": null }, { "name": "tsStatus", "encodingType": "DICTIONARY", "indexTypes": [], "indexes": { "inverted": {} }, "tierOverwrites": null }, { "name": "pickupPointId", "encodingType": "DICTIONARY", "indexTypes": [], "indexes": { "inverted": {} }, "tierOverwrites": null }, { "name": "tenantId", "encodingType": "DICTIONARY", "indexTypes": [], "indexes": { "inverted": {} }, "tierOverwrites": null } ], "upsertConfig": { "hashFunction": "NONE", "defaultPartialUpsertStrategy": "OVERWRITE", "upsertViewRefreshIntervalMs": 3000, "partialUpsertStrategies": {}, "comparisonColumns": [ "rapTimeStamp", "fmsTimeStamp", "srmIngressTimeStamp", "srmEgressTimeStamp", "tsTimeStamp" ], "enableSnapshot": true, "metadataTTL": 0, "deletedKeysTTL": 0, "enablePreload": true, "consistencyMode": "NONE", "newSegmentTrackingTimeMs": 10000, "dropOutOfOrderRecord": false, "enableDeletedKeysCompactionConsistency": false, "allowPartialUpsertConsumptionDuringCommit": false, "mode": "PARTIAL" }, "ingestionConfig": { "continueOnError": false, "rowTimeValueCheck": false, "segmentTimeValueCheck": true }, "isDimTable": false } }

I was also facing high consumer lags once. How can I debug this, this is not happening regularly but may be once or twice in a month

@Mayank Gentle reminder

@Mayank We were facing high consumer lags but only very few days, how can I debug this?

I just got one recommendation that segments should be distributed across all server and if we have small usecases.

If we don't pass replicagroup sever and replica count the segments are created across all server

For debugging consumer lag, do you have grafana setup? If so, check if • Input stream event rate is spiking • Servers are using all CPUs (number of partitions) • Was ingestion temporarily stopped