RMQSink extends RichSinkFunction<IN> { @Override public void close() { Exception t = null; try { if (channel != null) { channel.close(); } } catch (IOException | TimeoutException e) { t = e; } try { if (connection != null) { connection.close(); } } catch (IOException e) { if (t != null) { LOG.warn("Both channel and connection closing failed. Logging channel exception and failing with connection exception", t); } t = e; } if (t != null) { throw new RuntimeException("Error while closing RMQ connection with " + queueName + " at " + rmqConnectionConfig.getHost(), t); } } RMQSink(RMQConnectionConfig rmqConnectionConfig, String queueName, SerializationSchema<IN> schema); void setLogFailuresOnly(boolean logFailuresOnly); @Override void open(Configuration config); @Override void invoke(IN value); @Override void close();  }
@Test public void closeAllResources() throws Exception { RMQSink<String> rmqSink = createRMQSink(); rmqSink.close(); verify(channel).close(); verify(connection).close(); }