big-data/hadoop
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Revert "HADOOP-18024. SocketChannel is not closed when IOException happens in Server$Listener.doAccept (#3719)"

Browse Source 
This reverts commit 6ed01585eb.

Breaks TestIPC#testIOEOnListenerAccept
This commit is contained in:
Ayush Saxena 2022-02-01 11:00:54 +05:30
parent e8f767f2f4
commit aeae5716cc
No known key found for this signature in database
GPG Key ID: D09AE71061AB564D
2 changed files with 4 additions and 68 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

25
hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/Server.java
View File

@ -1329,14 +1329,6 @@ public String toString() {
} }
} }
@VisibleForTesting
@InterfaceAudience.Private
protected void configureSocketChannel(SocketChannel channel) throws IOException {
channel.configureBlocking(false);
channel.socket().setTcpNoDelay(tcpNoDelay);
channel.socket().setKeepAlive(true);
}
/** Listens on the socket. Creates jobs for the handler threads*/ /** Listens on the socket. Creates jobs for the handler threads*/
private class Listener extends Thread { private class Listener extends Thread {
@ -1543,24 +1535,15 @@ private void closeCurrentConnection(SelectionKey key, Throwable e) {
InetSocketAddress getAddress() { InetSocketAddress getAddress() {
return (InetSocketAddress)acceptChannel.socket().getLocalSocketAddress(); return (InetSocketAddress)acceptChannel.socket().getLocalSocketAddress();
} }
void doAccept(SelectionKey key) throws InterruptedException, IOException, OutOfMemoryError { void doAccept(SelectionKey key) throws InterruptedException, IOException, OutOfMemoryError {
ServerSocketChannel server = (ServerSocketChannel) key.channel(); ServerSocketChannel server = (ServerSocketChannel) key.channel();
SocketChannel channel; SocketChannel channel;
while ((channel = server.accept()) != null) { while ((channel = server.accept()) != null) {
try { channel.configureBlocking(false);
configureSocketChannel(channel); channel.socket().setTcpNoDelay(tcpNoDelay);
} catch (IOException e) { channel.socket().setKeepAlive(true);
LOG.warn("Error in an accepted SocketChannel", e);
try {
channel.socket().close();
channel.close();
} catch (IOException ex) {
LOG.warn("Error in closing SocketChannel", ex);
}
continue;
}
Reader reader = getReader(); Reader reader = getReader();
Connection c = connectionManager.register(channel, Connection c = connectionManager.register(channel,

47
hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestIPC.java
View File

@ -34,7 +34,6 @@
import java.io.ByteArrayOutputStream; import java.io.ByteArrayOutputStream;
import java.io.DataInput; import java.io.DataInput;
import java.io.DataOutput; import java.io.DataOutput;
import java.io.EOFException;
import java.io.File; import java.io.File;
import java.io.IOException; import java.io.IOException;
import java.io.InputStream; import java.io.InputStream;
@ -49,7 +48,6 @@
import java.net.SocketException; import java.net.SocketException;
import java.net.SocketTimeoutException; import java.net.SocketTimeoutException;
import java.net.UnknownHostException; import java.net.UnknownHostException;
import java.nio.channels.SocketChannel;
import java.util.ArrayList; import java.util.ArrayList;
import java.util.Collections; import java.util.Collections;
import java.util.HashSet; import java.util.HashSet;
@ -618,51 +616,6 @@ public void testIOEOnWriteAfterPingClient() throws Exception {
WRITABLE_FAULTS_SLEEP = 0; WRITABLE_FAULTS_SLEEP = 0;
} }
} }
/**
* Test for HADOOP-18024.
*/
@Test(timeout=60000)
public void testIOEOnListenerAccept() throws Exception {
// start server
Server server = new TestServer(1, false,
LongWritable.class, LongWritable.class) {
@Override
protected void configureSocketChannel(SocketChannel channel) throws IOException {
maybeThrowIOE();
super.configureSocketChannel(channel);
}
};
InetSocketAddress addr = NetUtils.getConnectAddress(server);
server.start();
// start client
WRITABLE_FAULTS_ENABLED = true;
Client client = new Client(LongWritable.class, conf);
try {
LongWritable param = LongWritable.class.newInstance();
try {
call(client, param, addr, 0, conf);
fail("Expected an exception to have been thrown");
} catch (EOFException e) {
LOG.info("Got expected exception", e);
} catch (Throwable t) {
LOG.warn("Got unexpected error", t);
fail("Expected an EOFException to have been thrown");
}
// Doing a second call with faults disabled should return fine --
// ie the internal state of the client or server should not be broken
// by the failed call
WRITABLE_FAULTS_ENABLED = false;
call(client, param, addr, 0, conf);
} finally {
client.stop();
server.stop();
}
}
private static void assertExceptionContains( private static void assertExceptionContains(
Throwable t, String substring) { Throwable t, String substring) {