HDDS-16867. ECBlockInputStream should read data cells from datanodes in parallel

[yandrey321]

What changes were proposed in this pull request?

On the healthy EC read path, ECBlockInputStream reads one cell at a time, so a read that spans N cells costs
N serial datanode round trips, even though the cells are on different datanodes. With this change, a read that
spans more than one cell reads the cells of different data indexes in parallel. The reconstruction path is already
parallel (HDDS-5952), and HDFS does the same in StripeReader.readStripe.

Key decisions:

• Off by default. The new flag ozone.client.ec.parallel.read.enabled defaults to false, and with it off,
  behavior is unchanged. Reads within a single cell always stay on the existing serial path.
• Dedicated pool. Parallel reads use a new pool (ozone.client.ec.parallel.read.pool.limit, default 30), not
  the reconstruct pool, so healthy and degraded reads do not compete for threads. The pool is created lazily.
  When it is saturated, the calling thread reads the remaining cells itself.
• One task per data index. That index's cells from consecutive stripes are contiguous in its stream, so the
  fan-out is at most the number of data units.
• Cannot hang. The caller runs the first task itself and then any task the pool has not started. It waits for
  every task before returning, so nothing writes into the caller's buffer after read returns.
• Failure semantics. A failure throws, as the serial path does, instead of returning the good prefix:
  ECBlockInputStreamProxy treats each inner read as all-or-nothing. Every failed index of the call is reported
  in one BadDataLocationException, so spare-location retry and the failover to reconstruction each happen once.
• Compatible API. Existing constructors and factory methods are unchanged and read serially; overloads take
  the pool.
• Request context follows the read onto pool threads. A pool thread can call OM: BlockInputStream refreshes
  block locations after a StorageContainerException or a connectivity error. The S3 authentication and the
  read-consistency hint that OM requests carry are thread-locals of the OM client, set on the request thread by the
  S3 Gateway. The pool's execute therefore wraps each task: it copies the submitter's values onto the worker, runs
  the task, and then restores the worker's previous values, or clears them if it had none. Without this, a refresh
  from a worker fails with S3 Auth expected when the S3 auth check is on, or silently uses the default read
  consistency.

Out of scope: reading ahead to the end of the stripe (HDFS curStripeBuf). Byte-array reads on encrypted (TDE)
keys do not benefit, because CryptoInputStream reads through an 8 KB buffer, which never spans more than one
cell.

Generated-by: Claude Code (Claude Opus 5.5)

What is the link to the Apache JIRA

https://issues.apache.org/jira/browse/HDDS-16867

How was this patch tested?

• Both modes in every run: the existing EC read tests now run with cells read one at a time and in parallel,
  with no extra configuration. TestECBlockInputStream and TestOzoneECClient use @ParameterizedClass, the
  two EC tests in TestOzoneFSInputStream are parameterized, and TestKeyInputStream reads through both kinds of
  client. All pass.
• New unit tests: they check byte-exact data for rs-3-2, rs-6-3 and rs-10-4. A latch-based test proves reads
  are in flight on all indexes at once, and it fails with Timed out waiting for 2 concurrent reads when the
  executor hand-off is removed. Other tests cover multi-index failure reporting and spare-location retry.
• Request context on pool threads:
  • TestOzoneECClient.testLocationRefreshDuringReadKeepsRequestContext sets S3 auth and a LOCAL_LEASE hint on the
    reading thread. It then makes one GetBlock fail with CONTAINER_NOT_FOUND, which forces an OM refresh, from a
    pool thread in parallel mode. Every recorded OM request must carry both. Without the fix it fails in parallel
    mode only (GetKeyInfo sent without S3 authentication, and expected LOCAL_LEASE but was READ_CONSISTENCY_UNSPECIFIED).
  • TestRpcClient.testWithSubmitterContextPropagatesAndRestores covers the wrapper: what is copied, restoring the
    worker's own values, clearing when it had none, and restoring when the task throws.
  • End to end: TestSecureOzoneRpcClient.testEcReadWithS3AuthAndReadConsistency reads a multi-cell EC key with
    S3 auth and a read-consistency hint, in both modes. AbstractS3SDKV2Tests gets an S3 GET of a multi-cell
    object in an EC bucket with each read-consistency header. The new TestS3SDKWithEcParallelRead runs the whole
    S3 SDK suite with gateways that read in parallel. Both parallel cases assert that the pool was used.
    TestS3SDK and TestS3SDKWithEcParallelRead: 470 run, 0 failures. These tests cannot force a worker-side
    refresh (datanode restart does not work in the secure suite), so the unit test above is the regression test.
• Regression: TestOzoneIntegrationNonHA (340 run, 0 failures), TestECKeyOutputStream, TestKeyInputStream.
• Test housekeeping: the integration-test auditlog.properties and AuditLogTestUtils now write and read
  target/audit.log and target/system_audit.log. Before, they used the module root, and the new secure test
  (through OzoneRpcClientTests, which never deletes them) left both files there. TestOmAcls (9/9) reads the new
  location.

Benchmark (TestEcParallelReadBenchmark, tagged benchmark so it does not run in CI): mini cluster, 48 MB
keys, 1 MB cells, 7 rounds after warm-up, run twice. Speed-up of parallel over serial throughput:

Cluster	Policy	1 reader, buffer of 1 stripe or more	4 clients x 1 thread each	1 MB buffer (one cell, control)
5 DN	rs-3-2	1.75-1.87x	1.25-1.32x	0.99-1.02x
9 DN	rs-3-2	1.71-1.83x	1.16-1.23x	0.96-1.06x
9 DN	rs-6-3	2.17-2.30x	1.19-1.22x	0.99-1.03x

Observations:

• The gain grows with the number of data units (rs-6-3 over rs-3-2). The one-cell control stays within noise
  (about +-8%).
• With 4 clients, median latency drops 17-24% both per file and per read() call. p99 mostly drops too, but 7
  rounds are too few to support tail claims.
• These numbers are a lower bound. The datanodes are in-JVM and on loopback, so the network round trips the change
  removes are nearly free here.
• The 4-client gain is capped by one machine's CPUs. That is also why serial throughput is lower on 9 datanodes
  than on 5, so the 5 vs 9 difference reflects CPU load, not how cells are spread over datanodes.

[yandrey321]

@jojochuang could you please take a look?

[yandrey321]

@ss77892 could you please take a look?

[smengcl]

Thanks @yandrey321 for the update

[smengcl]

EC_PARALLEL_READ_POOL_MIN_SIZE is 3, but nothing keeps ozone.client.ec.parallel.read.pool.limit at or above it. With a limit of 1 or 2, new ThreadPoolExecutor throws IllegalArgumentException (maximumPoolSize < corePoolSize) on the first get(). That is unchecked, so ECBlockInputStreamProxy does not fail over to reconstruction, and every multi-cell read fails the same way.

Could the core size be Math.min(EC_PARALLEL_READ_POOL_MIN_SIZE, limit), or the limit validated? The reconstruct pool has the same gap, but this new key is more likely to be tuned down.

[smengcl]

The request context problem fixed here for the parallel read pool also applies to ecReconstructExecutor. ECBlockReconstructedStripeInputStream runs readIntoBuffer on these threads, through a BlockInputStream that can refresh block locations from OM after a StorageContainerException. On an S3 Gateway with the S3 auth check on, that refresh has no S3 auth on the worker thread and fails with S3 Auth expected, and it also loses the read consistency hint.

Should this pool also be created with createRequestContextPropagatingExecutor? A follow-up JIRA is fine if you prefer to keep this PR scoped.

[smengcl]

And some minor comments

[smengcl]

Minor: when this method ends by throwing, position only moves past the cells before the first failure (and not at all when line 504 rethrows an unchecked exception). Indexes whose tasks succeeded have already moved their inner streams further, though. If seeked is still false, a later read on this stream skips the position check in readCells and seekStreamIfNecessary and reads from the wrong offset.

The proxy closes this reader on BadDataLocationException, so the main path is safe, but a caller that retries after a RuntimeException would silently get wrong data. Setting seeked = true before throwing would cover it.

[smengcl]

This starts another cluster and reruns the whole S3 SDK suite to flip one client flag, but only the multi-cell EC GET exercises the new path. Could the parallel case run on the existing cluster instead, e.g. as a parameterized variant of the EC GET test added to AbstractS3SDKV2Tests? That would avoid roughly doubling the S3 SDK test time.

[smengcl]

Since this is tagged benchmark and never runs in CI, it is likely to rot. Would it be better to leave it out of the tree (or move it to a separate JIRA) and keep the results in the PR description?

Similarly, moving the audit log files under target/ looks useful but is independent of this feature, so it could be its own small change.