From cc80df19156353524f2390a541a017bf09aae3be Mon Sep 17 00:00:00 2001 From: shekharrajak Date: Fri, 5 Jun 2026 23:42:20 +0530 Subject: [PATCH 1/3] Harden Consul test readiness --- .../consul/ConsulClusterResource.java | 107 ++++++++++++++++++ 1 file changed, 107 insertions(+) diff --git a/embedded-tests/src/test/java/org/apache/druid/testing/embedded/consul/ConsulClusterResource.java b/embedded-tests/src/test/java/org/apache/druid/testing/embedded/consul/ConsulClusterResource.java index a655a3e78b0c..474b67d42e8f 100644 --- a/embedded-tests/src/test/java/org/apache/druid/testing/embedded/consul/ConsulClusterResource.java +++ b/embedded-tests/src/test/java/org/apache/druid/testing/embedded/consul/ConsulClusterResource.java @@ -31,7 +31,15 @@ import org.testcontainers.utility.DockerImageName; import javax.annotation.Nullable; +import javax.net.ssl.KeyManagerFactory; +import javax.net.ssl.SSLContext; +import javax.net.ssl.TrustManagerFactory; +import java.io.FileInputStream; import java.net.URI; +import java.net.http.HttpClient; +import java.net.http.HttpRequest; +import java.net.http.HttpResponse; +import java.security.KeyStore; import java.time.Duration; /** @@ -44,6 +52,8 @@ public class ConsulClusterResource extends TestcontainerResource createContainer() { @@ -106,6 +123,96 @@ protected GenericContainer createContainer() } } + private void waitForConsulApi() + { + final long deadline = System.nanoTime() + READINESS_TIMEOUT.toNanos(); + final HttpClient httpClient; + Exception lastException = null; + + try { + httpClient = createHttpClient(); + } + catch (Exception e) { + throw new RuntimeException("Failed to create Consul readiness client", e); + } + + while (System.nanoTime() < deadline) { + try { + final HttpRequest request = HttpRequest.newBuilder(getHttpUri("/v1/status/leader")) + .timeout(Duration.ofSeconds(5)) + .GET() + .build(); + final HttpResponse response = httpClient.send(request, HttpResponse.BodyHandlers.ofString()); + if (response.statusCode() == 200 && response.body() != null && !response.body().trim().isEmpty()) { + log.info("Consul API is ready at [%s].", getHttpUri("/v1/status/leader")); + return; + } + lastException = new RuntimeException( + StringUtils.format( + "Consul leader endpoint returned status[%d] body[%s]", + response.statusCode(), + response.body() + ) + ); + } + catch (Exception e) { + lastException = e; + } + + try { + Thread.sleep(READINESS_RETRY_DELAY.toMillis()); + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new RuntimeException("Interrupted while waiting for Consul API readiness", e); + } + } + + throw new RuntimeException( + StringUtils.format("Consul API did not become ready within [%s]", READINESS_TIMEOUT), + lastException + ); + } + + private HttpClient createHttpClient() throws Exception + { + if (securityMode == ConsulSecurityMode.PLAIN) { + return HttpClient.newBuilder() + .connectTimeout(Duration.ofSeconds(5)) + .build(); + } + + if (certBundle == null) { + throw new IllegalStateException("Consul TLS certificate bundle is not initialized"); + } + + final KeyStore trustStore = KeyStore.getInstance("PKCS12"); + try (FileInputStream fis = new FileInputStream(certBundle.getTrustStorePath())) { + trustStore.load(fis, getStorePassword().toCharArray()); + } + + final TrustManagerFactory tmf = TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm()); + tmf.init(trustStore); + + KeyManagerFactory kmf = null; + if (securityMode == ConsulSecurityMode.MTLS) { + final KeyStore keyStore = KeyStore.getInstance("PKCS12"); + try (FileInputStream fis = new FileInputStream(certBundle.getKeyStorePath())) { + keyStore.load(fis, getStorePassword().toCharArray()); + } + kmf = KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm()); + kmf.init(keyStore, getStorePassword().toCharArray()); + } + + final SSLContext sslContext = SSLContext.getInstance("TLS"); + sslContext.init(kmf == null ? null : kmf.getKeyManagers(), tmf.getTrustManagers(), null); + + return HttpClient.newBuilder() + .sslContext(sslContext) + .connectTimeout(Duration.ofSeconds(5)) + .build(); + } + @Override public void onStarted(EmbeddedDruidCluster cluster) { From 595224c57416b874aed56ee8a79d80fca8c0fcee Mon Sep 17 00:00:00 2001 From: shekharrajak Date: Sat, 6 Jun 2026 11:43:29 +0530 Subject: [PATCH 2/3] Retry Consul service announce --- .../discovery/ConsulDruidNodeAnnouncer.java | 32 ++++++++++++++++++- .../ConsulDruidNodeAnnouncerTest.java | 27 ++++++++++++++++ 2 files changed, 58 insertions(+), 1 deletion(-) diff --git a/extensions-contrib/consul-extensions/src/main/java/org/apache/druid/consul/discovery/ConsulDruidNodeAnnouncer.java b/extensions-contrib/consul-extensions/src/main/java/org/apache/druid/consul/discovery/ConsulDruidNodeAnnouncer.java index bdeb957f6a7d..237263549a40 100644 --- a/extensions-contrib/consul-extensions/src/main/java/org/apache/druid/consul/discovery/ConsulDruidNodeAnnouncer.java +++ b/extensions-contrib/consul-extensions/src/main/java/org/apache/druid/consul/discovery/ConsulDruidNodeAnnouncer.java @@ -20,12 +20,14 @@ package org.apache.druid.consul.discovery; import com.google.common.base.Preconditions; +import com.google.common.base.Throwables; import com.google.inject.Inject; import org.apache.druid.concurrent.LifecycleLock; import org.apache.druid.discovery.DiscoveryDruidNode; import org.apache.druid.discovery.DruidNodeAnnouncer; import org.apache.druid.guice.ManageLifecycle; import org.apache.druid.java.util.common.ISE; +import org.apache.druid.java.util.common.RetryUtils; import org.apache.druid.java.util.common.concurrent.Execs; import org.apache.druid.java.util.common.lifecycle.LifecycleStart; import org.apache.druid.java.util.common.lifecycle.LifecycleStop; @@ -33,6 +35,7 @@ import org.apache.druid.java.util.emitter.service.ServiceEmitter; import javax.annotation.Nullable; +import java.io.IOException; import java.util.Map; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; @@ -48,6 +51,7 @@ public class ConsulDruidNodeAnnouncer implements DruidNodeAnnouncer { private static final Logger LOGGER = new Logger(ConsulDruidNodeAnnouncer.class); + private static final int MAX_ANNOUNCE_REGISTRATION_TRIES = 3; private final ConsulApiClient consulApiClient; private final ConsulDiscoveryConfig config; @@ -159,7 +163,7 @@ public void announce(DiscoveryDruidNode discoveryDruidNode) long registerStart = System.nanoTime(); // Register in Consul, then track locally atomically in this block - consulApiClient.registerService(discoveryDruidNode); + registerServiceWithRetry(serviceId, discoveryDruidNode); announcedNodes.put(serviceId, discoveryDruidNode); long registerLatency = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - registerStart); @@ -197,6 +201,32 @@ public void announce(DiscoveryDruidNode discoveryDruidNode) } } + private void registerServiceWithRetry(String serviceId, DiscoveryDruidNode discoveryDruidNode) throws Exception + { + RetryUtils.retry( + () -> { + consulApiClient.registerService(discoveryDruidNode); + return null; + }, + ConsulDruidNodeAnnouncer::isTransientConsulFailure, + 1, + MAX_ANNOUNCE_REGISTRATION_TRIES, + null, + "Registering Consul service [" + serviceId + "] failed" + ); + } + + private static boolean isTransientConsulFailure(Throwable throwable) + { + for (Throwable cause : Throwables.getCausalChain(throwable)) { + if (cause instanceof IOException) { + return true; + } + } + + return false; + } + @Override public void unannounce(DiscoveryDruidNode discoveryDruidNode) { diff --git a/extensions-contrib/consul-extensions/src/test/java/org/apache/druid/consul/discovery/ConsulDruidNodeAnnouncerTest.java b/extensions-contrib/consul-extensions/src/test/java/org/apache/druid/consul/discovery/ConsulDruidNodeAnnouncerTest.java index 11dfe1f86064..65c4dcdb085a 100644 --- a/extensions-contrib/consul-extensions/src/test/java/org/apache/druid/consul/discovery/ConsulDruidNodeAnnouncerTest.java +++ b/extensions-contrib/consul-extensions/src/test/java/org/apache/druid/consul/discovery/ConsulDruidNodeAnnouncerTest.java @@ -22,6 +22,7 @@ import org.apache.druid.discovery.DiscoveryDruidNode; import org.apache.druid.discovery.NodeRole; import org.apache.druid.server.DruidNode; +import org.apache.http.NoHttpResponseException; import org.easymock.Capture; import org.easymock.EasyMock; import org.joda.time.Duration; @@ -95,6 +96,32 @@ public void testAnnounce() throws Exception EasyMock.verify(mockConsulApiClient); } + @Test + public void testAnnounceRetriesTransientFailure() throws Exception + { + mockConsulApiClient.registerService(EasyMock.eq(testNode)); + EasyMock.expectLastCall().andThrow(new NoHttpResponseException("Consul did not respond")); + + mockConsulApiClient.registerService(EasyMock.eq(testNode)); + EasyMock.expectLastCall().once(); + + mockConsulApiClient.passTtlCheck(EasyMock.anyString(), EasyMock.anyString()); + EasyMock.expectLastCall().anyTimes(); + + mockConsulApiClient.deregisterService(EasyMock.anyString()); + EasyMock.expectLastCall().once(); + + EasyMock.replay(mockConsulApiClient); + + announcer = new ConsulDruidNodeAnnouncer(mockConsulApiClient, config); + announcer.start(); + + announcer.announce(testNode); + announcer.stop(); + + EasyMock.verify(mockConsulApiClient); + } + @Test public void testUnannounce() throws Exception { From 17f5f2ce6a4c72a3d1303ffa135bfa91a1c6b9ca Mon Sep 17 00:00:00 2001 From: shekharrajak Date: Tue, 9 Jun 2026 23:43:15 +0530 Subject: [PATCH 3/3] Tighten Consul readiness check --- .../consul/ConsulClusterResource.java | 20 ++++++- .../consul/ConsulClusterResourceTest.java | 58 +++++++++++++++++++ 2 files changed, 77 insertions(+), 1 deletion(-) create mode 100644 embedded-tests/src/test/java/org/apache/druid/testing/embedded/consul/ConsulClusterResourceTest.java diff --git a/embedded-tests/src/test/java/org/apache/druid/testing/embedded/consul/ConsulClusterResource.java b/embedded-tests/src/test/java/org/apache/druid/testing/embedded/consul/ConsulClusterResource.java index 474b67d42e8f..8282cba4aec5 100644 --- a/embedded-tests/src/test/java/org/apache/druid/testing/embedded/consul/ConsulClusterResource.java +++ b/embedded-tests/src/test/java/org/apache/druid/testing/embedded/consul/ConsulClusterResource.java @@ -19,6 +19,7 @@ package org.apache.druid.testing.embedded.consul; +import com.fasterxml.jackson.databind.ObjectMapper; import org.apache.druid.java.util.common.StringUtils; import org.apache.druid.java.util.common.logger.Logger; import org.apache.druid.testing.embedded.EmbeddedDruidCluster; @@ -35,6 +36,7 @@ import javax.net.ssl.SSLContext; import javax.net.ssl.TrustManagerFactory; import java.io.FileInputStream; +import java.io.IOException; import java.net.URI; import java.net.http.HttpClient; import java.net.http.HttpRequest; @@ -49,6 +51,7 @@ public class ConsulClusterResource extends TestcontainerResource> { private static final Logger log = new Logger(ConsulClusterResource.class); + private static final ObjectMapper JSON_MAPPER = new ObjectMapper(); private static final int CONSUL_HTTP_PORT = 8500; private static final int CONSUL_HTTPS_PORT = 8501; private static final DockerImageName CONSUL_IMAGE = DockerImageName.parse("hashicorp/consul:1.18"); @@ -143,7 +146,7 @@ private void waitForConsulApi() .GET() .build(); final HttpResponse response = httpClient.send(request, HttpResponse.BodyHandlers.ofString()); - if (response.statusCode() == 200 && response.body() != null && !response.body().trim().isEmpty()) { + if (isConsulLeaderReady(response.statusCode(), response.body())) { log.info("Consul API is ready at [%s].", getHttpUri("/v1/status/leader")); return; } @@ -174,6 +177,21 @@ private void waitForConsulApi() ); } + static boolean isConsulLeaderReady(int statusCode, @Nullable String body) + { + if (statusCode != 200 || body == null || body.trim().isEmpty()) { + return false; + } + + try { + final String leader = JSON_MAPPER.readValue(body, String.class); + return leader != null && !leader.trim().isEmpty(); + } + catch (IOException e) { + return false; + } + } + private HttpClient createHttpClient() throws Exception { if (securityMode == ConsulSecurityMode.PLAIN) { diff --git a/embedded-tests/src/test/java/org/apache/druid/testing/embedded/consul/ConsulClusterResourceTest.java b/embedded-tests/src/test/java/org/apache/druid/testing/embedded/consul/ConsulClusterResourceTest.java new file mode 100644 index 000000000000..2f80e6cfb201 --- /dev/null +++ b/embedded-tests/src/test/java/org/apache/druid/testing/embedded/consul/ConsulClusterResourceTest.java @@ -0,0 +1,58 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.druid.testing.embedded.consul; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +public class ConsulClusterResourceTest +{ + @Test + public void testLeaderEndpointWithLeaderIsReady() + { + Assertions.assertTrue(ConsulClusterResource.isConsulLeaderReady(200, "\"127.0.0.1:8300\"")); + } + + @Test + public void testLeaderEndpointWithoutLeaderIsNotReady() + { + Assertions.assertFalse(ConsulClusterResource.isConsulLeaderReady(200, "\"\"")); + } + + @Test + public void testBlankBodyIsNotReady() + { + Assertions.assertFalse(ConsulClusterResource.isConsulLeaderReady(200, "")); + Assertions.assertFalse(ConsulClusterResource.isConsulLeaderReady(200, " ")); + Assertions.assertFalse(ConsulClusterResource.isConsulLeaderReady(200, null)); + } + + @Test + public void testNonOkStatusIsNotReady() + { + Assertions.assertFalse(ConsulClusterResource.isConsulLeaderReady(503, "\"127.0.0.1:8300\"")); + } + + @Test + public void testMalformedBodyIsNotReady() + { + Assertions.assertFalse(ConsulClusterResource.isConsulLeaderReady(200, "127.0.0.1:8300")); + } +}