yandrey321 commented on code in PR #11783:
URL: https://github.com/apache/ozone/pull/11783#discussion_r4385820609
##########
hadoop-hdds/client/src/main/java/org/apache/hadoop/ozone/client/io/ECBlockInputStream.java:
##########
@@ -327,15 +335,54 @@ protected long remaining() {
return blockLength() - position;
}
+ /**
+ * Reads the cells covered by the buffer from the internal block streams of
their data indexes with positioned
+ * reads, so neither this stream nor its internal streams change position,
and calls may run concurrently with each
+ * other and with sequential reads. The indexes are read in parallel if the
parallel read executor is set.
+ *
+ * <p>A failed index is not switched to a spare. Instead, a {@link
BadDataLocationException} carrying every failed
+ * index is thrown, so that the caller can fall back to a reconstruction
read for this call.
+ */
@Override
- protected int readPositioned(long offset, ByteBuffer buffer) {
- throw new UnsupportedOperationException("Use ECBlockInputStreamProxy for
positioned reads");
- }
+ protected int readPositioned(long offset, ByteBuffer buffer) throws
IOException {
+ checkOpen();
+ final long available = blockLength() - offset;
+ if (available <= 0) {
+ return EOF;
+ }
+ final int length = (int) Math.min(buffer.remaining(), available);
+ final int bufferStart = buffer.position();
- @Override
- public boolean hasCapability(String capability) {
- return !StreamCapabilities.PREADBYTEBUFFER.equalsIgnoreCase(capability)
- && super.hasCapability(capability);
+ final Map<Integer, DatanodeDetails> locations = new TreeMap<>();
+ final List<FutureTask<IndexReadResult>> tasks = new ArrayList<>();
+ for (Map.Entry<Integer, List<CellRead>> entry : splitCells(offset, length,
new ArrayList<>()).entrySet()) {
+ int index = entry.getKey();
+ BlockExtendedInputStream stream;
+ synchronized (streamLock) {
+ locations.put(index, dataLocations[index]);
+ try {
+ stream = getOrOpenStream(index);
+ } catch (IOException e) {
+ throw new BadDataLocationException(dataLocations[index], index, e);
+ }
+ }
+ List<CellRead> indexCells = entry.getValue();
+ tasks.add(new FutureTask<>(() -> readCells(stream, indexCells, buffer,
bufferStart, false, true)));
+ }
+
+ Map<Integer, IndexReadResult> results = new TreeMap<>();
+ Throwable unexpected = runIndexReads(tasks, results);
+ if (unexpected != null) {
+ Throwables.throwIfUnchecked(unexpected);
+ throw new IOException(unexpected);
+ }
+ BadDataLocationException failure = readFailure(results, locations::get);
+ if (failure != null) {
+ throw failure;
Review Comment:
Good catch!
Fixed:
- `ECBlockInputStream.readPositioned` now calls `checkOpen()` once the index
reads have finished, before it reports anything. A read that failed because the
stream was closed concurrently, for example by a sequential failover on the
same proxy, now fails with an `IOException` instead of a
`BadDataLocationException`. So its locations are never added to the shared
`failedLocations`, and the proxy falls back to a new reader for that call. The
check also covers an unchecked exception from a closed index stream, which
would otherwise be rethrown.
- `close()` sets `closed` before `closeStreams()`, so there is no window in
which the streams are closed but the flag is not set.
- `BlockInputStream.initialize()` calls `checkOpen()` after the
`initialized` check. A closed stream that was never initialized now fails with
`IOException("BlockInputStream has been closed.")` instead of an NPE from
`ContainerProtocolCalls.getBlock`. `StreamBlockInputStream` already checks
in `acquireClient()`.
Tests:
-
`TestECBlockInputStream#testPositionedReadFailingOnConcurrentCloseDoesNotReportLocations`,
serial and parallel: a pread is held in the index stream, the EC stream is
closed, and the index read then fails. The pread must fail with an
`IOException` that is not a `BadDataLocationException`. Without the
`checkOpen()` it fails with a `BadDataLocationException`.
-
`TestBlockInputStream#testPositionedReadOfClosedUninitializedStreamFailsWithIOException`:
without the check in `initialize()` it fails with `NullPointerException`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
