This is an automated email from the ASF dual-hosted git repository. cegerton pushed a commit to branch trunk in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push: new 62914129c7d KAFKA-14099 - Fix request logging in connect (#12434) 62914129c7d is described below commit 62914129c7dbf856a7702f33ed456109c6980b0c Author: Alexandre Garnier <zig...@users.noreply.github.com> AuthorDate: Wed Oct 12 16:28:55 2022 +0200 KAFKA-14099 - Fix request logging in connect (#12434) Reviewers: Chris Egerton <chr...@aiven.io> --- build.gradle | 1 + checkstyle/import-control.xml | 1 + .../kafka/connect/runtime/rest/RestServer.java | 16 +-- .../kafka/connect/runtime/rest/RestServerTest.java | 137 ++++++++++++--------- .../unit/kafka/utils/LogCaptureAppender.scala | 6 + 5 files changed, 94 insertions(+), 67 deletions(-) diff --git a/build.gradle b/build.gradle index 642c904d766..7cf6cc44b1c 100644 --- a/build.gradle +++ b/build.gradle @@ -1281,6 +1281,7 @@ project(':clients') { testRuntimeOnly libs.jacksonJDK8Datatypes testImplementation libs.jose4j testImplementation libs.jacksonJaxrsJsonProvider + testImplementation libs.log4j generator project(':generator') } diff --git a/checkstyle/import-control.xml b/checkstyle/import-control.xml index cad049935b5..b2959d9ae6b 100644 --- a/checkstyle/import-control.xml +++ b/checkstyle/import-control.xml @@ -573,6 +573,7 @@ <allow pkg="com.fasterxml.jackson" /> <allow pkg="org.apache.http"/> <allow pkg="io.swagger.v3.oas.annotations"/> + <allow pkg="kafka.utils" /> <subpackage name="resources"> <allow pkg="org.apache.log4j" /> </subpackage> diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/RestServer.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/RestServer.java index 3c89ddb55fc..baa9b041520 100644 --- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/RestServer.java +++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/RestServer.java @@ -41,8 +41,6 @@ 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; import org.eclipse.jetty.server.handler.StatisticsHandler; import org.eclipse.jetty.servlet.FilterHolder; import org.eclipse.jetty.servlet.ServletContextHandler; @@ -190,6 +188,11 @@ public class RestServer { public void initializeServer() { log.info("Initializing REST server"); + Slf4jRequestLogWriter slf4jRequestLogWriter = new Slf4jRequestLogWriter(); + slf4jRequestLogWriter.setLoggerName(RestServer.class.getCanonicalName()); + CustomRequestLog requestLog = new CustomRequestLog(slf4jRequestLogWriter, CustomRequestLog.EXTENDED_NCSA_FORMAT + " %{ms}T"); + jettyServer.setRequestLog(requestLog); + /* Needed for graceful shutdown as per `setStopTimeout` documentation */ StatisticsHandler statsHandler = new StatisticsHandler(); statsHandler.setHandler(handlers); @@ -284,15 +287,6 @@ public class RestServer { configureHttpResponsHeaderFilter(context); } - RequestLogHandler requestLogHandler = new RequestLogHandler(); - Slf4jRequestLogWriter slf4jRequestLogWriter = new Slf4jRequestLogWriter(); - slf4jRequestLogWriter.setLoggerName(RestServer.class.getCanonicalName()); - CustomRequestLog requestLog = new CustomRequestLog(slf4jRequestLogWriter, CustomRequestLog.EXTENDED_NCSA_FORMAT + " %{ms}T"); - requestLogHandler.setRequestLog(requestLog); - - contextHandlers.add(new DefaultHandler()); - contextHandlers.add(requestLogHandler); - handlers.setHandlers(contextHandlers.toArray(new Handler[0])); try { context.start(); diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/rest/RestServerTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/rest/RestServerTest.java index 2aedb32cdf3..82f9e0395e3 100644 --- a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/rest/RestServerTest.java +++ b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/rest/RestServerTest.java @@ -16,10 +16,12 @@ */ package org.apache.kafka.connect.runtime.rest; +import kafka.utils.LogCaptureAppender; import com.fasterxml.jackson.core.type.TypeReference; import com.fasterxml.jackson.databind.ObjectMapper; import org.apache.http.HttpHost; import org.apache.http.HttpRequest; +import org.apache.http.HttpResponse; import org.apache.http.client.methods.CloseableHttpResponse; import org.apache.http.client.methods.HttpGet; import org.apache.http.client.methods.HttpOptions; @@ -44,8 +46,11 @@ import org.slf4j.LoggerFactory; import javax.ws.rs.core.MediaType; import java.io.ByteArrayOutputStream; import java.io.IOException; +import java.net.URI; import java.nio.charset.StandardCharsets; +import java.util.ArrayList; import java.util.Arrays; +import java.util.Collection; import java.util.Collections; import java.util.HashMap; import java.util.Map; @@ -60,10 +65,12 @@ import static org.mockito.Mockito.doReturn; import static org.mockito.Mockito.mock; public class RestServerTest { - + private Herder herder; private Plugins plugins; private RestServer server; + private CloseableHttpClient httpClient; + private Collection<CloseableHttpResponse> responses = new ArrayList<>(); protected static final String KAFKA_CLUSTER_ID = "Xbafgnagvar"; @@ -71,10 +78,17 @@ public class RestServerTest { public void setUp() { herder = mock(Herder.class); plugins = mock(Plugins.class); + httpClient = HttpClients.createMinimal(); } @After - public void tearDown() { + public void tearDown() throws IOException { + for (CloseableHttpResponse response: responses) { + response.close(); + } + if (httpClient != null) { + httpClient.close(); + } if (server != null) { server.stop(); } @@ -113,6 +127,7 @@ public class RestServerTest { server = new RestServer(config); Assert.assertEquals("http://localhost:8080/", server.advertisedUrl().toString()); + server.stop(); // Advertised URI from listeners with protocol configMap = new HashMap<>(baseWorkerProps()); @@ -122,6 +137,7 @@ public class RestServerTest { server = new RestServer(config); Assert.assertEquals("https://localhost:8443/", server.advertisedUrl().toString()); + server.stop(); // Advertised URI from listeners with only SSL available configMap = new HashMap<>(baseWorkerProps()); @@ -130,6 +146,7 @@ public class RestServerTest { server = new RestServer(config); Assert.assertEquals("https://localhost:8443/", server.advertisedUrl().toString()); + server.stop(); // Listener is overriden by advertised values configMap = new HashMap<>(baseWorkerProps()); @@ -141,6 +158,7 @@ public class RestServerTest { server = new RestServer(config); Assert.assertEquals("http://somehost:10000/", server.advertisedUrl().toString()); + server.stop(); // correct listener is chosen when https listener is configured before http listener and advertised listener is http configMap = new HashMap<>(baseWorkerProps()); @@ -149,6 +167,7 @@ public class RestServerTest { config = new DistributedConfig(configMap); server = new RestServer(config); Assert.assertEquals("http://plaintext-localhost:4761/", server.advertisedUrl().toString()); + server.stop(); } @Test @@ -166,12 +185,7 @@ public class RestServerTest { HttpOptions request = new HttpOptions("/connectors"); request.addHeader("Content-Type", MediaType.WILDCARD); - CloseableHttpClient httpClient = HttpClients.createMinimal(); - HttpHost httpHost = new HttpHost( - server.advertisedUrl().getHost(), - server.advertisedUrl().getPort() - ); - CloseableHttpResponse response = httpClient.execute(httpHost, request); + HttpResponse response = executeRequest(server.advertisedUrl(), request); Assert.assertEquals(MediaType.TEXT_PLAIN, response.getEntity().getContentType().getValue()); ByteArrayOutputStream baos = new ByteArrayOutputStream(); response.getEntity().writeTo(baos); @@ -196,15 +210,12 @@ public class RestServerTest { server = new RestServer(workerConfig); server.initializeServer(); server.initializeResources(herder); + URI serverUrl = server.advertisedUrl(); + HttpRequest request = new HttpGet("/connectors"); request.addHeader("Referer", origin + "/page"); request.addHeader("Origin", origin); - CloseableHttpClient httpClient = HttpClients.createMinimal(); - HttpHost httpHost = new HttpHost( - server.advertisedUrl().getHost(), - server.advertisedUrl().getPort() - ); - CloseableHttpResponse response = httpClient.execute(httpHost, request); + HttpResponse response = executeRequest(serverUrl, request); Assert.assertEquals(200, response.getStatusLine().getStatusCode()); @@ -217,7 +228,7 @@ public class RestServerTest { request.addHeader("Referer", origin + "/page"); request.addHeader("Origin", origin); request.addHeader("Access-Control-Request-Method", method); - response = httpClient.execute(httpHost, request); + response = executeRequest(serverUrl, request); Assert.assertEquals(404, response.getStatusLine().getStatusCode()); if (expectedHeader != null) { Assert.assertEquals(expectedHeader, @@ -244,9 +255,7 @@ public class RestServerTest { server.initializeServer(); server.initializeResources(herder); HttpRequest request = new HttpGet("/connectors"); - CloseableHttpClient httpClient = HttpClients.createMinimal(); - HttpHost httpHost = new HttpHost(server.advertisedUrl().getHost(), server.advertisedUrl().getPort()); - CloseableHttpResponse response = httpClient.execute(httpHost, request); + HttpResponse response = executeRequest(server.advertisedUrl(), request); Assert.assertEquals(200, response.getStatusLine().getStatusCode()); } @@ -269,12 +278,11 @@ public class RestServerTest { ObjectMapper mapper = new ObjectMapper(); - String host = server.advertisedUrl().getHost(); - int port = server.advertisedUrl().getPort(); + URI serverUrl = server.advertisedUrl(); - executePut(host, port, "/admin/loggers/a.b.c.s.W", "{\"level\": \"INFO\"}"); + executePut(serverUrl, "/admin/loggers/a.b.c.s.W", "{\"level\": \"INFO\"}"); - String responseStr = executeGet(host, port, "/admin/loggers"); + String responseStr = executeGet(serverUrl, "/admin/loggers"); Map<String, Map<String, ?>> loggers = mapper.readValue(responseStr, new TypeReference<Map<String, Map<String, ?>>>() { }); assertNotNull("expected non null response for /admin/loggers" + prettyPrint(loggers), loggers); @@ -305,12 +313,10 @@ public class RestServerTest { assertNotEquals(server.advertisedUrl(), server.adminUrl()); - executeGet(server.adminUrl().getHost(), server.adminUrl().getPort(), "/admin/loggers"); + executeGet(server.adminUrl(), "/admin/loggers"); HttpRequest request = new HttpGet("/admin/loggers"); - CloseableHttpClient httpClient = HttpClients.createMinimal(); - HttpHost httpHost = new HttpHost(server.advertisedUrl().getHost(), server.advertisedUrl().getPort()); - CloseableHttpResponse response = httpClient.execute(httpHost, request); + HttpResponse response = executeRequest(server.advertisedUrl(), request); Assert.assertEquals(404, response.getStatusLine().getStatusCode()); } @@ -332,12 +338,37 @@ public class RestServerTest { assertNull(server.adminUrl()); HttpRequest request = new HttpGet("/admin/loggers"); - CloseableHttpClient httpClient = HttpClients.createMinimal(); - HttpHost httpHost = new HttpHost(server.advertisedUrl().getHost(), server.advertisedUrl().getPort()); - CloseableHttpResponse response = httpClient.execute(httpHost, request); + HttpResponse response = executeRequest(server.advertisedUrl(), request); Assert.assertEquals(404, response.getStatusLine().getStatusCode()); } + @Test + public void testRequestLogs() throws IOException, InterruptedException { + Map<String, String> configMap = new HashMap<>(baseWorkerProps()); + DistributedConfig workerConfig = new DistributedConfig(configMap); + + doReturn(KAFKA_CLUSTER_ID).when(herder).kafkaClusterId(); + doReturn(plugins).when(herder).plugins(); + doReturn(Collections.emptyList()).when(plugins).newPlugins(Collections.emptyList(), workerConfig, ConnectRestExtension.class); + + server = new RestServer(workerConfig); + server.initializeServer(); + server.initializeResources(herder); + + LogCaptureAppender restServerAppender = LogCaptureAppender.createAndRegister(); + HttpRequest request = new HttpGet("/"); + HttpResponse response = executeRequest(server.advertisedUrl(), request); + + // Stop the server to flush all logs + server.stop(); + + Collection<String> logMessages = restServerAppender.getRenderedMessages(); + LogCaptureAppender.unregister(restServerAppender); + restServerAppender.close(); + String expectedlogContent = "\"GET / HTTP/1.1\" " + String.valueOf(response.getStatusLine().getStatusCode()); + assertTrue(logMessages.stream().anyMatch(logMessage -> logMessage.contains(expectedlogContent))); + } + @Test public void testValidCustomizedHttpResponseHeaders() throws IOException { String headerConfig = @@ -368,51 +399,45 @@ public class RestServerTest { doReturn(Arrays.asList("a", "b")).when(herder).connectors(); server = new RestServer(workerConfig); - try { - server.initializeServer(); - server.initializeResources(herder); - HttpRequest request = new HttpGet("/connectors"); - try (CloseableHttpClient httpClient = HttpClients.createMinimal()) { - HttpHost httpHost = new HttpHost(server.advertisedUrl().getHost(), server.advertisedUrl().getPort()); - try (CloseableHttpResponse response = httpClient.execute(httpHost, request)) { - Assert.assertEquals(200, response.getStatusLine().getStatusCode()); - if (!headerConfig.isEmpty()) { - expectedHeaders.forEach((k, v) -> - Assert.assertEquals(response.getFirstHeader(k).getValue(), v)); - } else { - Assert.assertNull(response.getFirstHeader("X-Frame-Options")); - } - } - } - } finally { - server.stop(); - server = null; + server.initializeServer(); + server.initializeResources(herder); + HttpRequest request = new HttpGet("/connectors"); + HttpResponse response = executeRequest(server.advertisedUrl(), request); + Assert.assertEquals(200, response.getStatusLine().getStatusCode()); + if (!headerConfig.isEmpty()) { + expectedHeaders.forEach((k, v) -> + Assert.assertEquals(response.getFirstHeader(k).getValue(), v)); + } else { + Assert.assertNull(response.getFirstHeader("X-Frame-Options")); } } - private String executeGet(String host, int port, String endpoint) throws IOException { + private String executeGet(URI serverUrl, String endpoint) throws IOException { HttpRequest request = new HttpGet(endpoint); - CloseableHttpClient httpClient = HttpClients.createMinimal(); - HttpHost httpHost = new HttpHost(host, port); - CloseableHttpResponse response = httpClient.execute(httpHost, request); + HttpResponse response = executeRequest(serverUrl, request); Assert.assertEquals(200, response.getStatusLine().getStatusCode()); return new BasicResponseHandler().handleResponse(response); } - private String executePut(String host, int port, String endpoint, String jsonBody) throws IOException { + private String executePut(URI serverUrl, String endpoint, String jsonBody) throws IOException { HttpPut request = new HttpPut(endpoint); StringEntity entity = new StringEntity(jsonBody, StandardCharsets.UTF_8.name()); entity.setContentType("application/json"); request.setEntity(entity); - CloseableHttpClient httpClient = HttpClients.createMinimal(); - HttpHost httpHost = new HttpHost(host, port); - CloseableHttpResponse response = httpClient.execute(httpHost, request); + HttpResponse response = executeRequest(serverUrl, request); Assert.assertEquals(200, response.getStatusLine().getStatusCode()); return new BasicResponseHandler().handleResponse(response); } + private HttpResponse executeRequest(URI serverUrl, HttpRequest request) throws IOException { + HttpHost httpHost = new HttpHost(serverUrl.getHost(), serverUrl.getPort()); + CloseableHttpResponse response = httpClient.execute(httpHost, request); + responses.add(response); + return response; + } + private static String prettyPrint(Map<String, ?> map) throws IOException { ObjectMapper mapper = new ObjectMapper(); return mapper.writerWithDefaultPrettyPrinter().writeValueAsString(map); diff --git a/core/src/test/scala/unit/kafka/utils/LogCaptureAppender.scala b/core/src/test/scala/unit/kafka/utils/LogCaptureAppender.scala index 2d071452829..bec77356e1e 100644 --- a/core/src/test/scala/unit/kafka/utils/LogCaptureAppender.scala +++ b/core/src/test/scala/unit/kafka/utils/LogCaptureAppender.scala @@ -20,6 +20,7 @@ package kafka.utils import org.apache.log4j.{AppenderSkeleton, Level, Logger} import org.apache.log4j.spi.LoggingEvent +import scala.jdk.CollectionConverters._ import scala.collection.mutable.ListBuffer class LogCaptureAppender extends AppenderSkeleton { @@ -37,6 +38,11 @@ class LogCaptureAppender extends AppenderSkeleton { } } + def getRenderedMessages: java.util.List[String] = { + return getMessages.map(e => e.getRenderedMessage).asJava + } + + override def close(): Unit = { events.synchronized { events.clear()