An empty partition can block data ingestion from Kafka/Kinesis

### Affected Version

All version since 0.10

### Description

When the supervisor is restarted or reset, it first finds the valid offsets to continue reading from metadata storage. If there's no offsets in metadata storage, then it fetches the latest/earliest offset from Kafka/Kinesis per partition. It's implemented here.

```java
private ImmutableMap> generateStartingSequencesForPartitionGroup(
int groupId
)
{
ImmutableMap.Builder> builder = ImmutableMap.builder();
for (Entry entry : partitionGroups.get(groupId).entrySet()) {
PartitionIdType partition = entry.getKey();
SequenceOffsetType sequence = entry.getValue();

if (!getNotSetMarker().equals(sequence)) {
// if we are given a startingOffset (set by a previous task group which is pending completion) then use it
if (!isEndOfShard(sequence)) {
builder.put(partition, makeSequenceNumber(sequence, useExclusiveStartSequenceNumberForNonFirstSequence()));
}
} else {
// if we don't have a startingOffset (first run or we had some previous failures and reset the sequences) then
// get the sequence from metadata storage (if available) or Kafka/Kinesis (otherwise)
OrderedSequenceNumber offsetFromStorage = getOffsetFromStorageForPartition(partition);

if (offsetFromStorage != null) {
builder.put(partition, offsetFromStorage);
}
}
}
return builder.build();
}
```

If the supervisor sees an exception while fetching the offset from Kafka/Kinesis for some reason (e.g., timeout because of an empty partition), then this method would return immediately and handling the runNotice would fail because of the exception. This may block the entire data ingestion until any data is ingested to all partitions.

I think the supervisor should be able to skip some partitions if it sees some error while getting the offset and continue ingestion for other partitions.

Start with generateStartingSequencesForPartitionGroup, where offsets are read from metadata storage or Kafka/Kinesis and an exception can stop processing the partition group. Trace the runNotice handling around this method and identify the existing tests for supervisor restart or offset fetching. Done means an offset-fetching error can skip the affected partition while ingestion continues for other partitions.

Written by the indexing model from the issue text.