Implement re-usable activeBatchIterator and activeBatchesIterator

[jeffxiang]

We were hitting AWS SDK v2 Netty HealthCheckedChannelPool acquire timeouts in S3PartitionConsumer.poll() (“Acquire operation took longer than 10000ms”) because our S3 NIO S3SeekableByteChannel usage drives a very high rate of concurrent ranged GETs under load. Due to creating a new iterator every poll() call, our iterator pattern amplifies heap utilization / allocations and connection acquisition rates. An example of this problem looks like the following:

Assume:
startPos = 0
max-read-fragments = 50
fragmentSize = 5mb
s3Records --> some 1gb log segment on S3

poll() --> s3Records.batchesFrom(0).iterator() --> fragments: [
  [0, 5),
  [5, 10),
  [10, 15),
  ...
  [245, 250)
]

This poll call loads 50 * 5mb = 250mb across 50 chunks into on-heap memory, making 50 GET calls in the process (1 for each chunk). With startPos = 0, this poll call buffered [0, 250] byte range in the s3 log segment onto heap memory.

Now assume we only read 10 chunks (50 mb) before returning the poll result. Next poll has startPos = 50mb.

poll() --> s3Records.batchesFrom(50).iterator() --> fragments: [
  [50, 55),
  [55, 60),
  [60, 65),
  ...
  [295, 300)
]

This next poll call loads [50, 300] byte range in the S3 log segment. Notice that we are redoing I/O for bytes [50,250] in the target s3Records object, amplifying both heap and connection utilization. Note that in reality, it is unlikely that the entire [50,300] byte range is loaded all at once - instead, there is some read-ahead logic that might schedule a smaller byte range to be fetched in to memory. Regardless, the wasted I/O and memory + connection amplification problem exists.

The improved design in this PR allows us to re-use the same iterator from the first poll call in the second poll call until we exhaust the first iterator's byte range (activeBatchesIterator). We also keep the activeBatchIterator (iterator for each chunk) across polls for same-batch granularity. When we exhaust the current chunk's activeBatchIterator, we advance to the next chunk by calling activeBatchesIterator.next(). Once activeBatchesIterator is fully exhausted, only then we will call s3Records.batchesFrom(startPos).iterator() to get a new activeBatchesIterator for the next byte range.

[vahidhashemian]

Any harm in resetting activeBatchIterator as well?

[jeffxiang]

good point, done.

[vahidhashemian]

Is this a repeat of the call in line 211? If so, can we avoid the duplicate call?

[jeffxiang]

for safety we should get the bytePosition again when we are reloading the batchIterators after the current ones are exhausted, because the following line s3Records.batchesFrom(bytePosition).iterator() should give us the next range of bytes given the current offset position we are trying to read. this is especially the case if we exhaust the iterators in the middle of a log segment file. For example, with 1G log segment file and 512 MB read-ahead loaded into the iterator, when we exhaust the first 512 MB in the iterators and get to this logic, we should instantiate the new iterators from the middle of the log segment. without this line it might rewind to the beginning. this call is relatively infrequent now compared to before.

[vahidhashemian]

Any risk of flooding the log?

[jeffxiang]

this is relatively infrequent (maybe once every 1 min) since we only log this when the entire iterator is exhausted.