[#1143] Do not report a Notice of Disconnection that cannot reach an alread...

Valery Kharseko

10 hours ago cdecf531161639f2f9fe98b5a8cc3facd43bb116

[#1143] Do not report a Notice of Disconnection that cannot reach an already-closed client as an unhandled error (#1146)

## Problem

The server can drop a connection with a Notice of Disconnection after
the client has already closed its end. For example,
`AuthenticatedUsers.doPostResponse` does this after a DELETE of the
bound user. The notice then cannot be written, and the failure is
reported as an uncaught RxJava error. The server prints an
`OnErrorNotImplementedException ... | java.io.EOFException` stack trace.
According to the code, on a worker thread it also logs
`ERR_UNCAUGHT_THREAD_EXCEPTION` and raises an
`ALERT_TYPE_UNCAUGHT_EXCEPTION` alert. This produces 7 to 11 traces per
run in the `build-docker-alpine` benchmark log. See #1143.

## Cause

`LDAPServerFilter.ClientConnectionImpl.disconnect(ResultCode, String)`
subscribed to the notification with a bare `.subscribe()`. In RxJava
3.1.10, `EmptyCompletableObserver.onError` does not throw. It hands the
error to `RxJavaPlugins.onError`, which prints it and passes it to the
thread's uncaught-exception handler. So the `try/catch
(OnErrorNotImplementedException)` around `s.onError()` in
`sendUnsolicitedNotification()` (added in #555) never saw anything.

## Change

- `disconnect(ResultCode, String)` subscribes with an error consumer
that only traces the failure: a client that is already gone is an
expected outcome. `doAfterTerminate(connection.closeSilently())` is
unchanged, so the connection is closed whether or not the notice was
written.
- The dead `try/catch` and its import are removed.

## Test


`ConnectionFactoryTestCase.testDisconnectWithNotificationToClosedClientIsNotReportedAsUnhandledError`
closes the client and waits until the server context reports
`isClosed()`. It then calls `disconnect(BUSY, "busy")` with a capturing
`RxJavaPlugins` error handler installed, and asserts that nothing
reached it. The write fails before `disconnect()` returns, so the
assertion needs no wait.

- Without the fix (`LDAPServerFilter` from master): fails with
`expecting empty, but was:<[OnErrorNotImplementedException ... |
java.io.EOFException]>`.
- With the fix: passes. The whole `opendj-grizzly` suite (1040 tests) is
green.

Fixes #1143

 opendj-grizzly/src/main/java/org/forgerock/opendj/grizzly/LDAPServerFilter.java

@@ -82,10 +82,10 @@

import com.forgerock.reactive.Action;
import com.forgerock.reactive.Completable;
import com.forgerock.reactive.Consumer;
import com.forgerock.reactive.ReactiveHandler;
import com.forgerock.reactive.Stream;

import io.reactivex.rxjava3.exceptions.OnErrorNotImplementedException;
import org.openidentityplatform.rxjava3.internal.util.BackpressureHelper;

/**
@@ -542,7 +542,19 @@
                    // handleClose() will be invoked once this connection has been closed.
                    connection.closeSilently();
                }
            }).subscribe();
            }).subscribe(new Action() {
                @Override
                public void run() throws Exception {
                    // Nothing to do: the connection is closed on either outcome.
                }
            }, new Consumer<Throwable>() {
                @Override
                public void accept(final Throwable error) throws Exception {
                    // Expected when the client has already closed its end: the notice cannot be delivered,
                    // and the connection is closed anyway.
                    logger.traceException(error);
                }
            });
        }

        private void notifyConnectionClosedRawUnbind(final LdapRequestEnvelope rawUnbindRequest) {
@@ -645,13 +657,7 @@
                    }).thenOnException(new ExceptionHandler<Exception>() {
                        @Override
                        public void handleException(Exception exception) {
                            try {
                                 s.onError(exception);
                            } catch (Throwable t) {
                                if (!(t instanceof OnErrorNotImplementedException)) {
                                    throw t;
                                }
                            }
                            s.onError(exception);
                        }
                    }).thenOnRuntimeException(new RuntimeExceptionHandler() {
                        @Override

 opendj-grizzly/src/test/java/org/forgerock/opendj/grizzly/ConnectionFactoryTestCase.java

@@ -37,7 +37,9 @@
import java.net.InetSocketAddress;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
@@ -95,6 +97,8 @@

import com.forgerock.reactive.ServerConnectionFactoryAdapter;

import io.reactivex.rxjava3.plugins.RxJavaPlugins;

/**
 * Tests the {@code ConnectionFactory} classes.
 */
@@ -665,6 +669,64 @@
        }
    }

    /**
     * A Notice of Disconnection that cannot be written because the client has already closed its end is an
     * expected outcome: it must not be reported as an unhandled error (issue #1143).
     */
    @SuppressWarnings("unchecked")
    @Test
    public void testDisconnectWithNotificationToClosedClientIsNotReportedAsUnhandledError() throws Exception {
        final CountDownLatch connectLatch = new CountDownLatch(1);
        final AtomicReference<LDAPClientContext> contextHolder = new AtomicReference<>();
        final ServerConnectionFactory<LDAPClientContext, Integer> mockServer =
                mock(ServerConnectionFactory.class);
        when(mockServer.handleAccept(any(LDAPClientContext.class))).thenAnswer(
                new Answer<ServerConnection<Integer>>() {
                    @Override
                    public ServerConnection<Integer> answer(InvocationOnMock invocation) throws Throwable {
                        contextHolder.set((LDAPClientContext) invocation.getArguments()[0]);
                        connectLatch.countDown();
                        return mock(ServerConnection.class);
                    }
                });

        final List<Throwable> unhandledErrors = new CopyOnWriteArrayList<>();
        final io.reactivex.rxjava3.functions.Consumer<? super Throwable> previousErrorHandler =
                RxJavaPlugins.getErrorHandler();
        RxJavaPlugins.setErrorHandler(new io.reactivex.rxjava3.functions.Consumer<Throwable>() {
            @Override
            public void accept(Throwable error) {
                unhandledErrors.add(error);
            }
        });
        LDAPListener listener = new LDAPListener(Collections.singleton(loopbackWithDynamicPort()),
                new ServerConnectionFactoryAdapter(Options.defaultOptions().get(LDAP_DECODE_OPTIONS), mockServer));
        try {
            final InetSocketAddress listenerAddr = listener.getSocketAddresses().iterator().next();
            final Connection client = new LDAPConnectionFactory(listenerAddr.getHostName(),
                    listenerAddr.getPort()).getConnection();
            assertThat(connectLatch.await(TEST_TIMEOUT, TimeUnit.SECONDS)).isTrue();
            final LDAPClientContext context = contextHolder.get();

            // The client leaves first: wait until the server has seen the connection close.
            client.close();
            waitForCondition(new Callable<Boolean>() {
                @Override
                public Boolean call() throws Exception {
                    return context.isClosed();
                }
            });

            // Writing the notice now fails, and does so before disconnect() returns.
            context.disconnect(ResultCode.BUSY, "busy");

            assertThat(unhandledErrors).isEmpty();
        } finally {
            RxJavaPlugins.setErrorHandler(previousErrorHandler);
            listener.close();
        }
    }

    @Test(description = "Test for OPENDJ-1121: Closing a connection after "
            + "closing the connection factory causes NPE")
    public void testFactoryCloseBeforeConnectionClose() throws Exception {