[FLINK-12838][network] netty-fy SSL configuration

Refactor the SSL configuration done for Netty to have it more like the way
Netty intends it to be: using its SslContextBuilder. This will make it much
easier to set a different Netty SSL engine provider.

[hotfix][network] extract key and trust manager factory creation

flink-mesos/src/main/java/org/apache/flink/mesos/util/MesosArtifactServer.java

@@ -126,7 +126,6 @@ public MesosArtifactServer(String prefix, String serverHostname, int configuredP
 
  router = new Router();
 
- final Configuration sslConfig = config;
  ChannelInitializer<SocketChannel> initializer = new ChannelInitializer<SocketChannel>() {
 
  @Override
@@ -135,7 +134,8 @@ protected void initChannel(SocketChannel ch) {
 
  // SSL should be the first handler in the pipeline
  if (sslFactory != null) {
- ch.pipeline().addLast("ssl", sslFactory.createNettySSLHandler());
+ ch.pipeline().addLast("ssl",
+ sslFactory.createNettySSLHandler(ch.alloc()));
  }
 
  ch.pipeline()

flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/utils/WebFrontendBootstrap.java

@@ -79,7 +79,8 @@ protected void initChannel(SocketChannel ch) {
 
  // SSL should be the first handler in the pipeline
  if (serverSSLFactory != null) {
- ch.pipeline().addLast("ssl", serverSSLFactory.createNettySSLHandler());
+ ch.pipeline().addLast("ssl",
+ serverSSLFactory.createNettySSLHandler(ch.alloc()));
  }
 
  ch.pipeline()

flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/NettyClient.java

@@ -179,6 +179,7 @@ public void initChannel(SocketChannel channel) throws Exception {
  // SSL handler should be added first in the pipeline
  if (clientSSLFactory != null) {
  SslHandler sslHandler = clientSSLFactory.createNettySSLHandler(
+ channel.alloc(),
  serverSocketAddress.getAddress().getCanonicalHostName(),
  serverSocketAddress.getPort());
  channel.pipeline().addLast("ssl", sslHandler);

flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/NettyServer.java

@@ -147,7 +147,8 @@ int init(final NettyProtocol protocol, NettyBufferPool nettyBufferPool) throws I
  @Override
  public void initChannel(SocketChannel channel) throws Exception {
  if (sslHandlerFactory != null) {
- channel.pipeline().addLast("ssl", sslHandlerFactory.createNettySSLHandler());
+ channel.pipeline().addLast("ssl",
+ sslHandlerFactory.createNettySSLHandler(channel.alloc()));
  }
 
  channel.pipeline().addLast(protocol.getServerChannelHandlers());

flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/SSLHandlerFactory.java

@@ -18,9 +18,10 @@
 
 package org.apache.flink.runtime.io.network.netty;
 
+import org.apache.flink.shaded.netty4.io.netty.buffer.ByteBufAllocator;
+import org.apache.flink.shaded.netty4.io.netty.handler.ssl.SslContext;
 import org.apache.flink.shaded.netty4.io.netty.handler.ssl.SslHandler;
 
-import javax.net.ssl.SSLContext;
 import javax.net.ssl.SSLEngine;
 
 import static java.util.Objects.requireNonNull;
@@ -30,15 +31,7 @@
  */
 public class SSLHandlerFactory {
 
- private final SSLContext sslContext;
-
- private final String[] enabledProtocols;
-
- private final String[] enabledCipherSuites;
-
- private final boolean clientMode;
-
- private final boolean clientAuthentication;
+ private final SslContext sslContext;
 
  private final int handshakeTimeoutMs;
 
@@ -54,29 +47,21 @@ public class SSLHandlerFactory {
  * default)
  */
  public SSLHandlerFactory(
- final SSLContext sslContext,
- final String[] enabledProtocols,
- final String[] enabledCipherSuites,
- final boolean clientMode,
- final boolean clientAuthentication,
+ final SslContext sslContext,
  final int handshakeTimeoutMs,
  final int closeNotifyFlushTimeoutMs) {
 
  this.sslContext = requireNonNull(sslContext, "sslContext must not be null");
- this.enabledProtocols = requireNonNull(enabledProtocols, "enabledProtocols must not be null");
- this.enabledCipherSuites = requireNonNull(enabledCipherSuites, "cipherSuites must not be null");
- this.clientMode = clientMode;
- this.clientAuthentication = clientAuthentication;
  this.handshakeTimeoutMs = handshakeTimeoutMs;
  this.closeNotifyFlushTimeoutMs = closeNotifyFlushTimeoutMs;
  }
 
- public SslHandler createNettySSLHandler() {
- return createNettySSLHandler(createSSLEngine());
+ public SslHandler createNettySSLHandler(ByteBufAllocator allocator) {
+ return createNettySSLHandler(createSSLEngine(allocator));
  }
 
- public SslHandler createNettySSLHandler(String hostname, int port) {
- return createNettySSLHandler(createSSLEngine(hostname, port));
+ public SslHandler createNettySSLHandler(ByteBufAllocator allocator, String hostname, int port) {
+ return createNettySSLHandler(createSSLEngine(allocator, hostname, port));
  }
 
  private SslHandler createNettySSLHandler(SSLEngine sslEngine) {
@@ -91,24 +76,11 @@ private SslHandler createNettySSLHandler(SSLEngine sslEngine) {
  return sslHandler;
  }
 
- private SSLEngine createSSLEngine() {
- final SSLEngine sslEngine = sslContext.createSSLEngine();
- configureSSLEngine(sslEngine);
- return sslEngine;
+ private SSLEngine createSSLEngine(ByteBufAllocator allocator) {
+ return sslContext.newEngine(allocator);
  }
 
- private SSLEngine createSSLEngine(String hostname, int port) {
- final SSLEngine sslEngine = sslContext.createSSLEngine(hostname, port);
- configureSSLEngine(sslEngine);
- return sslEngine;
- }
-
- private void configureSSLEngine(SSLEngine sslEngine) {
- sslEngine.setEnabledProtocols(enabledProtocols);
- sslEngine.setEnabledCipherSuites(enabledCipherSuites);
- sslEngine.setUseClientMode(clientMode);
- if (!clientMode) {
- sslEngine.setNeedClientAuth(clientAuthentication);
- }
+ private SSLEngine createSSLEngine(ByteBufAllocator allocator, String hostname, int port) {
+ return sslContext.newEngine(allocator, hostname, port);
  }
 }

flink-runtime/src/main/java/org/apache/flink/runtime/net/RedirectingSslHandler.java

@@ -78,7 +78,7 @@ protected void decode(ChannelHandlerContext context, ByteBuf in, List<Object> ou
  }
 
  private void handleSsl(ChannelHandlerContext context) {
- SslHandler sslHandler = sslHandlerFactory.createNettySSLHandler();
+ SslHandler sslHandler = sslHandlerFactory.createNettySSLHandler(context.alloc());
  try {
  context.pipeline().replace(this, SSL_HANDLER_NAME, sslHandler);
  } catch (Throwable t) {