Data not getting consumed from all Kafka partitions

Issue currently being faced one of our users.

```
Topic has 3 partitions

Data getting pushed to all 3 partitions

When table was created, it was consuming data from all 3 partitions

However, after 24 hours, it has been consuming data only from partition 1. partition 0 an 2 segments show in the CONSUMING state and their metadata is also present in PROPERTY_STORE.

Resetting the stuck segments didn't work as well.
```

Here is what segments status looks like

```
{
"id": "tSCalibrationAttempt_REALTIME",
"simpleFields": {
"BATCH_MESSAGE_MODE": "false",
"IDEAL_STATE_MODE": "CUSTOMIZED",
"INSTANCE_GROUP_TAG": "tSCalibrationAttempt_REALTIME",
"MAX_PARTITIONS_PER_INSTANCE": "1",
"NUM_PARTITIONS": "13",
"REBALANCE_MODE": "CUSTOMIZED",
"REPLICAS": "1",
"STATE_MODEL_DEF_REF": "SegmentOnlineOfflineStateModel",
"STATE_MODEL_FACTORY_NAME": "DEFAULT"
},
"mapFields": {
"tSCalibrationAttempt__0__0__20220505T1656Z": {
"Server_pinot-server-0.pinot-server-headless.pinot.svc.cluster.local_8098": "ONLINE"
},
"tSCalibrationAttempt__0__1__20220506T1656Z": {
"Server_pinot-server-0.pinot-server-headless.pinot.svc.cluster.local_8098": "CONSUMING"
},
"tSCalibrationAttempt__1__0__20220505T1656Z": {
"Server_pinot-server-1.pinot-server-headless.pinot.svc.cluster.local_8098": "ONLINE"
},
"tSCalibrationAttempt__1__1__20220506T1656Z": {
"Server_pinot-server-1.pinot-server-headless.pinot.svc.cluster.local_8098": "ONLINE"
},
"tSCalibrationAttempt__1__2__20220509T0456Z": {
"Server_pinot-server-1.pinot-server-headless.pinot.svc.cluster.local_8098": "ONLINE"
},
"tSCalibrationAttempt__1__3__20220510T0456Z": {
"Server_pinot-server-1.pinot-server-headless.pinot.svc.cluster.local_8098": "ONLINE"
},
"tSCalibrationAttempt__1__4__20220511T0456Z": {
"Server_pinot-server-1.pinot-server-headless.pinot.svc.cluster.local_8098": "ONLINE"
},
"tSCalibrationAttempt__1__5__20220512T0456Z": {
"Server_pinot-server-1.pinot-server-headless.pinot.svc.cluster.local_8098": "ONLINE"
},
"tSCalibrationAttempt__1__6__20220513T0456Z": {
"Server_pinot-server-1.pinot-server-headless.pinot.svc.cluster.local_8098": "ONLINE"
},
"tSCalibrationAttempt__1__7__20220514T0456Z": {
"Server_pinot-server-1.pinot-server-headless.pinot.svc.cluster.local_8098": "ONLINE"
},
"tSCalibrationAttempt__1__8__20220516T0657Z": {
"Server_pinot-server-1.pinot-server-headless.pinot.svc.cluster.local_8098": "CONSUMING"
},
"tSCalibrationAttempt__2__0__20220505T1656Z": {
"Server_pinot-server-0.pinot-server-headless.pinot.svc.cluster.local_8098": "ONLINE"
},
"tSCalibrationAttempt__2__1__20220506T1656Z": {
"Server_pinot-server-0.pinot-server-headless.pinot.svc.cluster.local_8098": "CONSUMING"
}
},
"listFields": {}
}
```

Data is not getting consumed from partition 0 and 2. For Partition 2, here's what metadata looks like

```
{
"id": "tSCalibrationAttempt__2__0__20220505T1656Z",
"simpleFields": {
"segment.crc": "2706892624",
"segment.creation.time": "1651769803410",
"segment.download.url": "http://pinot-controller-0.pinot-controller-headless.pinot.svc.cluster.local:9000/segments/tSCalibrationAttempt/tSCalibrationAttempt__2__0__20220505T1656Z",
"segment.end.time": "1651847119229",
"segment.flush.threshold.size": "100000",
"segment.index.version": "v3",
"segment.realtime.endOffset": "819",
"segment.realtime.numReplicas": "1",
"segment.realtime.startOffset": "746",
"segment.realtime.status": "DONE",
"segment.start.time": "1651648021433",
"segment.time.unit": "MILLISECONDS",
"segment.total.docs": "73"
},
"mapFields": {},
"listFields": {}
}

{
"id": "tSCalibrationAttempt__2__1__20220506T1656Z",
"simpleFields": {
"segment.creation.time": "1651856206431",
"segment.flush.threshold.size": "10000",
"segment.realtime.numReplicas": "1",
"segment.realtime.startOffset": "819",
"segment.realtime.status": "IN_PROGRESS"
},
"mapFields": {},
"listFields": {}
}

```

### Table config:

```json
{
"REALTIME": {
"tableName": "tSCalibrationAttempt_REALTIME",
"tableType": "REALTIME",
"segmentsConfig": {
"schemaName": "tSCalibrationAttempt",
"replication": "1",
"timeColumnName": "coeff.calTimestamp",
"allowNullTimeValue": false,
"replicasPerPartition": "1",
"completionConfig": {
"completionMode": "DOWNLOAD"
}
},
"tenants": {
"broker": "DefaultTenant",
"server": "DefaultTenant",
"tagOverrideConfig": {}
},
"tableIndexConfig": {
"invertedIndexColumns": [],
"streamConfigs": {
"streamType": "kafka",
"stream.kafka.topic.name": "tscalibrationattempt",
"stream.kafka.broker.list": "",
"stream.kafka.consumer.type": "lowlevel",
"stream.kafka.consumer.prop.auto.offset.reset": "largest",
"stream.kafka.schema.registry.url": "",
"stream.kafka.consumer.factory.class.name": "org.apache.pinot.plugin.stream.kafka20.KafkaConsumerFactory",
"stream.kafka.decoder.prop.schema.registry.rest.url": "",
"sasl.mechanism": "PLAIN",
"stream.kafka.decoder.prop.basic.auth.credentials.source": "USER_INFO",
"stream.kafka.decoder.prop.schema.registry.basic.auth.user.info": "",
"security.protocol": "SASL_SSL",
"sasl.jaas.config": "org.apache.kafka.common.security.scram.ScramLoginModule required username=\"\" password=\"\";",
"stream.kafka.decoder.class.name": "org.apache.pinot.plugin.inputformat.avro.confluent.KafkaConfluentSchemaRegistryAvroMessageDecoder",
"realtime.segment.flush.threshold.rows": "0",
"realtime.segment.flush.threshold.time": "24h",
"realtime.segment.flush.threshold.segment.size": "150M"
},
"rangeIndexColumns": [
"coeff.calTimestamp"
],
"rangeIndexVersion": 2,
"autoGeneratedInvertedIndex": false,
"createInvertedIndexDuringSegmentGeneration": false,
"sortedColumn": [
"deviceId"
],
"bloomFilterColumns": [],
"loadMode": "MMAP",
"onHeapDictionaryColumns": [],
"varLengthDictionaryColumns": [],
"enableDefaultStarTree": false,
"enableDynamicStarTreeCreation": false,
"aggregateMetrics": false,
"nullHandlingEnabled": false
},
"metadata": {},
"quota": {},
"routing": {
"instanceSelectorType": "strictReplicaGroup"
},
"query": {},
"ingestionConfig": {
"complexTypeConfig": {
"fieldsToUnnest": [
"coeff"
]
}
},
"isDimTable": false
}
}
```

### Update:

Once the problematic segments (`tSCalibrationAttempt__2__1__20220506T1656Z` and `tSCalibrationAttempt__0__1__20220506T1656Z`) are deleted via API call (`DELETE /segments/{tableName}/{segmentName}`) , the consumption resumes succesfully for partition 2 but no segments still created for partition 0.

Start with the table's low-level Kafka stream configuration and compare the segment states and realtime offsets for partitions 0, 1, and 2 in PROPERTY_STORE. Use the DELETE /segments/{tableName}/{segmentName} behavior as a comparison point; done means identifying why consumption stops for individual partitions and documenting a reproducible resolution.

Written by the indexing model from the issue text.