diff --git a/CHANGELOG.md b/CHANGELOG.md index 59153ed5..510b85a3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,10 @@ 1. [#250](https://github.com/InfluxCommunity/influxdb3-java/pull/250) Upgrade Netty version to 4.2.3.Final. 2. [#251](https://github.com/InfluxCommunity/influxdb3-java/pull/251) Add comment warning null when calling getMeasurement function. +### Documentation + +1. [#253](https://github.com/InfluxCommunity/influxdb3-java/pull/253) New Durable example showing client reuse for better resource management. + ## 1.2.0 [2025-06-26] ### Features diff --git a/examples/README.md b/examples/README.md index 4e1833a2..79291016 100644 --- a/examples/README.md +++ b/examples/README.md @@ -30,3 +30,30 @@ mvn compile exec:java -Dexec.main="com.influxdb.v3.RetryExample" ``` - Repeat previous step to force an HTTP 429 response and rewrite attempt. + +## Durable example + +This example illustrates one approach to ensuring clients, once initialized, are long-lived and reused. + +The underlying write (HTTP/REST) and query (Apache arrow Flight/GRPC) transports are designed to be robust and to be able to recover from most errors. The InfluxDBClient query API is based on GRPC stubs and channels. [GRPC best practices](https://grpc.io/docs/guides/performance/) recommends reusing them and their resources for the life of an application if at all possible. Unnecessary frequent regeneration of InfluxDBClient instances is wasteful of system resources. Recreating the query transport means fully recreating a GRPC channel, its connection pool and its management API. Fully recreating a client only to use it for a single query also means recreating an unused write transport alongside the query transport. This example attempts to show a more resource friendly use of the API by leveraging already used client instances. + +- [DurableExample](src/main/java/com/influxdb/v3/durable/DurableExample.java) +- [InfluxClientPool](src/main/java/com/influxdb/v3/durable/InfluxClientPool.java) + +### Command line run + +- Set environment variables + +```bash + +export INFLUX_HOST= +export INFLUX_TOKEN= +export INFLUX_DATABASE= + +``` + +- Run with maven + +```bash +MAVEN_OPTS="--add-opens=java.base/java.nio=org.apache.arrow.memory.core,ALL-UNNAMED" mvn compile exec:java -Dexec.main="com.influxdb.v3.durable.DurableExample" +``` diff --git a/examples/src/main/java/com/influxdb/v3/durable/DurableExample.java b/examples/src/main/java/com/influxdb/v3/durable/DurableExample.java new file mode 100644 index 00000000..55766170 --- /dev/null +++ b/examples/src/main/java/com/influxdb/v3/durable/DurableExample.java @@ -0,0 +1,264 @@ +/* + * The MIT License + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN + * THE SOFTWARE. + */ +package com.influxdb.v3.durable; + +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.locks.LockSupport; +import java.util.logging.Logger; +import java.util.stream.Stream; + +import com.influxdb.v3.client.InfluxDBApiException; +import com.influxdb.v3.client.InfluxDBClient; +import com.influxdb.v3.client.Point; +import com.influxdb.v3.client.PointValues; +import com.influxdb.v3.client.config.ClientConfig; + +/** + * The example depends on the "influxdb3-java" module and this module should be built first + * by running "mvn install" in the root directory. + *

+ * This example illustrates how to reuse InfluxDBClient instances. The underlying write (REST) and + * query (apache arrow Flight/GRPC) transports are designed to be robust and long-lived. Frequent creation or + * recreation of InfluxDBClients and then discarding them and their underlying transports is + * inefficient. GRPC best practices recommends trying to use their channels, on which the InfluxDBClient + * query transport is based, for the life of an application, if at all possible. The write transport is + * also designed to recover from most errors and to be reusable. This example is one approach to reusing + * InfluxDBClient instances for as long as possible. + *

+ * At its core this example uses a client pool and four processing threads. The threads borrow + * clients from the pool as needed and then return them once they are no longer needed. Two threads + * are used for writing data and two additional threads are used for executing queries. + * One write thread is designed to occasionally force an error response from the server. Like wise one + * query thread is designed to occasionally elicit error responses in the GRPC channel. Even though + * errors occur in these transactions, the clients involved can continue to be used for later writes + * and queries. Furthermore, while four processing threads are running, the pool need only instantiate three + * clients, if handled properly. + */ +public final class DurableExample { + + static Logger logger = Logger.getLogger(DurableExample.class.getName()); + + public static ClientConfig clientConfig; + + private DurableExample() { + } + + public static void setup() { + + String influxHost = System.getenv("INFLUX_HOST") != null + ? System.getenv("INFLUX_HOST") : "http://localhost:8181"; + String influxToken = System.getenv("INFLUX_TOKEN") != null + ? System.getenv("INFLUX_TOKEN") : "my-token"; + String influxDatabase = System.getenv("INFLUX_DATABASE") != null + ? System.getenv("INFLUX_DATABASE") : "my-db"; + + clientConfig = new ClientConfig.Builder() + .host(influxHost) + .token(influxToken.toCharArray()) + .database(influxDatabase) + .build(); + } + + public static void main(final String[] args) { + + setup(); + + // A basic control signal + AtomicBoolean shutdownAll = new AtomicBoolean(false); + + // time to run the example in minutes + int runTime = 2; + + // a set of sensors as a source of data + List sensors = List.of( + new Sensor("Alfa", "Univac51", "libava"), + new Sensor("Bravo", "Eniac45", "brezina"), + new Sensor("Charlie", "Ordvac52", "boletice"), + new Sensor("Delta", "HAL2001", "hradiste"), + new Sensor("Echo", "BESM68", "brdy") + ); + + // standard query string + String query = String.format("SELECT * FROM %s ORDER BY time DESC", Sensor.class.getSimpleName()); + + // query string to elicit error response + String badQuery = String.format("SELECT * FOO %s ORDER BY time DESC", Sensor.class.getSimpleName()); + + // Set up the autoclosable client pool + try (InfluxClientPool clientPool = new InfluxClientPool(clientConfig)) { + + // thread controller + final ExecutorService executors = Executors.newFixedThreadPool(5); + + // an error free write thread + final Runnable writeOK = () -> { + int count = 0; + while (!shutdownAll.get()) { + List points = new ArrayList<>(); + for (Sensor sensor : sensors) { + points.add(sensor.randomPoint().toPoint()); + } + + // borrow then return a client + InfluxDBClient client = clientPool.borrowClient(); + try { + logger.info(" [writeTaskPointsOK " + count + "] Writing " + points.size() + + " points with client " + client.hashCode()); + client.writePoints(points); + } catch (Exception e) { + logger.severe(" [writeTaskPointsOK " + count + "] Unexpected Error writing points " + + e.getMessage()); + } finally { + clientPool.returnClient(client); + } + + LockSupport.parkNanos(TimeUnit.SECONDS.toNanos(10)); + count++; + } + logger.info(" [writeTaskPointsOK] shutting down"); + }; + + // An error-prone write thread + final Runnable writeErrorRecover = () -> { + // delay start by 2 seconds + LockSupport.parkNanos(TimeUnit.SECONDS.toNanos(2)); + int count = 0; + while (!shutdownAll.get()) { + List lps = new ArrayList<>(); + for (Sensor sensor : sensors) { + // every fourth write attempt uses an invalid Line protocol line + if (count > 0 && count % 4 == 0 && sensor.getName().equals("Charlie")) { + // add the invalid LP line + lps.add(sensor.randomPoint().toLPBroken()); + } else { + lps.add(sensor.randomPoint().toLP()); + } + } + // borrow a client from the pool + InfluxDBClient client = clientPool.borrowClient(); + try { + logger.info("[writeErrorRecover " + count + "] Writing " + lps.size() + + " lps with client " + client.hashCode()); + client.writeRecords(lps); + } catch (InfluxDBApiException ie) { + logger.warning("[writeErrorRecover " + count + "] Write Error " + ie.getMessage()); + } finally { + // make sure the client is returned to the pool even after an error + clientPool.returnClient(client); + } + LockSupport.parkNanos(TimeUnit.SECONDS.toNanos(10)); + count++; + } + logger.info(" [writeErrorRecover] shutting down"); + }; + + // an error free query thread + final Runnable queryOK = () -> { + // delay start by 4 seconds + LockSupport.parkNanos(TimeUnit.SECONDS.toNanos(4)); + int count = 0; + while (!shutdownAll.get()) { + // borrow a client from the pool + InfluxDBClient client = clientPool.borrowClient(); + + // initiate the query and process the results + try (Stream pvs = client.queryPoints(query)) { + logger.info("[queryOK " + count + "] with client " + client.hashCode() + + ": query returned " + pvs.toArray().length + " records"); + } catch (Exception e) { + logger.severe("[queryOK " + count + "] unexpected query Error " + e.getMessage()); + } finally { + // ensure the client is returned to the pool + clientPool.returnClient(client); + } + LockSupport.parkNanos(TimeUnit.SECONDS.toNanos(10)); + count++; + } + logger.info(" [queryOK] shutting down"); + }; + + // an error-prone query thread + final Runnable queryErrorRecover = () -> { + // delay start by 6 seconds + LockSupport.parkNanos(TimeUnit.SECONDS.toNanos(6)); + int count = 0; + while (!shutdownAll.get()) { + // borrow a client from the pool + InfluxDBClient client = clientPool.borrowClient(); + // every third query attempt results in an error + String effectiveQuery = count > 0 && count % 3 == 0 ? badQuery : query; + + // attempt to execute the query and process the results + try (Stream pvs = client.queryPoints(effectiveQuery)) { + logger.info("[queryErrorRecover " + count + "] with client " + client.hashCode() + + ": query returned " + pvs.toArray().length + " records"); + } catch (Exception e) { + logger.warning("[queryErrorRecover " + count + "] with client " + client.hashCode() + + ": query Error " + e.getMessage()); + } finally { + // ensure the client is returned to the pool even after an error + clientPool.returnClient(client); + } + LockSupport.parkNanos(TimeUnit.SECONDS.toNanos(10)); + count++; + } + logger.info(" [queryErrorRecover] shutting down"); + }; + + // control how long the example runs + final Runnable timer = () -> { + LockSupport.parkNanos(TimeUnit.MINUTES.toNanos(runTime)); + shutdownAll.set(true); + logger.info(" [timer] Shutting down"); + logger.info("clientPool clients: active " + + clientPool.activeCount() + + " idle: " + clientPool.idleCount()); + executors.shutdown(); + }; + + // trigger all threads + executors.execute(writeOK); + executors.execute(writeErrorRecover); + executors.execute(queryOK); + executors.execute(queryErrorRecover); + executors.execute(timer); + + // ensure termination + boolean returned; + try { + returned = executors.awaitTermination(runTime + 1, TimeUnit.MINUTES); + } catch (InterruptedException e) { + throw new RuntimeException(e); + } + + logger.info("executors terminated cleanly: " + returned); + + } catch (Exception e) { + throw new RuntimeException(e); + } + } +} diff --git a/examples/src/main/java/com/influxdb/v3/durable/InfluxClientPool.java b/examples/src/main/java/com/influxdb/v3/durable/InfluxClientPool.java new file mode 100644 index 00000000..159f3772 --- /dev/null +++ b/examples/src/main/java/com/influxdb/v3/durable/InfluxClientPool.java @@ -0,0 +1,165 @@ +/* + * The MIT License + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN + * THE SOFTWARE. + */ +package com.influxdb.v3.durable; + +import java.util.ArrayList; +import java.util.List; +import java.util.Stack; +import java.util.logging.Logger; + +import com.influxdb.v3.client.InfluxDBClient; +import com.influxdb.v3.client.config.ClientConfig; + +/** + * And example pool for InfluxDBClient clients. + *

+ * All clients handled by the pool will share the same basic configuration. + */ +public class InfluxClientPool implements AutoCloseable { + + Logger logger = Logger.getLogger(InfluxClientPool.class.getName()); + + private static final int DEFAULT_MAX_SIZE = 4; + + // Container for clients waiting to be used. + Stack idlers = new Stack<>(); + // Container for clients currently in use. + List runners = new ArrayList<>(); + + int maxSize; + + // The shared configuration. + final ClientConfig clientConfig; + + /** + * Basic Constructor, uses DEFAULT_MAX_SIZE for maxSize. + *

+ * @param clientConfig - the standard configuration for all clients managed by the pool. + */ + public InfluxClientPool(final ClientConfig clientConfig) { + this(clientConfig, DEFAULT_MAX_SIZE); + } + + /** + * Basic constructor. + *

+ * @param clientConfig - the standard configuration for all clients managed by the pool. + */ + public InfluxClientPool(final ClientConfig clientConfig, final int maxSize) { + this.clientConfig = clientConfig; + this.maxSize = maxSize; + } + + /** + * Checks for a free client in the idle stack. If the idle stack + * is empty, it generates a new client. + * + * @return - An InfluxDBClient ready for use. + */ + public synchronized InfluxDBClient borrowClient() { + InfluxDBClient client; + if (idlers.isEmpty()) { + client = InfluxDBClient.getInstance(clientConfig); + runners.add(client); + if (activeCount() >= maxSize) { + // N.B. this is just an example implementation. + // For simplicity this _example_ will allow the maxSize value to be exceeded with a severe warning. + // In a production environment maxSize should be managed appropriately. + logger.severe("Max pool size " + maxSize + " exceeded: " + "actives " + + activeCount() + " idles " + idleCount() + + " (hint: Is there a process hogging zombie clients?)"); + } + } else { + client = idlers.pop(); + runners.add(client); + } + logger.info("Lending client " + client.hashCode()); + return client; + } + + /** + * Invalidate a client if some unwanted exception state is encountered, + * or if for some other reason it is unusable or no longer needed. + * + * @param client - client to be closed and flagged for garbage collection. + */ + public synchronized void invalidateClient(final InfluxDBClient client) { + runners.remove(client); + try { + client.close(); + } catch (Exception e) { + logger.warning("Exception occurred when invalidating client " + + client.hashCode() + ": " + e.getMessage()); + } + } + + /** + * Return the client to the idle stack within the pool + * when it is no longer needed but can still be reused. + * + * @param client - the client to be returned. + */ + public synchronized void returnClient(final InfluxDBClient client) { + logger.info("Returning client " + client.hashCode()); + runners.remove(client); + idlers.push(client); + } + + /** + * Handle closing all resources. + *

+ * First return active clients to the idle stack. + * Then close all clients in the idle stack. + * + * @throws Exception + */ + @Override + public synchronized void close() throws Exception { + logger.info("Closing client pool"); + int stillActiveCount = activeCount(); + for (int i = stillActiveCount - 1; i >= 0; i--) { + returnClient(runners.get(i)); + } + while (!idlers.isEmpty()) { + try (InfluxDBClient client = idlers.pop()) { + logger.info("Closing client " + client.hashCode()); + } catch (IllegalStateException e) { + StringBuilder msg = new StringBuilder("IllegalStateException when closing client. "); + msg.append(e.getMessage()); + if (e.getMessage().contains("leaked")) { + msg.append(" (hint: were all streams returned from queries closed correctly?)"); + } + logger.warning(msg.toString()); + } catch (Exception e) { // client close should be automatic + logger.warning("Exception when closing client " + e.getMessage()); + } + } + } + + public synchronized int activeCount() { + return runners.size(); + } + + public synchronized int idleCount() { + return idlers.size(); + } +} diff --git a/examples/src/main/java/com/influxdb/v3/durable/Sensor.java b/examples/src/main/java/com/influxdb/v3/durable/Sensor.java new file mode 100644 index 00000000..0bf1d7cb --- /dev/null +++ b/examples/src/main/java/com/influxdb/v3/durable/Sensor.java @@ -0,0 +1,211 @@ +/* + * The MIT License + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN + * THE SOFTWARE. + */ +package com.influxdb.v3.durable; + +import java.time.Instant; + +import com.influxdb.v3.client.Point; + +/** + * A generic Sensor. + */ +public class Sensor { + + long id; + String name; + String model; + String location; + + public Sensor(final String name, final String model, final String location) { + this.id = Math.round(Math.random() * 1000000); + this.name = name; + this.model = model; + this.location = location; + } + + public long getId() { + return id; + } + + public String getName() { + return name; + } + + public String getModel() { + return model; + } + + public String getLocation() { + return location; + } + + @Override + public String toString() { + return String.format("[%d] name: %s model: %s location: %s", id, name, model, location); + } + + /** + * Creates random data for illustrative or testing purposes only. + * + * @param timestamp - a timestamp for the data point. If null returns now. + * @return A Sensor.DataPoint + */ + public Sensor.DataPoint randomPoint(final Instant timestamp) { + return new Sensor.DataPoint(this, randomAccel(), randomVel(), randomBearing(), + randomAccel(), randomVel(), randomBearing(), timestamp); + } + + /** + * Creates a random data point with Instant.now(). + * + * @return A Sensor.DataPoint. + */ + public Sensor.DataPoint randomPoint() { + return randomPoint(Instant.now()); + } + + private static double randomAccel() { + return Math.random(); + } + + private static double randomVel() { + return Math.random() + Math.random(); + } + + private static double randomBearing() { + return Math.random() * 360.0; + } + + + public static class DataPoint { + Sensor sensor; + + double hAccel; + double hVel; + double hBearing; + double vAccel; + double vVel; + double vBearing; + Instant timestamp; + + String lpFormat = "%s,id=%s,location=%s,model=%s,name=%s " + + "hAccel=%.6f,hBearing=%.6f,hVel=%.6f,vAccel=%.6f,vBearing=%.6f,vVel=%.6f %d"; + + /** + * Used for illustrating client fail over when incorrect line protocol values are sent. + */ + String lpFormatBroken = "%s,id=%s,location=%s,model=%s,name=%s " + + "hAccel=,hBearing=%.6f,hVel=%.6f,vAccel=%.6f,vBearing=%.6f,vVel=%.6f %d"; + + + public DataPoint(final Sensor sensor, + final double hAccel, + final double hVel, + final double hBearing, + final double vAccel, + final double vVel, + final double vBearing, + final Instant timestamp) { + this.sensor = sensor; + this.hAccel = hAccel; + this.hVel = hVel; + this.hBearing = hBearing; + this.vAccel = vAccel; + this.vVel = vVel; + this.vBearing = vBearing; + this.timestamp = timestamp == null ? Instant.now() : timestamp; + } + + @Override + public String toString() { + + return String.format("%s - hAccel: %.3f, hVel: %.3f, hBearing: %.3f, vAccel: %.3f, vVel: %.3f, vBearing: %.3f %s", + sensor, hAccel, hVel, hBearing, vAccel, vVel, vBearing, timestamp); + } + + /** + * Converts a Sensor.DataPoint to an InfluxDBClient v3 Point. + * + * @return - an InfluxDBClient v3 Point. + */ + public Point toPoint() { + return Point.measurement(sensor.getClass().getSimpleName().toLowerCase()) + .setTag("name", sensor.getName()) + .setTag("model", sensor.getModel()) + .setTag("location", sensor.getLocation()) + .setTag("id", Long.toString(sensor.getId())) + .setFloatField("hAccel", hAccel) + .setFloatField("hVel", hVel) + .setFloatField("hBearing", hBearing) + .setFloatField("vAccel", vAccel) + .setFloatField("vVel", vVel) + .setFloatField("vBearing", vBearing) + .setTimestamp(timestamp); + } + + /** + * Convert the Sensor.DataPoint to a valid Line protocol string. + * + * @return - a valid Line protocol string. + */ + public String toLP() { + long nanos = timestamp.getEpochSecond() * 1_000_000_000 + timestamp.getNano(); + return String.format(lpFormat, + sensor.getClass().getSimpleName().toLowerCase(), + sensor.getId(), + sensor.getLocation(), + sensor.getModel(), + sensor.getName(), + hAccel, + hBearing, + hVel, + vAccel, + vBearing, + vVel, + nanos + ); + } + + /** + * To be used to illustrate how InfluxDBClient recovers from + * submitting incorrect data to InfluxDB. + * + * @return - an invalid Line protocol string. + */ + public String toLPBroken() { + long nanos = timestamp.getEpochSecond() * 1_000_000_000 + timestamp.getNano(); + return String.format(lpFormatBroken, + sensor.getClass().getSimpleName().toLowerCase(), + sensor.getId(), + sensor.getLocation(), + sensor.getModel(), + sensor.getName(), + hBearing, + hVel, + vAccel, + vBearing, + vVel, + nanos + ); + } + } +} diff --git a/src/test/java/com/influxdb/v3/client/query/QueryOptionsTest.java b/src/test/java/com/influxdb/v3/client/query/QueryOptionsTest.java index fe94db3e..845c9522 100644 --- a/src/test/java/com/influxdb/v3/client/query/QueryOptionsTest.java +++ b/src/test/java/com/influxdb/v3/client/query/QueryOptionsTest.java @@ -21,6 +21,8 @@ */ package com.influxdb.v3.client.query; +import java.io.IOException; +import java.net.ServerSocket; import java.net.URI; import java.nio.charset.StandardCharsets; import java.util.ArrayList; @@ -64,6 +66,13 @@ class QueryOptionsTest { private ClientConfig.Builder configBuilder; + private static int findFreePort() throws IOException { + ServerSocket s = new ServerSocket(0); + int port = s.getLocalPort(); + s.close(); + return port; + } + @BeforeEach void before() { configBuilder = new ClientConfig.Builder() @@ -109,7 +118,8 @@ void optionsOverrideQueryType() { @Test void setInboundMessageSizeSmall() throws Exception { - URI uri = URI.create("http://127.0.0.1:33333"); + int freePort = findFreePort(); + URI uri = URI.create("http://127.0.0.1:" + freePort); int rowCount = 100; try (VectorSchemaRoot vectorSchemaRoot = generateVectorSchemaRoot(10, rowCount); BufferAllocator allocator = new RootAllocator(Long.MAX_VALUE); @@ -147,7 +157,8 @@ void setInboundMessageSizeSmall() throws Exception { @Test void setInboundMessageSizeLarge() throws Exception { - URI uri = URI.create("http://127.0.0.1:33333"); + int freePort = findFreePort(); + URI uri = URI.create("http://127.0.0.1:" + freePort); int rowCount = 100; try (VectorSchemaRoot vectorSchemaRoot = generateVectorSchemaRoot(10, rowCount); BufferAllocator allocator = new RootAllocator(Long.MAX_VALUE);