C0urante commented on code in PR #12434:
URL: https://github.com/apache/kafka/pull/12434#discussion_r930423970
##########
connect/runtime/src/test/java/org/apache/kafka/connect/runtime/rest/RestServerTest.java:
##########
@@ -338,6 +341,41 @@ public void testDisableAdminEndpoint() throws IOException {
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);
+ URI advertisedUrl = server.advertisedUrl(); // Call before capturing
logs as this creates a log
+
+ try (LogCaptureAppender restServerAppender =
LogCaptureAppender.createAndRegister(RestServer.class)) {
+ HttpRequest request = new HttpGet("/");
+ CloseableHttpClient httpClient = HttpClients.createMinimal();
+ HttpHost httpHost = new HttpHost(advertisedUrl.getHost(),
advertisedUrl.getPort());
+ CloseableHttpResponse response = httpClient.execute(httpHost,
request);
+
+ // Wait for server to handle request completion and create request
log
+ int tries = 0;
+ while (restServerAppender.getMessages().size() < 1 && tries < 10) {
+ ++tries;
+ System.err.println("Wait for log message to arrive, retry " +
tries);
+ Thread.sleep(1000);
+ }
+
+ List<String> logMessages = restServerAppender.getMessages();
+ System.err.println(logMessages);
Review Comment:
If this information could still be valuable, we should convert this line to
use an SLF4J logger instead of printing to stderr.
##########
connect/runtime/src/test/java/org/apache/kafka/connect/runtime/rest/RestServerTest.java:
##########
@@ -338,6 +341,41 @@ public void testDisableAdminEndpoint() throws IOException {
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);
+ URI advertisedUrl = server.advertisedUrl(); // Call before capturing
logs as this creates a log
+
+ try (LogCaptureAppender restServerAppender =
LogCaptureAppender.createAndRegister(RestServer.class)) {
+ HttpRequest request = new HttpGet("/");
+ CloseableHttpClient httpClient = HttpClients.createMinimal();
Review Comment:
I don't like that we're not closing this client, but this style does match
the other tests. It might be nice to have a class-level `httpClient` field
that, if non-null, gets closed at the end of each test in `tearDown`, but I
don't think it's worth blocking this PR on that change, so if you'd prefer, I
can do that in a follow-up.
##########
connect/runtime/src/test/java/org/apache/kafka/connect/runtime/rest/RestServerTest.java:
##########
@@ -338,6 +341,41 @@ public void testDisableAdminEndpoint() throws IOException {
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);
+ URI advertisedUrl = server.advertisedUrl(); // Call before capturing
logs as this creates a log
+
+ try (LogCaptureAppender restServerAppender =
LogCaptureAppender.createAndRegister(RestServer.class)) {
+ HttpRequest request = new HttpGet("/");
+ CloseableHttpClient httpClient = HttpClients.createMinimal();
+ HttpHost httpHost = new HttpHost(advertisedUrl.getHost(),
advertisedUrl.getPort());
+ CloseableHttpResponse response = httpClient.execute(httpHost,
request);
+
+ // Wait for server to handle request completion and create request
log
+ int tries = 0;
+ while (restServerAppender.getMessages().size() < 1 && tries < 10) {
+ ++tries;
+ System.err.println("Wait for log message to arrive, retry " +
tries);
+ Thread.sleep(1000);
+ }
+
+ List<String> logMessages = restServerAppender.getMessages();
+ System.err.println(logMessages);
+ String requestlog = logMessages.get(0);
Review Comment:
This may be brittle if we add other log messages to the `RestServer` class.
Do you think we can relax the constraint here to check that at least one log
message matches the expected format, but not necessarily the first one?
##########
connect/runtime/src/test/java/org/apache/kafka/connect/runtime/rest/RestServerTest.java:
##########
@@ -338,6 +341,41 @@ public void testDisableAdminEndpoint() throws IOException {
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);
+ URI advertisedUrl = server.advertisedUrl(); // Call before capturing
logs as this creates a log
+
+ try (LogCaptureAppender restServerAppender =
LogCaptureAppender.createAndRegister(RestServer.class)) {
+ HttpRequest request = new HttpGet("/");
+ CloseableHttpClient httpClient = HttpClients.createMinimal();
+ HttpHost httpHost = new HttpHost(advertisedUrl.getHost(),
advertisedUrl.getPort());
+ CloseableHttpResponse response = httpClient.execute(httpHost,
request);
+
+ // Wait for server to handle request completion and create request
log
+ int tries = 0;
+ while (restServerAppender.getMessages().size() < 1 && tries < 10) {
+ ++tries;
+ System.err.println("Wait for log message to arrive, retry " +
tries);
+ Thread.sleep(1000);
+ }
Review Comment:
Would invoking `server.stop()` be enough to guarantee that it completes the
request here? I haven't dug too deeply into the Jetty server internals yet but
it seems promising.
##########
connect/runtime/src/test/java/org/apache/kafka/connect/runtime/rest/RestServerTest.java:
##########
@@ -338,6 +341,41 @@ public void testDisableAdminEndpoint() throws IOException {
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);
+ URI advertisedUrl = server.advertisedUrl(); // Call before capturing
logs as this creates a log
+
+ try (LogCaptureAppender restServerAppender =
LogCaptureAppender.createAndRegister(RestServer.class)) {
+ HttpRequest request = new HttpGet("/");
+ CloseableHttpClient httpClient = HttpClients.createMinimal();
+ HttpHost httpHost = new HttpHost(advertisedUrl.getHost(),
advertisedUrl.getPort());
+ CloseableHttpResponse response = httpClient.execute(httpHost,
request);
Review Comment:
Same thoughts w/r/t potential resource leaks here as well.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
