[
https://issues.apache.org/jira/browse/HADOOP-18546?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17643914#comment-17643914
]
ASF GitHub Bot commented on HADOOP-18546:
-----------------------------------------
steveloughran commented on code in PR #5176:
URL: https://github.com/apache/hadoop/pull/5176#discussion_r1041097514
##########
hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/services/TestAbfsInputStream.java:
##########
@@ -495,6 +499,63 @@ public void testSuccessfulReadAhead() throws Exception {
checkEvictedStatus(inputStream, 0, true);
}
+ /**
+ * This test expects InProgressList is not purged by the inputStream close.
+ */
+ @Test
+ public void testStreamPurgeDuringReadAheadCallExecuting() throws Exception {
+ AbfsClient client = getMockAbfsClient();
+ AbfsRestOperation successOp = getMockRestOp();
+ final Long serverCommunicationMockLatency = 3_000L;
+ final Long readBufferTransferToInProgressProbableTime = 1_000L;
+ final Integer readBufferQueuedCount = 3;
+
+ Mockito.doAnswer(invocationOnMock -> {
+ //sleeping thread to mock the network latency from client to backend.
+ Thread.sleep(serverCommunicationMockLatency);
+ return successOp;
+ })
+ .when(client)
+ .read(any(String.class), any(Long.class), any(byte[].class),
+ any(Integer.class), any(Integer.class), any(String.class),
+ any(String.class), any(TracingContext.class));
+
+ AbfsInputStream inputStream = getAbfsInputStream(client,
+ "testSuccessfulReadAhead.txt");
+ queueReadAheads(inputStream);
+
+ final ReadBufferManager readBufferManager
+ = ReadBufferManager.getBufferManager();
+
+ final int readBufferTotal = readBufferManager.getNumBuffers();
+
+ //Sleeping to give ReadBufferWorker to pick the readBuffers for processing.
+ Thread.sleep(readBufferTransferToInProgressProbableTime);
+
+ Assertions.assertThat(readBufferManager.getInProgressCopiedList())
+ .describedAs("InProgressList should have " + readBufferQueuedCount + "
elements")
+ .hasSize(readBufferQueuedCount);
+ final int freeListBufferCount = readBufferTotal - readBufferQueuedCount;
+ Assertions.assertThat(readBufferManager.getFreeListCopy())
+ .describedAs("FreeList should have " + freeListBufferCount +
"elements")
+ .hasSize(freeListBufferCount);
+ Assertions.assertThat(readBufferManager.getCompletedReadListCopy())
+ .describedAs("CompletedList should have 0 elements")
+ .hasSize(0);
+
+ inputStream.close();
Review Comment:
the problem with the close() here is that it will only be reached if the
assertions hold. if anything goes wrong, an exception is raised and the stream
kept open, with whatever resources it consumes.
it should be closed in a finally block *or* the stream opened in a
try-with-resources clause. thanks
> disable purging list of in progress reads in abfs stream closed
> ---------------------------------------------------------------
>
> Key: HADOOP-18546
> URL: https://issues.apache.org/jira/browse/HADOOP-18546
> Project: Hadoop Common
> Issue Type: Sub-task
> Components: fs/azure
> Affects Versions: 3.3.4
> Reporter: Steve Loughran
> Assignee: Steve Loughran
> Priority: Major
> Labels: pull-request-available
>
> turn off the prune of in progress reads in
> ReadBufferManager::purgeBuffersForStream
> this will ensure active prefetches for a closed stream complete. they wiill
> then get to the completed list and hang around until evicted by timeout, but
> at least prefetching will be safe.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: common-issues-unsubscr...@hadoop.apache.org
For additional commands, e-mail: common-issues-h...@hadoop.apache.org
