HBASE-28417 TestBlockingIPC.testBadPreambleHeader sometimes fails with… #5740
+112
−95
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -56,6 +56,7 @@ | |
| import io.opentelemetry.sdk.trace.data.SpanData; | ||
| import java.io.IOException; | ||
| import java.net.InetSocketAddress; | ||
| import java.nio.channels.SocketChannel; | ||
| import java.time.Duration; | ||
| import java.util.ArrayList; | ||
| import java.util.Collections; | ||
|
|
@@ -73,15 +74,18 @@ | |
| import org.apache.hadoop.hbase.ServerName; | ||
| import org.apache.hadoop.hbase.Waiter; | ||
| import org.apache.hadoop.hbase.ipc.RpcServer.BlockingServiceAndInterface; | ||
| import org.apache.hadoop.hbase.nio.ByteBuff; | ||
| import org.apache.hadoop.hbase.security.User; | ||
| import org.apache.hadoop.hbase.util.Bytes; | ||
| import org.apache.hadoop.hbase.util.EnvironmentEdgeManager; | ||
| import org.apache.hadoop.io.compress.GzipCodec; | ||
| import org.apache.hadoop.ipc.RemoteException; | ||
| import org.apache.hadoop.util.StringUtils; | ||
| import org.hamcrest.Matcher; | ||
| import org.junit.Before; | ||
| import org.junit.Rule; | ||
| import org.junit.Test; | ||
| import org.junit.runners.Parameterized.Parameter; | ||
| import org.slf4j.Logger; | ||
| import org.slf4j.LoggerFactory; | ||
|
|
||
|
|
@@ -114,14 +118,12 @@ public abstract class AbstractTestIPC { | |
| private static final KeyValue CELL = new KeyValue(CELL_BYTES, CELL_BYTES, CELL_BYTES, CELL_BYTES); | ||
|
|
||
| protected static final Configuration CONF = HBaseConfiguration.create(); | ||
|
|
||
| protected RpcServer createRpcServer(Server server, String name, | ||
| List<BlockingServiceAndInterface> services, InetSocketAddress bindAddress, Configuration conf, | ||
| RpcScheduler scheduler) throws IOException { | ||
| return RpcServerFactory.createRpcServer(server, name, services, bindAddress, conf, scheduler); | ||
| } | ||
|
|
||
| private RpcServer createRpcServer(String name, List<BlockingServiceAndInterface> services, | ||
| InetSocketAddress bindAddress, Configuration conf, RpcScheduler scheduler) throws IOException { | ||
|
|
@@ -133,6 +135,14 @@ private RpcServer createRpcServer(String name, List<BlockingServiceAndInterface> | |
| @Rule | ||
| public OpenTelemetryRule traceRule = OpenTelemetryRule.create(); | ||
|
|
||
| @Parameter(0) | ||
| public Class<? extends RpcServer> rpcServerImpl; | ||
|
|
||
| @Before | ||
| public void setUpBeforeTest() { | ||
| CONF.setClass(RpcServerFactory.CUSTOM_RPC_SERVER_IMPL_CONF_KEY, rpcServerImpl, RpcServer.class); | ||
| } | ||
|
|
||
| /** | ||
| * Ensure we do not HAVE TO HAVE a codec. | ||
| */ | ||
|
|
@@ -348,9 +358,43 @@ public void testTimeout() throws IOException { | |
| } | ||
| } | ||
|
|
||
| private static class FailingSimpleRpcServer extends SimpleRpcServer { | ||
|
|
||
| FailingSimpleRpcServer(Server server, String name, | ||
| List<RpcServer.BlockingServiceAndInterface> services, InetSocketAddress bindAddress, | ||
| Configuration conf, RpcScheduler scheduler) throws IOException { | ||
| super(server, name, services, bindAddress, conf, scheduler, true); | ||
| } | ||
|
|
||
| final class FailingConnection extends SimpleServerRpcConnection { | ||
| private FailingConnection(FailingSimpleRpcServer rpcServer, SocketChannel channel, | ||
| long lastContact) { | ||
| super(rpcServer, channel, lastContact); | ||
| } | ||
|
|
||
| @Override | ||
| public void processRequest(ByteBuff buf) throws IOException, InterruptedException { | ||
| // this will throw exception after the connection header is read, and an RPC is sent | ||
| // from client | ||
| throw new DoNotRetryIOException("Failing for test"); | ||
| } | ||
| } | ||
|
|
||
| @Override | ||
| protected SimpleServerRpcConnection getConnection(SocketChannel channel, long time) { | ||
| return new FailingConnection(this, channel, time); | ||
| } | ||
| } | ||
|
|
||
| protected RpcServer createTestFailingRpcServer(final String name, | ||
| final List<BlockingServiceAndInterface> services, final InetSocketAddress bindAddress, | ||
| Configuration conf, RpcScheduler scheduler) throws IOException { | ||
| if (rpcServerImpl.equals(NettyRpcServer.class)) { | ||
| return new FailingNettyRpcServer(null, name, services, bindAddress, conf, scheduler); | ||
| } else { | ||
| return new FailingSimpleRpcServer(null, name, services, bindAddress, conf, scheduler); | ||
| } | ||
| } | ||
|
|
||
| /** Tests that the connection closing is handled by the client with outstanding RPC calls */ | ||
| @Test | ||
|
|
@@ -570,19 +614,33 @@ public void testTracingErrorIpc() throws IOException { | |
|
|
||
| protected abstract AbstractRpcClient<?> createBadAuthRpcClient(Configuration conf); | ||
|
|
||
| private IOException doBadPreableHeaderCall(BlockingInterface stub) { | ||
| ServiceException se = assertThrows(ServiceException.class, | ||
| () -> stub.echo(null, EchoRequestProto.newBuilder().setMessage("hello").build())); | ||
| return ProtobufUtil.handleRemoteException(se); | ||
| } | ||
|
|
||
| @Test | ||
| public void testBadPreambleHeader() throws Exception { | ||
| Configuration clientConf = new Configuration(CONF); | ||
| RpcServer rpcServer = createRpcServer("testRpcServer", Collections.emptyList(), | ||
| new InetSocketAddress("localhost", 0), CONF, new FifoRpcScheduler(CONF, 1)); | ||
| try (AbstractRpcClient<?> client = createBadAuthRpcClient(clientConf)) { | ||
| rpcServer.start(); | ||
| BlockingInterface stub = newBlockingStub(client, rpcServer.getListenerAddress()); | ||
| BadAuthException error = null; | ||
| // for SimpleRpcServer, it is possible that we get a broken pipe before getting the | ||
| // BadAuthException, so we add some retries here, see HBASE-28417 | ||
| for (int i = 0; i < 10; i++) { | ||
|
There was a problem hiding this comment. Haha, same request as I received from @NihalJain -- can this use Waiter instead? There was a problem hiding this comment. Here the things are a bit different. It is not something that after some time, things will go right. We just need to try several times, and we should get at least one time, things go as expected... |
||
| IOException ioe = doBadPreableHeaderCall(stub); | ||
| if (ioe instanceof BadAuthException) { | ||
| error = (BadAuthException) ioe; | ||
| break; | ||
| } | ||
| Thread.sleep(100); | ||
| } | ||
| assertNotNull("Can not get expected BadAuthException", error); | ||
| assertThat(error.getMessage(), containsString("authName=unknown")); | ||
| } finally { | ||
| rpcServer.stop(); | ||
| } | ||
|
|
||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Can you get rid of this static CONF instance entirely. All of its accesses in this class are using it with a copy constructor, but TestNettyTlsIPC is using/modifying it directly.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
We can not get rid of it, we need to change some configurations in setUp and tearDown. And in TestNettyTlsIPC, we need to change the configurations in setUpBeforeClass, so it has to be static...
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
The copying in the tests method is because we need to customize some client configurations, so we want to affect other test methods.