test: add test of FlightStream overhead.

Author: karel-rehor

src/main/java/com/influxdb/v3/client/internal/FlightSqlClient.java

@@ -75,13 +75,13 @@ final class FlightSqlClient implements AutoCloseable {
 
     private static final Logger LOG = LoggerFactory.getLogger(FlightSqlClient.class);
     private static final int AUTOCLOSEABLE_CHECK_LIMIT = 10;
-    private static final Map<AutoCloseable, Boolean> CLOSEABLE_CLOSED_LEDGER = new ConcurrentHashMap<>();
+    static final Map<AutoCloseable, Boolean> CLOSEABLE_CLOSED_LEDGER = new ConcurrentHashMap<>();
 
     private final FlightClient client;
 
     private final Map<String, String> defaultHeaders = new HashMap<>();
     private final ObjectMapper objectMapper = new ObjectMapper();
-    private final List<AutoCloseable> autoCloseables = new ArrayList<>();
+    final List<AutoCloseable> autoCloseables = new ArrayList<>();
 
     FlightSqlClient(@Nonnull final ClientConfig config) {
         this(config, null);
@@ -144,17 +144,17 @@ Stream<VectorSchemaRoot> execute(@Nonnull final String query,
         return StreamSupport.stream(spliterator, false).onClose(iterator::close);
     }
 
-    private void addToAutoCloseable(@Nonnull final AutoCloseable closeable) {
+    private synchronized void addToAutoCloseable(@Nonnull final AutoCloseable closeable) {
         // need to occasionally clean up references to closed streams
         // in order to ensure memory can get freed.
         if (autoCloseables.size() > AUTOCLOSEABLE_CHECK_LIMIT) {
-            LOG.info("checking to cleanup stale flight streams from {} known streams", autoCloseables.size());
+            LOG.debug("checking to cleanup stale flight streams from {} known streams", autoCloseables.size());
 
             ListIterator<AutoCloseable> iter = autoCloseables.listIterator();
             while (iter.hasNext()) {
                 AutoCloseable autoCloseable = iter.next();
                 if (CLOSEABLE_CLOSED_LEDGER.get(autoCloseable)) {
-                    LOG.info("removing closed stream {}", autoCloseable);
+                    LOG.debug("removing closed stream {}", autoCloseable);
                     CLOSEABLE_CLOSED_LEDGER.keySet().remove(autoCloseable);
                     iter.remove();
                 }

src/test/java/com/influxdb/v3/client/integration/E2ETest.java

@@ -470,7 +470,6 @@ public void testMultipleQueries() throws Exception {
             client.writePoints(points);
             String query = "SELECT * FROM " + measurement;
 
-            // TODO just checking FlightStream cleanup should be test in FlightSqlClient
             for (int i = 0; i < 20; i++) {
                 try (Stream<PointValues> stream = client.queryPoints(query)) {
                     stream.forEach(pointValues -> {

src/test/java/com/influxdb/v3/client/internal/FlightSqlClientTest.java

@@ -370,6 +370,36 @@ void createProxyDetector() {
         }
     }
 
+
+    @Test
+    public void multipleFlightStreamsFreed() throws Exception {
+        ClientConfig clientConfig = new ClientConfig.Builder()
+            .host(server.getLocation().getUri().toString())
+            .token("my-token".toCharArray())
+            .build();
+
+        try (FlightSqlClient flightSqlClient = new FlightSqlClient(clientConfig)) {
+
+            for (int i = 0; i < 20; i++) {
+                Stream<VectorSchemaRoot> stream = flightSqlClient.execute(
+                    "select * from cpu",
+                    "mydb",
+                    QueryType.SQL,
+                    Map.of(),
+                    Map.of());
+
+                stream.forEach(VectorSchemaRoot::contentToTSVString);
+            }
+            Assertions.assertThat(flightSqlClient.autoCloseables.size()).isEqualTo(9);
+            // N.B. can pick up references from other tests...
+            Assertions.assertThat(FlightSqlClient.CLOSEABLE_CLOSED_LEDGER.size()).isLessThan(20);
+            for (AutoCloseable closeable : FlightSqlClient.CLOSEABLE_CLOSED_LEDGER.keySet()) {
+                Assertions.assertThat(FlightSqlClient.CLOSEABLE_CLOSED_LEDGER.get(closeable)).isTrue(); // Is closed
+            }
+
+        }
+    }
+
     static class HeaderCaptureMiddleware implements FlightServerMiddleware {
 
         private final Map<String, String> headers = new HashMap<>();
@@ -418,4 +448,5 @@ public HeaderCaptureMiddleware onCallStarted(final CallInfo callInfo,
             return lastInstance;
         }
     }
+
 }