diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/XceiverServerDomainSocket.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/XceiverServerDomainSocket.java index 10844417db2..c1826d38d2c 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/XceiverServerDomainSocket.java +++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/XceiverServerDomainSocket.java @@ -232,9 +232,19 @@ public void run() { // another thread closed our listener socket - that's expected during shutdown, but not in other circumstances LOG.info("XceiverServerDomainSocket is closed", ace); } catch (IOException ie) { - // usually when the xceiver count limit is hit. - LOG.warn("Got an exception. Peer {}", peer, ie); - IOUtils.closeQuietly(peer); + if (peer == null) { + // accept() itself failed, for example with no file descriptor left, and would fail again right away + LOG.warn("accept() failed. Will retry after 1 second.", ie); + try { + Thread.sleep(TimeUnit.SECONDS.toMillis(1L)); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } else { + // usually when the xceiver count limit is hit. + LOG.warn("Got an exception. Peer {}", peer, ie); + IOUtils.closeQuietly(peer); + } } catch (OutOfMemoryError ie) { IOUtils.closeQuietly(peer); // DataNode can run out of memory if there is too many transfers. diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/TestXceiverServerDomainSocket.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/TestXceiverServerDomainSocket.java index d5e9b3c0eb1..62f6b5d1251 100644 --- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/TestXceiverServerDomainSocket.java +++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/TestXceiverServerDomainSocket.java @@ -28,7 +28,10 @@ import static org.junit.jupiter.api.Assertions.assertSame; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.junit.jupiter.api.Assertions.fail; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.RETURNS_DEEP_STUBS; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.mockStatic; import static org.mockito.Mockito.when; import com.google.common.collect.Maps; @@ -52,6 +55,7 @@ import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; import org.apache.commons.io.FileUtils; import org.apache.commons.io.IOUtils; import org.apache.commons.lang3.RandomStringUtils; @@ -95,6 +99,7 @@ import org.junit.jupiter.api.io.TempDir; import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.CsvSource; +import org.mockito.MockedStatic; import org.slf4j.LoggerFactory; /** @@ -636,6 +641,36 @@ public void testMaxXceiverCount() throws Exception { } } + /** + * A failed accept(), for example with no file descriptor left, is retried after a pause instead of at once. + */ + @Test + public void testAcceptFailureIsRetriedAfterPause() throws Exception { + conf.set(ScmConfigKeys.HDDS_DATANODE_DIR_KEY, dir.getAbsolutePath()); + AtomicInteger accepts = new AtomicInteger(); + DomainSocket listener = mock(DomainSocket.class, RETURNS_DEEP_STUBS); + when(listener.accept()).thenAnswer(invocation -> { + accepts.incrementAndGet(); + throw new SocketException("accept(2) error: Too many open files"); + }); + DomainSocketFactory factory = mock(DomainSocketFactory.class); + when(factory.isServiceEnabled()).thenReturn(true); + when(factory.isServiceReady()).thenReturn(true); + XceiverServerDomainSocket server; + try (MockedStatic domainSocket = mockStatic(DomainSocket.class)) { + domainSocket.when(() -> DomainSocket.bindAndListen(any())).thenReturn(listener); + server = new XceiverServerDomainSocket(randomDatanodeDetails(), conf, null, readExecutors, metrics, factory); + } + try { + server.start(); + GenericTestUtils.waitFor(() -> accepts.get() > 0, 10, 5000); + Thread.sleep(500); + assertTrue(accepts.get() <= 2, "accept() was called " + accepts.get() + " times"); + } finally { + server.stop(); + } + } + /** * When server receives any message which doesn't follow the version, request type, request body sequence, server * will treat it as a critical error, close the connection.