root
/
kafka
mirror of https://github.com/apache/kafka.git
1
0
Fork 0
Code Issues Actions 2 Packages Projects Releases Wiki Activity

KAFKA-8316; Remove deprecated usage of Slf4jRequestLog, SslContextFactory (#6668) 

* Remove deprecated class Slf4jRequestLog: use Slf4jRequestLogWriter, CustomRequestLog instread.

1. Remove '@SuppressWarnings("deprecation")' from RestServer#initializeResources, JsonRestServer#start.
2. Remove unused JsonRestServer#httpRequest.

* Fix deprecated class usage: SslContextFactory -> SslContextFactory.[Server, Client]

1. Split SSLUtils#createSslContextFactory into SSLUtils#create[Server, Client]SideSslContextFactory: each method instantiates SslContextFactory.[Server, Client], respectively.
2. SSLUtils#configureSslContextFactoryAuthentication is called from SSLUtils#createServerSideSslContextFactory only.
3. Update SSLUtilsTest following splittion; for client-side SSL Context Factory, SslContextFactory#get[Need, Want]ClientAuth is always false. (SSLUtilsTest#testCreateClientSideSslContextFactory)

Reviewers: Ismael Juma <ismael@juma.me.uk>, Jason Gustafson <jason@confluent.io>
This commit is contained in:
Lee Dongjin 2019-05-21 02:15:15 +09:00 committed by Jason Gustafson
parent 614ea55ad7
commit b43f5446ac
5 changed files with 104 additions and 46 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/RestClient.java
View File

@ -60,7 +60,7 @@ public class RestClient {
HttpClient client;
if (url.startsWith("https://")) {
client = new HttpClient(SSLUtils.createSslContextFactory(config, true));
client = new HttpClient(SSLUtils.createClientSideSslContextFactory(config));
} else {
client = new HttpClient();
}

12
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/RestServer.java
View File

@ -32,9 +32,11 @@ import org.apache.kafka.connect.runtime.rest.resources.ConnectorsResource;
import org.apache.kafka.connect.runtime.rest.resources.RootResource;
import org.apache.kafka.connect.runtime.rest.util.SSLUtils;
import org.eclipse.jetty.server.Connector;
import org.eclipse.jetty.server.CustomRequestLog;
import org.eclipse.jetty.server.Handler;
import org.eclipse.jetty.server.Server;
import org.eclipse.jetty.server.ServerConnector;
import org.eclipse.jetty.server.Slf4jRequestLogWriter;
import org.eclipse.jetty.server.handler.ContextHandlerCollection;
import org.eclipse.jetty.server.handler.DefaultHandler;
import org.eclipse.jetty.server.handler.RequestLogHandler;
@ -146,7 +148,7 @@ public class RestServer {
ServerConnector connector;
if (PROTOCOL_HTTPS.equals(protocol)) {
SslContextFactory ssl = SSLUtils.createSslContextFactory(config);
SslContextFactory ssl = SSLUtils.createServerSideSslContextFactory(config);
connector = new ServerConnector(jettyServer, ssl);
connector.setName(String.format("%s_%s%d", PROTOCOL_HTTPS, hostname, port));
} else {
@ -181,7 +183,6 @@ public class RestServer {
log.info("REST server listening at " + jettyServer.getURI() + ", advertising URL " + advertisedUrl());
}
@SuppressWarnings("deprecation")
public void initializeResources(Herder herder) {
log.info("Initializing REST resources");
@ -217,10 +218,9 @@ public class RestServer {
}
RequestLogHandler requestLogHandler = new RequestLogHandler();
// Use fully qualified name to avoid deprecation warning
org.eclipse.jetty.server.Slf4jRequestLog requestLog = new org.eclipse.jetty.server.Slf4jRequestLog();
requestLog.setLoggerName(RestServer.class.getCanonicalName());
requestLog.setLogLatency(true);
Slf4jRequestLogWriter slf4jRequestLogWriter = new Slf4jRequestLogWriter();
slf4jRequestLogWriter.setLoggerName(RestServer.class.getCanonicalName());
CustomRequestLog requestLog = new CustomRequestLog(slf4jRequestLogWriter, CustomRequestLog.EXTENDED_NCSA_FORMAT + " %msT");
requestLogHandler.setRequestLog(requestLog);
handlers.setHandlers(new Handler[]{context, new DefaultHandler(), requestLogHandler});

34
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/util/SSLUtils.java
View File

@ -35,28 +35,33 @@ public class SSLUtils {
private static final Pattern COMMA_WITH_WHITESPACE = Pattern.compile("\\s*,\\s*");
/**
* Configures SSL/TLS for HTTPS Jetty Server / Client
* Configures SSL/TLS for HTTPS Jetty Server
*/
public static SslContextFactory createSslContextFactory(WorkerConfig config) {
return createSslContextFactory(config, false);
}
/**
* Configures SSL/TLS for HTTPS Jetty Server / Client
*/
@SuppressWarnings("deprecation")
public static SslContextFactory createSslContextFactory(WorkerConfig config, boolean client) {
public static SslContextFactory createServerSideSslContextFactory(WorkerConfig config) {
Map<String, Object> sslConfigValues = config.valuesWithPrefixAllOrNothing("listeners.https.");
SslContextFactory ssl = new SslContextFactory();
final SslContextFactory.Server ssl = new SslContextFactory.Server();
configureSslContextFactoryKeyStore(ssl, sslConfigValues);
configureSslContextFactoryTrustStore(ssl, sslConfigValues);
configureSslContextFactoryAlgorithms(ssl, sslConfigValues);
configureSslContextFactoryAuthentication(ssl, sslConfigValues);
if (client)
configureSslContextFactoryEndpointIdentification(ssl, sslConfigValues);
return ssl;
}
/**
* Configures SSL/TLS for HTTPS Jetty Client
*/
public static SslContextFactory createClientSideSslContextFactory(WorkerConfig config) {
Map<String, Object> sslConfigValues = config.valuesWithPrefixAllOrNothing("listeners.https.");
final SslContextFactory.Client ssl = new SslContextFactory.Client();
configureSslContextFactoryKeyStore(ssl, sslConfigValues);
configureSslContextFactoryTrustStore(ssl, sslConfigValues);
configureSslContextFactoryAlgorithms(ssl, sslConfigValues);
configureSslContextFactoryEndpointIdentification(ssl, sslConfigValues);
return ssl;
}
@ -141,8 +146,7 @@ public class SSLUtils {
/**
* Configures Authentication related settings in SslContextFactory
*/
@SuppressWarnings("deprecation")
protected static void configureSslContextFactoryAuthentication(SslContextFactory ssl, Map<String, Object> sslConfigValues) {
protected static void configureSslContextFactoryAuthentication(SslContextFactory.Server ssl, Map<String, Object> sslConfigValues) {
String sslClientAuth = (String) getOrDefault(sslConfigValues, BrokerSecurityConfigs.SSL_CLIENT_AUTH_CONFIG, "none");
switch (sslClientAuth) {
case "requested":

77
connect/runtime/src/test/java/org/apache/kafka/connect/runtime/rest/util/SSLUtilsTest.java
View File

@ -59,7 +59,7 @@ public class SSLUtilsTest {
}
@Test
public void testCreateSslContextFactory() {
public void testCreateServerSideSslContextFactory() {
Map<String, String> configMap = new HashMap<>(DEFAULT_CONFIG);
configMap.put("ssl.keystore.location", "/path/to/keystore");
configMap.put("ssl.keystore.password", "123456");
@ -79,7 +79,7 @@ public class SSLUtilsTest {
configMap.put("ssl.trustmanager.algorithm", "PKIX");
DistributedConfig config = new DistributedConfig(configMap);
SslContextFactory ssl = SSLUtils.createSslContextFactory(config);
SslContextFactory ssl = SSLUtils.createServerSideSslContextFactory(config);
Assert.assertEquals("file:///path/to/keystore", ssl.getKeyStorePath());
Assert.assertEquals("file:///path/to/truststore", ssl.getTrustStorePath());
@ -87,6 +87,7 @@ public class SSLUtilsTest {
Assert.assertArrayEquals(new String[] {"SSL_RSA_WITH_RC4_128_SHA", "SSL_RSA_WITH_RC4_128_MD5"}, ssl.getIncludeCipherSuites());
Assert.assertEquals("SHA1PRNG", ssl.getSecureRandomAlgorithm());
Assert.assertTrue(ssl.getNeedClientAuth());
Assert.assertFalse(ssl.getWantClientAuth());
Assert.assertEquals("JKS", ssl.getKeyStoreType());
Assert.assertEquals("JKS", ssl.getTrustStoreType());
Assert.assertEquals("TLS", ssl.getProtocol());
@ -96,7 +97,45 @@ public class SSLUtilsTest {
}
@Test
public void testCreateSslContextFactoryDefaultValues() {
public void testCreateClientSideSslContextFactory() {
Map<String, String> configMap = new HashMap<>(DEFAULT_CONFIG);
configMap.put("ssl.keystore.location", "/path/to/keystore");
configMap.put("ssl.keystore.password", "123456");
configMap.put("ssl.key.password", "123456");
configMap.put("ssl.truststore.location", "/path/to/truststore");
configMap.put("ssl.truststore.password", "123456");
configMap.put("ssl.provider", "SunJSSE");
configMap.put("ssl.cipher.suites", "SSL_RSA_WITH_RC4_128_SHA,SSL_RSA_WITH_RC4_128_MD5");
configMap.put("ssl.secure.random.implementation", "SHA1PRNG");
configMap.put("ssl.client.auth", "required");
configMap.put("ssl.endpoint.identification.algorithm", "HTTPS");
configMap.put("ssl.keystore.type", "JKS");
configMap.put("ssl.protocol", "TLS");
configMap.put("ssl.truststore.type", "JKS");
configMap.put("ssl.enabled.protocols", "TLSv1.2,TLSv1.1,TLSv1");
configMap.put("ssl.keymanager.algorithm", "SunX509");
configMap.put("ssl.trustmanager.algorithm", "PKIX");
DistributedConfig config = new DistributedConfig(configMap);
SslContextFactory ssl = SSLUtils.createClientSideSslContextFactory(config);
Assert.assertEquals("file:///path/to/keystore", ssl.getKeyStorePath());
Assert.assertEquals("file:///path/to/truststore", ssl.getTrustStorePath());
Assert.assertEquals("SunJSSE", ssl.getProvider());
Assert.assertArrayEquals(new String[] {"SSL_RSA_WITH_RC4_128_SHA", "SSL_RSA_WITH_RC4_128_MD5"}, ssl.getIncludeCipherSuites());
Assert.assertEquals("SHA1PRNG", ssl.getSecureRandomAlgorithm());
Assert.assertFalse(ssl.getNeedClientAuth());
Assert.assertFalse(ssl.getWantClientAuth());
Assert.assertEquals("JKS", ssl.getKeyStoreType());
Assert.assertEquals("JKS", ssl.getTrustStoreType());
Assert.assertEquals("TLS", ssl.getProtocol());
Assert.assertArrayEquals(new String[] {"TLSv1.2", "TLSv1.1", "TLSv1"}, ssl.getIncludeProtocols());
Assert.assertEquals("SunX509", ssl.getKeyManagerFactoryAlgorithm());
Assert.assertEquals("PKIX", ssl.getTrustManagerFactoryAlgorithm());
}
@Test
public void testCreateServerSideSslContextFactoryDefaultValues() {
Map<String, String> configMap = new HashMap<>(DEFAULT_CONFIG);
configMap.put(StandaloneConfig.OFFSET_STORAGE_FILE_FILENAME_CONFIG, "/tmp/offset/file");
configMap.put(WorkerConfig.KEY_CONVERTER_CLASS_CONFIG, "org.apache.kafka.connect.json.JsonConverter");
@ -113,7 +152,37 @@ public class SSLUtilsTest {
configMap.put("ssl.secure.random.implementation", "SHA1PRNG");
DistributedConfig config = new DistributedConfig(configMap);
SslContextFactory ssl = SSLUtils.createSslContextFactory(config);
SslContextFactory ssl = SSLUtils.createServerSideSslContextFactory(config);
Assert.assertEquals(SslConfigs.DEFAULT_SSL_KEYSTORE_TYPE, ssl.getKeyStoreType());
Assert.assertEquals(SslConfigs.DEFAULT_SSL_TRUSTSTORE_TYPE, ssl.getTrustStoreType());
Assert.assertEquals(SslConfigs.DEFAULT_SSL_PROTOCOL, ssl.getProtocol());
Assert.assertArrayEquals(Arrays.asList(SslConfigs.DEFAULT_SSL_ENABLED_PROTOCOLS.split("\\s*,\\s*")).toArray(), ssl.getIncludeProtocols());
Assert.assertEquals(SslConfigs.DEFAULT_SSL_KEYMANGER_ALGORITHM, ssl.getKeyManagerFactoryAlgorithm());
Assert.assertEquals(SslConfigs.DEFAULT_SSL_TRUSTMANAGER_ALGORITHM, ssl.getTrustManagerFactoryAlgorithm());
Assert.assertFalse(ssl.getNeedClientAuth());
Assert.assertFalse(ssl.getWantClientAuth());
}
@Test
public void testCreateClientSideSslContextFactoryDefaultValues() {
Map<String, String> configMap = new HashMap<>(DEFAULT_CONFIG);
configMap.put(StandaloneConfig.OFFSET_STORAGE_FILE_FILENAME_CONFIG, "/tmp/offset/file");
configMap.put(WorkerConfig.KEY_CONVERTER_CLASS_CONFIG, "org.apache.kafka.connect.json.JsonConverter");
configMap.put(WorkerConfig.VALUE_CONVERTER_CLASS_CONFIG, "org.apache.kafka.connect.json.JsonConverter");
configMap.put(WorkerConfig.INTERNAL_KEY_CONVERTER_CLASS_CONFIG, "org.apache.kafka.connect.json.JsonConverter");
configMap.put(WorkerConfig.INTERNAL_VALUE_CONVERTER_CLASS_CONFIG, "org.apache.kafka.connect.json.JsonConverter");
configMap.put("ssl.keystore.location", "/path/to/keystore");
configMap.put("ssl.keystore.password", "123456");
configMap.put("ssl.key.password", "123456");
configMap.put("ssl.truststore.location", "/path/to/truststore");
configMap.put("ssl.truststore.password", "123456");
configMap.put("ssl.provider", "SunJSSE");
configMap.put("ssl.cipher.suites", "SSL_RSA_WITH_RC4_128_SHA,SSL_RSA_WITH_RC4_128_MD5");
configMap.put("ssl.secure.random.implementation", "SHA1PRNG");
DistributedConfig config = new DistributedConfig(configMap);
SslContextFactory ssl = SSLUtils.createClientSideSslContextFactory(config);
Assert.assertEquals(SslConfigs.DEFAULT_SSL_KEYSTORE_TYPE, ssl.getKeyStoreType());
Assert.assertEquals(SslConfigs.DEFAULT_SSL_TRUSTSTORE_TYPE, ssl.getTrustStoreType());

25
tools/src/main/java/org/apache/kafka/trogdor/rest/JsonRestServer.java
View File

@ -23,9 +23,11 @@ import com.fasterxml.jackson.jaxrs.json.JacksonJsonProvider;
import org.apache.kafka.trogdor.common.JsonUtil;
import org.apache.kafka.trogdor.common.ThreadUtils;
import org.eclipse.jetty.server.Connector;
import org.eclipse.jetty.server.CustomRequestLog;
import org.eclipse.jetty.server.Handler;
import org.eclipse.jetty.server.Server;
import org.eclipse.jetty.server.ServerConnector;
import org.eclipse.jetty.server.Slf4jRequestLogWriter;
import org.eclipse.jetty.server.handler.DefaultHandler;
import org.eclipse.jetty.server.handler.HandlerCollection;
import org.eclipse.jetty.server.handler.RequestLogHandler;
@ -84,7 +86,6 @@ public class JsonRestServer {
*
* @param resources The path handling resources to register.
*/
@SuppressWarnings("deprecation")
public void start(Object... resources) {
log.info("Starting REST server");
ResourceConfig resourceConfig = new ResourceConfig();
@ -101,10 +102,9 @@ public class JsonRestServer {
context.addServlet(servletHolder, "/*");
RequestLogHandler requestLogHandler = new RequestLogHandler();
// Use fully qualified name to avoid deprecation warning in import statement
org.eclipse.jetty.server.Slf4jRequestLog requestLog = new org.eclipse.jetty.server.Slf4jRequestLog();
requestLog.setLoggerName(JsonRestServer.class.getCanonicalName());
requestLog.setLogLatency(true);
Slf4jRequestLogWriter slf4jRequestLogWriter = new Slf4jRequestLogWriter();
slf4jRequestLogWriter.setLoggerName(JsonRestServer.class.getCanonicalName());
CustomRequestLog requestLog = new CustomRequestLog(slf4jRequestLogWriter, CustomRequestLog.EXTENDED_NCSA_FORMAT + " %msT");
requestLogHandler.setRequestLog(requestLog);
HandlerCollection handlers = new HandlerCollection();
@ -159,21 +159,6 @@ public class JsonRestServer {
}
}
/**
* Make an HTTP request.
*
* @param url HTTP connection will be established with this url.
* @param method HTTP method ("GET", "POST", "PUT", etc.)
* @param requestBodyData Object to serialize as JSON and send in the request body.
* @param responseFormat Expected format of the response to the HTTP request.
* @param <T> The type of the deserialized response to the HTTP request.
* @return The deserialized response to the HTTP request, or null if no data is expected.
*/
public static <T> HttpResponse<T> httpRequest(String url, String method, Object requestBodyData,
TypeReference<T> responseFormat) throws IOException {
return httpRequest(log, url, method, requestBodyData, responseFormat);
}
/**
* Make an HTTP request.
*