[New Physical IO] Added memory manager changes to new physical IO

[ozkoca]

Description of change

This PR adopts the memory manager changes to new physicalIO/

Relevant issues

PR History:

• [INITIAL VERSION] New PhysicalIO #286
• [New Physical IO] Adopting OpenStremInformation changes to new PhysicalIO #287

Does this contribution introduce any breaking changes to the existing APIs or behaviors?

No

Does this contribution introduce any new public APIs or behaviors?

No

---

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license and I agree to the terms of the Developer Certificate of Origin (DCO).

[rajdchak]

putting into the index cache and adding the metrics should be done before dataReadyLatch.countDown(), as per the current logic we do everything inside the data future

            this.source.thenApply(
                objectContent -> {
                  try {
                    byte[] bytes =
                        StreamUtils.toByteArray(
                            objectContent,
                            this.blockKey.getObjectKey(),
                            this.blockKey.getRange(),
                            this.readTimeout);
                    int blockRange = blockKey.getRange().getLength();
                    this.aggregatingMetrics.add(MetricKey.MEMORY_USAGE, blockRange);
                    this.indexCache.put(blockKey, blockRange);
                    return bytes;
                  } catch (IOException | TimeoutException e) {
                    throw new RuntimeException(
                        "Error while converting InputStream to byte array", e);
                  }

and the read starts after we join on the data.

This is likely to cause issue in the cleanup logic where we are doing

if (block.isDataReady() && !indexCache.contains(block.getBlockKey())) {
        try {
          iterator.remove();
          BlockKey blockKey = block.getBlockKey();
          aggregatingMetrics.reduce(MetricKey.MEMORY_USAGE, blockKey.getRange().getLength());
          LOG.debug(
              "Removed block with key {}-{}-{} from block store during cleanup",
              blockKey.getObjectKey().getS3URI(),
              blockKey.getRange().getStart(),
              blockKey.getRange().getEnd());
        } catch (Exception e) {
          LOG.error("Error in removing block {}", e.getMessage());
        }
      }

[ozkoca]

good catch. Addressed in next rev

[rajdchak]

If we are adding the metrics while finding the missing blocks, do we need to publish the metrics from here? I don't think this is required.

[ozkoca]

This is required for positional read, but in next revision, I will remove direct block access for positional read and use makeRangeAvailable instead to be consistent.