docs: protected void accountForFetchedKey(byte[] key) {
          fetched++;
          if (streamStats != null) {
              streamStats.reportStreamingFetch(operation);
          }
  
          if (recordsPerPartition <= 0) {
              return;
          }
  
          Integer keyPartitionId = getKeyPartitionId(key);
          Long partitionFetch = partitionFetches.get(keyPartitionId);
          Utils.notNull(partitionFetch);
          partitionFetch++;
  
          partitionFetches.put(keyPartitionId, partitionFetch);
          if (partitionFetch == recordsPerPartition) {
              if (partitionsToFetch.contains(keyPartitionId)) {
                  partitionsToFetch.remove(keyPartitionId);
              } else {
                  logger.warn("Partitions to fetch did not contain expected partition ID: "
                          + keyPartitionId);
              }
          } else if (partitionFetch > recordsPerPartition) {
              logger.warn("Partition fetch count larger than expected for partition ID "
                      + keyPartitionId + " : " + partitionFetch);
          }
      }