NIFI-8672 Added syncUninterruptibly() to Netty shutdown and close methods

Signed-off-by: Nathan Gough <thenatog@gmail.com>

This closes #5139.

nifi-nar-bundles/nifi-extension-utils/nifi-event-transport/src/main/java/org/apache/nifi/event/transport/netty/NettyEventSender.java

@@ -80,7 +80,7 @@ class NettyEventSender<T> implements EventSender<T> {
         try {
             channelPool.close();
         } finally {
-            group.shutdownGracefully();
+            group.shutdownGracefully().syncUninterruptibly();
         }
     }
 

nifi-nar-bundles/nifi-extension-utils/nifi-event-transport/src/main/java/org/apache/nifi/event/transport/netty/NettyEventServer.java

@@ -45,9 +45,11 @@ class NettyEventServer implements EventServer {
     @Override
     public void shutdown() {
         try {
-            channel.close().syncUninterruptibly();
+            if (channel.isOpen()) {
+                channel.close().syncUninterruptibly();
+            }
         } finally {
-            group.shutdownGracefully();
+            group.shutdownGracefully().syncUninterruptibly();
         }
     }
 }

nifi-nar-bundles/nifi-extension-utils/nifi-event-transport/src/test/java/org/apache/nifi/event/transport/netty/NettyEventSenderTest.java

@@ -0,0 +1,57 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.event.transport.netty;
+
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.pool.ChannelPool;
+import io.netty.util.concurrent.Future;
+import org.apache.nifi.remote.io.socket.NetworkUtils;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.mockito.junit.MockitoJUnitRunner;
+
+import java.net.InetSocketAddress;
+import java.net.SocketAddress;
+
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.verify;
+
+@RunWith(MockitoJUnitRunner.class)
+public class NettyEventSenderTest {
+    private static final String LOCALHOST = "127.0.0.1";
+
+    @Mock
+    private ChannelPool channelPool;
+
+    @Mock
+    private EventLoopGroup group;
+
+    @Mock
+    private Future<?> shutdownFuture;
+
+    @Test
+    public void testClose() {
+        final SocketAddress socketAddress = InetSocketAddress.createUnresolved(LOCALHOST, NetworkUtils.getAvailableTcpPort());
+        final NettyEventSender<?> sender = new NettyEventSender<>(group, channelPool, socketAddress);
+        doReturn(shutdownFuture).when(group).shutdownGracefully();
+        sender.close();
+
+        verify(channelPool).close();
+        verify(group).shutdownGracefully();
+    }
+}

nifi-nar-bundles/nifi-extension-utils/nifi-event-transport/src/test/java/org/apache/nifi/event/transport/netty/NettyEventServerTest.java

@@ -0,0 +1,57 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.event.transport.netty;
+
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.EventLoopGroup;
+import io.netty.util.concurrent.Future;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.mockito.junit.MockitoJUnitRunner;
+
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+@RunWith(MockitoJUnitRunner.class)
+public class NettyEventServerTest {
+    @Mock
+    private Channel channel;
+
+    @Mock
+    private EventLoopGroup group;
+
+    @Mock
+    private ChannelFuture closeFuture;
+
+    @Mock
+    private Future<?> shutdownFuture;
+
+    @Test
+    public void testShutdown() {
+        final NettyEventServer server = new NettyEventServer(group, channel);
+        when(channel.isOpen()).thenReturn(true);
+        when(channel.close()).thenReturn(closeFuture);
+        doReturn(shutdownFuture).when(group).shutdownGracefully();
+        server.shutdown();
+
+        verify(channel).close();
+        verify(group).shutdownGracefully();
+    }
+}

nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestListenSyslog.java

@@ -41,6 +41,7 @@ import java.util.List;
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
@@ -165,6 +166,10 @@ public class TestListenSyslog {
         sendMessages(protocol, port, LineEnding.UNIX, VALID_MESSAGE);
 
         runner.run(1, STOP_ON_FINISH_ENABLED, INITIALIZE_DISABLED);
+
+        final List<MockFlowFile> invalidFlowFiles = runner.getFlowFilesForRelationship(ListenSyslog.REL_INVALID);
+        assertTrue("Invalid FlowFiles found", invalidFlowFiles.isEmpty());
+
         final List<MockFlowFile> successFlowFiles = runner.getFlowFilesForRelationship(ListenSyslog.REL_SUCCESS);
         assertEquals("Success FlowFiles not matched", 1, successFlowFiles.size());