@Test
  public void connectionWithHungReaderIsCloseableAndUnhangsReader()
      throws InterruptedException, TimeoutException {

    blackboard.clearGate(UPDATE_ENTERED_GATE);
    blackboard.clearGate(SUSPEND_UPDATE_GATE);

    final int locatorPort = createLocator(locator);
    createCacheAndRegion(sender, locatorPort);
    createCacheAndRegion(receiver, locatorPort);

    receiver
        .invoke("set up DistributionMessageObserver to 'hang' sender's put (on receiver)",
            () -> {
              final DistributionMessageObserver observer =
                  new DistributionMessageObserver() {

                    @Override
                    public void beforeProcessMessage(final ClusterDistributionManager dm,
                        final DistributionMessage message) {
                      guardMessageProcessingHook(message, () -> {
                        try {
                          blackboard.signalGate(UPDATE_ENTERED_GATE);
                          blackboard.waitForGate(SUSPEND_UPDATE_GATE);
                        } catch (TimeoutException | InterruptedException e) {
                          errorCollector.addError(e);
                        }
                      });
                    }
                  };
              DistributionMessageObserver.setInstance(observer);
            });

    final AsyncInvocation<Object> putInvocation = sender.invokeAsync("try a put", () -> {
      final Region<Object, Object> region = cache.getRegion(regionName);
      // test is going to close the cache while we are waiting for our ack
      assertThatThrownBy(() -> {
        region.put("hello", "world");
      }).isInstanceOf(DistributedSystemDisconnectedException.class);
    });

    // wait until our message observer is blocked
    blackboard.waitForGate(UPDATE_ENTERED_GATE);

    // at this point our put() is blocked waiting for a direct ack
    assertThat(putInvocation.isAlive()).as("put is waiting for remote region to ack").isTrue();

    /*
     * Now close the cache. The point of calling it is to test that we don't block while trying
     * to close connections. Cache.close() calls DistributedSystem.disconnect() which in turn
     * closes all the connections (and their sockets.) We want the sockets to close because that'll
     * cause our hung put() to see a DistributedSystemDisconnectedException.
     */
    sender.invoke("", () -> cache.close());

    // wait for put task to complete: with an exception, that is!
    putInvocation.get();

    // un-stick our message observer
    blackboard.signalGate(SUSPEND_UPDATE_GATE);
  }
It would be nice if this test didn't need to use the cache since the test's purpose is to test
that the {@link Connection} class can be closed while readers and writers hold locks on its
internal TLS {@link ByteBuffer}s

But this test does use the cache (region) because it enabled us to use existing cache messaging
and to use the DistributionMessageObserver (observer) hooks.

see also ClusterCommunicationsDUnitTest
geode-core/src/distributedTest/java/org/apache/geode/internal/tcp/ConnectionCloseSSLTLSDUnitTest.java