You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@polaris.apache.org by sn...@apache.org on 2026/04/30 13:22:49 UTC
(polaris) branch main updated: Adapt CacheInvalidationSender + test to dynamic Quarkus management port (#4314)
This is an automated email from the ASF dual-hosted git repository.
snazy pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/polaris.git
The following commit(s) were added to refs/heads/main by this push:
new d0eba46767 Adapt CacheInvalidationSender + test to dynamic Quarkus management port (#4314)
d0eba46767 is described below
commit d0eba46767fe0aaaa839bfe0b5aece0a81af9d98
Author: Robert Stupp <sn...@snazy.de>
AuthorDate: Thu Apr 30 15:22:44 2026 +0200
Adapt CacheInvalidationSender + test to dynamic Quarkus management port (#4314)
In recent Quarkus versions, we cannot use both a dynamic HTTP port and a static management port. OTOH, being able to eventually use a dynamic management port is good.
The downside however is, that the `CacheInvalidationSender` needs to queue invalidation messages as long as the `quarkus.management.port' configuration value yields `0` (which is only the case for tests BTW).
The test needed to be adapted as well, mostly with an initial "test" that drains the queued invalidations.
This unblocks #4297
---
.../quarkus/distcache/CacheInvalidationSender.java | 23 +++--
.../distcache/TestCacheInvalidationSender.java | 51 +++++++----
...rsistenceDistCacheInvalidationsIntegration.java | 98 ++++++++++++++++++----
3 files changed, 134 insertions(+), 38 deletions(-)
diff --git a/persistence/nosql/persistence/cdi/quarkus-distcache/src/main/java/org/apache/polaris/persistence/nosql/quarkus/distcache/CacheInvalidationSender.java b/persistence/nosql/persistence/cdi/quarkus-distcache/src/main/java/org/apache/polaris/persistence/nosql/quarkus/distcache/CacheInvalidationSender.java
index 71db9a7de3..38f7c6dfd9 100644
--- a/persistence/nosql/persistence/cdi/quarkus-distcache/src/main/java/org/apache/polaris/persistence/nosql/quarkus/distcache/CacheInvalidationSender.java
+++ b/persistence/nosql/persistence/cdi/quarkus-distcache/src/main/java/org/apache/polaris/persistence/nosql/quarkus/distcache/CacheInvalidationSender.java
@@ -59,7 +59,7 @@ import java.util.concurrent.locks.ReentrantLock;
import org.apache.polaris.persistence.nosql.api.cache.CacheInvalidations.CacheInvalidation;
import org.apache.polaris.persistence.nosql.api.cache.DistributedCacheInvalidation;
import org.apache.polaris.persistence.nosql.api.obj.ObjRef;
-import org.eclipse.microprofile.config.inject.ConfigProperty;
+import org.eclipse.microprofile.config.ConfigProvider;
import org.jspecify.annotations.NonNull;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -82,7 +82,6 @@ class CacheInvalidationSender implements DistributedCacheInvalidation.Sender {
private final AddressResolver addressResolver;
private final List<String> serviceNames;
- private final int httpPort;
private final String invalidationUri;
private final long requestTimeout;
@@ -93,6 +92,8 @@ class CacheInvalidationSender implements DistributedCacheInvalidation.Sender {
private boolean triggered;
private final String token;
+ private int quarkusManagementPort;
+
/** Contains the IPv4/6 addresses resolved from {@link #serviceNames}. */
private volatile List<String> resolvedAddresses = emptyList();
@@ -100,7 +101,6 @@ class CacheInvalidationSender implements DistributedCacheInvalidation.Sender {
CacheInvalidationSender(
@SuppressWarnings("CdiInjectionPointsInspection") Vertx vertx,
QuarkusDistributedCacheInvalidationsConfig config,
- @ConfigProperty(name = "quarkus.management.port") int httpPort,
ServerInstanceId serverInstanceId) {
this.vertx = vertx;
@@ -112,7 +112,6 @@ class CacheInvalidationSender implements DistributedCacheInvalidation.Sender {
.toMillis();
this.httpClient = vertx.createHttpClient();
this.serviceNames = config.cacheInvalidationServiceNames().orElse(emptyList());
- this.httpPort = httpPort;
this.invalidationUri =
config.cacheInvalidationUri() + "?sender=" + serverInstanceId.instanceId();
this.serviceNameLookupIntervalMillis =
@@ -196,7 +195,7 @@ class CacheInvalidationSender implements DistributedCacheInvalidation.Sender {
try {
invalidations.add(invalidation);
- if (!triggered) {
+ if (!triggered && quarkusManagementPort() != 0) {
LOGGER.trace("Triggered invalidation submission");
vertx.executeBlocking(this::sendInvalidations);
triggered = true;
@@ -208,6 +207,7 @@ class CacheInvalidationSender implements DistributedCacheInvalidation.Sender {
private Void sendInvalidations() {
var batch = new ArrayList<CacheInvalidation>(batchSize);
+ var httpPort = quarkusManagementPort();
try {
while (true) {
lock.lock();
@@ -221,7 +221,7 @@ class CacheInvalidationSender implements DistributedCacheInvalidation.Sender {
} finally {
lock.unlock();
}
- submit(batch, resolvedAddresses);
+ submit(batch, resolvedAddresses, httpPort);
batch = new ArrayList<>(batchSize);
}
} finally {
@@ -240,9 +240,18 @@ class CacheInvalidationSender implements DistributedCacheInvalidation.Sender {
return null;
}
+ @VisibleForTesting
+ int quarkusManagementPort() {
+ if (quarkusManagementPort == 0) {
+ quarkusManagementPort =
+ ConfigProvider.getConfig().getValue("quarkus.management.port", Integer.class);
+ }
+ return quarkusManagementPort;
+ }
+
@VisibleForTesting
List<Future<Map.Entry<HttpClientResponse, Buffer>>> submit(
- List<CacheInvalidation> batch, List<String> resolvedAddresses) {
+ List<CacheInvalidation> batch, List<String> resolvedAddresses, int httpPort) {
LOGGER.trace("Submitting {} invalidations", batch.size());
String json;
diff --git a/persistence/nosql/persistence/cdi/quarkus-distcache/src/test/java/org/apache/polaris/persistence/nosql/quarkus/distcache/TestCacheInvalidationSender.java b/persistence/nosql/persistence/cdi/quarkus-distcache/src/test/java/org/apache/polaris/persistence/nosql/quarkus/distcache/TestCacheInvalidationSender.java
index 0981747107..9223c6e5fe 100644
--- a/persistence/nosql/persistence/cdi/quarkus-distcache/src/test/java/org/apache/polaris/persistence/nosql/quarkus/distcache/TestCacheInvalidationSender.java
+++ b/persistence/nosql/persistence/cdi/quarkus-distcache/src/test/java/org/apache/polaris/persistence/nosql/quarkus/distcache/TestCacheInvalidationSender.java
@@ -105,7 +105,7 @@ public class TestCacheInvalidationSender {
soft.assertThatThrownBy(
() ->
- new CacheInvalidationSender(vertx, config, 80, senderId) {
+ new CacheInvalidationSender(vertx, config, senderId) {
@Override
Future<List<String>> resolveServiceNames(List<String> serviceNames) {
return failedFuture(new RuntimeException("foo"));
@@ -113,7 +113,7 @@ public class TestCacheInvalidationSender {
@Override
List<Future<Map.Entry<HttpClientResponse, Buffer>>> submit(
- List<CacheInvalidation> batch, List<String> resolvedAddresses) {
+ List<CacheInvalidation> batch, List<String> resolvedAddresses, int httpPort) {
soft.fail("Not expected");
return null;
}
@@ -147,7 +147,7 @@ public class TestCacheInvalidationSender {
try {
CacheInvalidationSender sender =
- new CacheInvalidationSender(vertx, config, 80, senderId) {
+ new CacheInvalidationSender(vertx, config, senderId) {
@Override
Future<List<String>> resolveServiceNames(List<String> serviceNames) {
try {
@@ -173,11 +173,16 @@ public class TestCacheInvalidationSender {
@Override
List<Future<Map.Entry<HttpClientResponse, Buffer>>> submit(
- List<CacheInvalidation> batch, List<String> resolvedAddresses) {
+ List<CacheInvalidation> batch, List<String> resolvedAddresses, int httpPort) {
submitResolvedAddresses.set(resolvedAddresses);
submittedSemaphore.release();
return null;
}
+
+ @Override
+ int quarkusManagementPort() {
+ return 42;
+ }
};
// "consume" after initial, blocking call to resolveServiceNames() from the constructor
@@ -246,7 +251,7 @@ public class TestCacheInvalidationSender {
buildConfig(tokens, Optional.empty(), Duration.ofSeconds(10), Duration.ofSeconds(10));
var sender =
- new CacheInvalidationSender(vertx, config, 80, senderId) {
+ new CacheInvalidationSender(vertx, config, senderId) {
@Override
Future<List<String>> resolveServiceNames(List<String> serviceNames) {
return succeededFuture(List.of());
@@ -254,7 +259,7 @@ public class TestCacheInvalidationSender {
@Override
List<Future<Map.Entry<HttpClientResponse, Buffer>>> submit(
- List<CacheInvalidation> batch, List<String> resolvedAddresses) {
+ List<CacheInvalidation> batch, List<String> resolvedAddresses, int httpPort) {
soft.fail("Not expected");
return null;
}
@@ -292,7 +297,7 @@ public class TestCacheInvalidationSender {
var sem = new Semaphore(0);
var sender =
- new CacheInvalidationSender(vertx, config, 80, senderId) {
+ new CacheInvalidationSender(vertx, config, senderId) {
@Override
Future<List<String>> resolveServiceNames(List<String> serviceNames) {
return succeededFuture(resolvedServiceNames);
@@ -300,10 +305,15 @@ public class TestCacheInvalidationSender {
@Override
List<Future<Map.Entry<HttpClientResponse, Buffer>>> submit(
- List<CacheInvalidation> batch, List<String> resolvedAddresses) {
+ List<CacheInvalidation> batch, List<String> resolvedAddresses, int httpPort) {
sem.release(1);
return null;
}
+
+ @Override
+ int quarkusManagementPort() {
+ return 42;
+ }
};
var senderSpy = spy(sender);
@@ -311,7 +321,7 @@ public class TestCacheInvalidationSender {
invalidation.accept(senderSpy);
assertThat(sem.tryAcquire(30, TimeUnit.SECONDS)).isTrue();
- verify(senderSpy).submit(singletonList(expected), resolvedServiceNames);
+ verify(senderSpy).submit(singletonList(expected), resolvedServiceNames, 42);
}
@Test
@@ -331,7 +341,7 @@ public class TestCacheInvalidationSender {
var sem = new Semaphore(0);
var received = new ConcurrentLinkedQueue<>();
var sender =
- new CacheInvalidationSender(vertx, config, 80, senderId) {
+ new CacheInvalidationSender(vertx, config, senderId) {
@Override
Future<List<String>> resolveServiceNames(List<String> serviceNames) {
return succeededFuture(resolvedServiceNames);
@@ -339,13 +349,18 @@ public class TestCacheInvalidationSender {
@Override
List<Future<Map.Entry<HttpClientResponse, Buffer>>> submit(
- List<CacheInvalidation> batch, List<String> resolvedAddresses) {
+ List<CacheInvalidation> batch, List<String> resolvedAddresses, int httpPort) {
received.addAll(batch);
soft.assertThat(resolvedAddresses)
.containsExactlyInAnyOrderElementsOf(resolvedServiceNames);
sem.release(batch.size());
return null;
}
+
+ @Override
+ int quarkusManagementPort() {
+ return 42;
+ }
};
var senderSpy = spy(sender);
@@ -404,7 +419,7 @@ public class TestCacheInvalidationSender {
var uri = receiver.getUri();
var sender =
- new CacheInvalidationSender(vertx, config, uri.getPort(), senderId) {
+ new CacheInvalidationSender(vertx, config, senderId) {
@Override
Future<List<String>> resolveServiceNames(List<String> serviceNames) {
return succeededFuture(List.of(uri.getHost()));
@@ -413,7 +428,9 @@ public class TestCacheInvalidationSender {
var future =
CompletableFuture.allOf(
- sender.submit(singletonList(expected), singletonList(uri.getHost())).stream()
+ sender
+ .submit(singletonList(expected), singletonList(uri.getHost()), uri.getPort())
+ .stream()
.map(Future::toCompletionStage)
.map(CompletionStage::toCompletableFuture)
.toArray(CompletableFuture[]::new));
@@ -464,7 +481,7 @@ public class TestCacheInvalidationSender {
var uri = receiver.getUri();
var sender =
- new CacheInvalidationSender(vertx, config, uri.getPort(), senderId) {
+ new CacheInvalidationSender(vertx, config, senderId) {
@Override
Future<List<String>> resolveServiceNames(List<String> serviceNames) {
return succeededFuture(List.of(uri.getHost()));
@@ -472,7 +489,7 @@ public class TestCacheInvalidationSender {
};
var future =
- Future.all(sender.submit(expected, singletonList(uri.getHost())))
+ Future.all(sender.submit(expected, singletonList(uri.getHost()), uri.getPort()))
.toCompletionStage()
.toCompletableFuture();
@@ -517,7 +534,7 @@ public class TestCacheInvalidationSender {
var uri = receiver.getUri();
var sender =
- new CacheInvalidationSender(vertx, config, uri.getPort(), senderId) {
+ new CacheInvalidationSender(vertx, config, senderId) {
@Override
Future<List<String>> resolveServiceNames(List<String> serviceNames) {
return succeededFuture(List.of(uri.getHost()));
@@ -526,7 +543,7 @@ public class TestCacheInvalidationSender {
var future =
CompletableFuture.allOf(
- sender.submit(expected, singletonList(uri.getHost())).stream()
+ sender.submit(expected, singletonList(uri.getHost()), uri.getPort()).stream()
.map(Future::toCompletionStage)
.map(CompletionStage::toCompletableFuture)
.toArray(CompletableFuture[]::new));
diff --git a/runtime/service/src/test/java/org/apache/polaris/service/distcache/TestPersistenceDistCacheInvalidationsIntegration.java b/runtime/service/src/test/java/org/apache/polaris/service/distcache/TestPersistenceDistCacheInvalidationsIntegration.java
index 9b8cc63b4f..41f9628578 100644
--- a/runtime/service/src/test/java/org/apache/polaris/service/distcache/TestPersistenceDistCacheInvalidationsIntegration.java
+++ b/runtime/service/src/test/java/org/apache/polaris/service/distcache/TestPersistenceDistCacheInvalidationsIntegration.java
@@ -51,14 +51,19 @@ import org.apache.polaris.persistence.nosql.api.cache.CacheInvalidations.CacheIn
import org.apache.polaris.persistence.nosql.api.obj.SimpleTestObj;
import org.apache.polaris.service.catalog.iceberg.AbstractIcebergCatalogTest;
import org.assertj.core.api.SoftAssertions;
+import org.eclipse.microprofile.config.ConfigProvider;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.MethodOrderer;
+import org.junit.jupiter.api.Order;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestMethodOrder;
import org.junit.jupiter.api.condition.EnabledOnOs;
import org.junit.jupiter.api.condition.OS;
@QuarkusTest
@TestProfile(TestPersistenceDistCacheInvalidationsIntegration.Profile.class)
+@TestMethodOrder(MethodOrderer.OrderAnnotation.class)
// Need two IPs in this test. macOS doesn't allow binding to arbitrary 127.x.x.x addresses.
@EnabledOnOs(OS.LINUX)
@SuppressWarnings("CdiInjectionPointsInspection")
@@ -69,8 +74,7 @@ public class TestPersistenceDistCacheInvalidationsIntegration {
public Map<String, String> getConfigOverrides() {
return ImmutableMap.<String, String>builder()
.putAll(super.getConfigOverrides())
- .put("quarkus.management.port", "" + QUARKUS_MANAGEMENT_PORT)
- .put("quarkus.management.test-port", "" + QUARKUS_MANAGEMENT_PORT)
+ .put("quarkus.management.port", "0")
.put("quarkus.management.host", "127.0.0.1")
.put("quarkus.management.enabled", "true")
.put("polaris.persistence.type", "nosql")
@@ -88,16 +92,6 @@ public class TestPersistenceDistCacheInvalidationsIntegration {
static final String TOKEN = "otherToken";
static final String ENDPOINT = "/polaris-management/cache-coherency";
- // MUST be constant for test AND service
- static final int QUARKUS_MANAGEMENT_PORT = 64321;
-
- static URI CACHE_INVALIDATIONS_ENDPOINT =
- URI.create(
- format(
- "http://127.0.0.1:%d%s?sender=" + UUID.randomUUID(),
- QUARKUS_MANAGEMENT_PORT,
- ENDPOINT));
-
SoftAssertions soft;
ObjectMapper mapper;
@@ -117,12 +111,69 @@ public class TestPersistenceDistCacheInvalidationsIntegration {
soft.assertAll();
}
+ private static int quarkusManagementPort() {
+ var quarkusManagementPort =
+ ConfigProvider.getConfig().getConfigValue("quarkus.management.port");
+ var managementPortString = quarkusManagementPort.getValue();
+ return Integer.parseInt(managementPortString);
+ }
+
+ private static URI invalidationEndpoint() {
+ return URI.create(
+ format(
+ "http://127.0.0.1:%d%s?sender=" + UUID.randomUUID(),
+ quarkusManagementPort(),
+ ENDPOINT));
+ }
+
+ /**
+ * This one is not a real test, but a necessity due to the set environment setup.
+ *
+ * <p>In this test, we can no longer use a statically configured management-port but have to use a
+ * dynamically bound one. That forces the {@code
+ * org.apache.polaris.persistence.nosql.quarkus.distcache.CacheInvalidationSender} to queue
+ * invalidation messages until the actual bound management port is available from the
+ * configuration.
+ *
+ * <p>This test-code effectively just drains the queue until it is empty, so that the actual tests
+ * have a clean state.
+ */
+ @Test
+ @Order(1)
+ public void clearQueuedInvalidations() throws Exception {
+ var queue = new LinkedBlockingQueue<Map.Entry<URI, String>>();
+ try (var ignore =
+ new HttpTestServer(
+ new InetSocketAddress("127.1.2.3", quarkusManagementPort()),
+ ENDPOINT,
+ exchange -> {
+ try (InputStream requestBody = exchange.getRequestBody()) {
+ queue.add(
+ entry(exchange.getRequestURI(), new String(requestBody.readAllBytes(), UTF_8)));
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ exchange.sendResponseHeaders(204, -1);
+ exchange.getResponseBody().close();
+ })) {
+ var persistence = realmPersistenceFactory.newBuilder().realmId("warmup").build();
+ var obj = SimpleTestObj.builder().id(persistence.generateId()).text("warmup").build();
+ var expected = cacheInvalidationEvictObj(persistence.realmId(), objRef(obj));
+
+ persistence.write(obj, SimpleTestObj.class);
+
+ awaitContainsInvalidation(queue, expected);
+ }
+ }
+
@Test
+ @Order(2)
public void systemRealm() throws Exception {
sendReceive(systemPersistence);
}
@Test
+ @Order(2)
public void otherRealm() throws Exception {
var persistence = realmPersistenceFactory.newBuilder().realmId("foo").build();
sendReceive(persistence);
@@ -132,7 +183,7 @@ public class TestPersistenceDistCacheInvalidationsIntegration {
var queue = new LinkedBlockingQueue<Map.Entry<URI, String>>();
try (var ignore =
new HttpTestServer(
- new InetSocketAddress("127.1.2.3", QUARKUS_MANAGEMENT_PORT),
+ new InetSocketAddress("127.1.2.3", quarkusManagementPort()),
ENDPOINT,
exchange -> {
try (InputStream requestBody = exchange.getRequestBody()) {
@@ -201,8 +252,27 @@ public class TestPersistenceDistCacheInvalidationsIntegration {
}
}
+ private void awaitContainsInvalidation(
+ LinkedBlockingQueue<Map.Entry<URI, String>> queue, CacheInvalidation expected)
+ throws Exception {
+ var tEnd = System.nanoTime() + TimeUnit.MINUTES.toNanos(1);
+ while (System.nanoTime() < tEnd) {
+ var invalidation = queue.poll(1, TimeUnit.SECONDS);
+ if (invalidation == null) {
+ continue;
+ }
+
+ var invalidations = mapper.readValue(invalidation.getValue(), CacheInvalidations.class);
+ if (invalidations.invalidations().contains(expected)) {
+ return;
+ }
+ }
+
+ throw new AssertionError("Timed out waiting for warmup invalidation " + expected);
+ }
+
private void send(CacheInvalidations invalidations) throws Exception {
- var urlConn = CACHE_INVALIDATIONS_ENDPOINT.toURL().openConnection();
+ var urlConn = invalidationEndpoint().toURL().openConnection();
urlConn.setDoOutput(true);
urlConn.setRequestProperty("Content-Type", APPLICATION_JSON);
urlConn.setRequestProperty("Polaris-Cache-Invalidation-Token", TOKEN);