Skip to content

Commit 2d1d997

Browse files
bluestreak01bluestreak01
committed
feat(facade): expose ingest errorHandler/connectionListener on QuestDB
The pooled QuestDB facade built its ingest Senders from config strings only (SenderPool -> Sender.fromConfig), so the programmatic ingest callbacks -- SenderErrorHandler and SenderConnectionListener -- were unreachable: a facade user got the default loud-not-silent handlers with no way to observe async ingest errors or connection transitions. Expose both as QuestDBBuilder setters and thread them to every pooled Sender: - QuestDBBuilder.errorHandler(...) / .connectionListener(...) - QuestDBImpl gains a full constructor carrying the callbacks; the public constructor forwards them and the 12-arg white-box test-seam constructor is preserved as a delegating shim (null callbacks). - SenderPool gains a full constructor + applyUserCallbacks() that applies the callbacks to every sender it builds (both the non-SF and SF paths); the 8-arg test-seam constructor is preserved as a shim. Recovery delegates (internal, short-lived, OFF-mode drain senders) are deliberately excluded so the user's callbacks never see events from internal machinery. Defaults are null -> behaviour is unchanged unless a callback is set. Tests: QuestDBFacadeCallbacksTest prewarms one ingest sender at a dead port in async mode with a tight reconnect budget and asserts the facade-wired errorHandler receives the budget-exhaustion SenderError and the facade-wired connectionListener observes connection events -- no server required.
1 parent 5480282 commit 2d1d997

4 files changed

Lines changed: 244 additions & 6 deletions

File tree

‎core/src/main/java/io/questdb/client/QuestDBBuilder.java‎

Lines changed: 40 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -59,6 +59,10 @@ public final class QuestDBBuilder {
5959
private static final int UNSET = -1;
6060

6161
private long acquireTimeoutMillis = UNSET;
62+
// Optional ingest-side async callbacks. Null -> each pooled Sender uses its
63+
// loud-not-silent default. Applied to every Sender the pool builds.
64+
private SenderConnectionListener connectionListener;
65+
private SenderErrorHandler errorHandler;
6266
private long housekeeperIntervalMillis = UNSET;
6367
private long idleTimeoutMillis = UNSET;
6468
private String ingestConfig;
@@ -85,6 +89,39 @@ public QuestDBBuilder acquireTimeoutMillis(long millis) {
8589
return this;
8690
}
8791

92+
/**
93+
* Sets the async connection-event listener applied to every pooled ingest
94+
* {@link Sender}. The listener observes connect / disconnect / failover
95+
* transitions across the whole sender pool; events are delivered on the
96+
* senders' I/O threads, so the listener must be thread-safe and must not
97+
* block. Pass {@code null} (the default) to keep each sender's
98+
* loud-not-silent default listener.
99+
*
100+
* @param listener the shared connection listener, or {@code null} for the default
101+
* @return this instance for method chaining
102+
*/
103+
public QuestDBBuilder connectionListener(SenderConnectionListener listener) {
104+
this.connectionListener = listener;
105+
return this;
106+
}
107+
108+
/**
109+
* Sets the async error handler applied to every pooled ingest
110+
* {@link Sender}. The handler receives terminal/async ingest errors
111+
* (connect-budget exhaustion, terminal upgrade failures, write errors)
112+
* from across the whole sender pool; notifications are delivered on the
113+
* senders' I/O threads, so the handler must be thread-safe and must not
114+
* block. Pass {@code null} (the default) to keep each sender's
115+
* loud-not-silent default handler.
116+
*
117+
* @param handler the shared error handler, or {@code null} for the default
118+
* @return this instance for method chaining
119+
*/
120+
public QuestDBBuilder errorHandler(SenderErrorHandler handler) {
121+
this.errorHandler = handler;
122+
return this;
123+
}
124+
88125
/**
89126
* Builds the {@link QuestDB} handle. Validates both connect strings up
90127
* front -- so a malformed config fails here even when both pools have
@@ -141,7 +178,9 @@ public QuestDB build() {
141178
acquireTimeoutMillis,
142179
idleTimeoutMillis,
143180
maxLifetimeMillis,
144-
housekeeperIntervalMillis
181+
housekeeperIntervalMillis,
182+
errorHandler,
183+
connectionListener
145184
);
146185
}
147186

‎core/src/main/java/io/questdb/client/impl/QuestDBImpl.java‎

Lines changed: 33 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -28,6 +28,8 @@
2828
import io.questdb.client.QuestDB;
2929
import io.questdb.client.Query;
3030
import io.questdb.client.Sender;
31+
import io.questdb.client.SenderConnectionListener;
32+
import io.questdb.client.SenderErrorHandler;
3133
import io.questdb.client.cutlass.qwp.client.QwpColumnBatchHandler;
3234
import io.questdb.client.cutlass.qwp.client.QwpQueryClient;
3335

@@ -58,11 +60,13 @@ public QuestDBImpl(
5860
long acquireTimeoutMillis,
5961
long idleTimeoutMillis,
6062
long maxLifetimeMillis,
61-
long housekeeperIntervalMillis
63+
long housekeeperIntervalMillis,
64+
SenderErrorHandler errorHandler,
65+
SenderConnectionListener connectionListener
6266
) {
6367
this(ingestConfig, queryConfig, senderMin, senderMax, queryMin, queryMax,
6468
acquireTimeoutMillis, idleTimeoutMillis, maxLifetimeMillis,
65-
housekeeperIntervalMillis, null, null);
69+
housekeeperIntervalMillis, null, null, errorHandler, connectionListener);
6670
}
6771

6872
// Package-private constructor exposing the senderFactory and connectHook test
@@ -84,6 +88,31 @@ public QuestDBImpl(
8488
long housekeeperIntervalMillis,
8589
IntFunction<Sender> senderFactory,
8690
Consumer<QwpQueryClient> connectHook
91+
) {
92+
this(ingestConfig, queryConfig, senderMin, senderMax, queryMin, queryMax,
93+
acquireTimeoutMillis, idleTimeoutMillis, maxLifetimeMillis,
94+
housekeeperIntervalMillis, senderFactory, connectHook, null, null);
95+
}
96+
97+
// Full constructor adding the ingest-side errorHandler/connectionListener,
98+
// applied by SenderPool to every Sender it builds. The 12-arg overload above
99+
// is the unchanged white-box test seam and delegates here with null
100+
// callbacks; the public overload delegates here with null test seams.
101+
QuestDBImpl(
102+
String ingestConfig,
103+
String queryConfig,
104+
int senderMin,
105+
int senderMax,
106+
int queryMin,
107+
int queryMax,
108+
long acquireTimeoutMillis,
109+
long idleTimeoutMillis,
110+
long maxLifetimeMillis,
111+
long housekeeperIntervalMillis,
112+
IntFunction<Sender> senderFactory,
113+
Consumer<QwpQueryClient> connectHook,
114+
SenderErrorHandler errorHandler,
115+
SenderConnectionListener connectionListener
87116
) {
88117
SenderPool builtSenderPool = null;
89118
QueryClientPool builtQueryPool = null;
@@ -95,7 +124,8 @@ public QuestDBImpl(
95124
// Defer SF startup recovery to the PoolHousekeeper thread so
96125
// build() never blocks on a slow / reachable-but-not-acking
97126
// server; the housekeeper drives it via runStartupRecoveryStep().
98-
true);
127+
true,
128+
errorHandler, connectionListener);
99129
builtQueryPool = new QueryClientPool(
100130
queryConfig, queryMin, queryMax, acquireTimeoutMillis,
101131
idleTimeoutMillis, maxLifetimeMillis, connectHook);

‎core/src/main/java/io/questdb/client/impl/SenderPool.java‎

Lines changed: 45 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,8 @@
2525
package io.questdb.client.impl;
2626

2727
import io.questdb.client.Sender;
28+
import io.questdb.client.SenderConnectionListener;
29+
import io.questdb.client.SenderErrorHandler;
2830
import io.questdb.client.cutlass.line.LineSenderException;
2931
import io.questdb.client.cutlass.qwp.client.QwpWebSocketSender;
3032
import io.questdb.client.cutlass.qwp.client.sf.cursor.OrphanScanner;
@@ -96,6 +98,10 @@ public final class SenderPool implements AutoCloseable {
9698
private final ArrayList<PooledSender> all;
9799
private final ArrayDeque<PooledSender> available;
98100
private final String configurationString;
101+
// User-supplied ingest callbacks, shared across every pooled Sender this
102+
// pool builds. Null -> each sender keeps its loud-not-silent default.
103+
private final SenderConnectionListener connectionListener;
104+
private final SenderErrorHandler errorHandler;
99105
private final long idleTimeoutMillis;
100106
// Test seam. Production builds delegates via defaultSender(); white-box
101107
// tests in io.questdb.client.test.impl reach the package-private
@@ -227,10 +233,33 @@ public SenderPool(
227233
long maxLifetimeMillis,
228234
IntFunction<Sender> senderFactory,
229235
boolean deferStartupRecovery
236+
) {
237+
this(configurationString, minSize, maxSize, acquireTimeoutMillis,
238+
idleTimeoutMillis, maxLifetimeMillis, senderFactory,
239+
deferStartupRecovery, null, null);
240+
}
241+
242+
// Full constructor adding the user-supplied ingest callbacks (error handler
243+
// and connection listener), applied to every Sender the pool builds (see
244+
// buildManagedSlotSender). The 8-arg overload above is the unchanged
245+
// white-box test seam and delegates here with null callbacks.
246+
SenderPool(
247+
String configurationString,
248+
int minSize,
249+
int maxSize,
250+
long acquireTimeoutMillis,
251+
long idleTimeoutMillis,
252+
long maxLifetimeMillis,
253+
IntFunction<Sender> senderFactory,
254+
boolean deferStartupRecovery,
255+
SenderErrorHandler errorHandler,
256+
SenderConnectionListener connectionListener
230257
) {
231258
if (minSize < 0 || maxSize < 1 || minSize > maxSize) {
232259
throw new IllegalArgumentException("invalid pool sizing: min=" + minSize + ", max=" + maxSize);
233260
}
261+
this.errorHandler = errorHandler;
262+
this.connectionListener = connectionListener;
234263
this.senderFactory = senderFactory != null ? senderFactory : this::defaultSender;
235264
// An injected factory (tests) drives recovery too, preserving the
236265
// white-box recovery seam; production recovery forces OFF-mode connects
@@ -1035,9 +1064,21 @@ private Sender defaultRecoverySender(int slotIndex) {
10351064
return buildManagedSlotSender(slotIndex, true);
10361065
}
10371066

1067+
// Applies the user-supplied ingest callbacks to a sender builder. Null
1068+
// callbacks are skipped so the sender keeps its loud-not-silent default.
1069+
private Sender.LineSenderBuilder applyUserCallbacks(Sender.LineSenderBuilder builder) {
1070+
if (errorHandler != null) {
1071+
builder.errorHandler(errorHandler);
1072+
}
1073+
if (connectionListener != null) {
1074+
builder.connectionListener(connectionListener);
1075+
}
1076+
return builder;
1077+
}
1078+
10381079
private Sender buildManagedSlotSender(int slotIndex, boolean forRecovery) {
10391080
if (!storeAndForward) {
1040-
return Sender.fromConfig(configurationString);
1081+
return applyUserCallbacks(Sender.builder(configurationString)).build();
10411082
}
10421083
// Give this pooled sender its own slot dir <sf_dir>/<base>-<index>
10431084
// so concurrent SF senders sharing one sf_dir never collide on
@@ -1091,7 +1132,9 @@ private Sender buildManagedSlotSender(int slotIndex, boolean forRecovery) {
10911132
// returns).
10921133
builder.drainOrphans(false);
10931134
}
1094-
return builder.build();
1135+
// Recovery delegates are internal, short-lived, OFF-mode drain senders;
1136+
// don't surface their connect/error events to the user's callbacks.
1137+
return (forRecovery ? builder : applyUserCallbacks(builder)).build();
10951138
}
10961139

10971140
/**

‎core/src/test/java/io/questdb/client/test/QuestDBFacadeCallbacksTest.java‎

Lines changed: 126 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,126 @@
1+
/*+*****************************************************************************
2+
* ___ _ ____ ____
3+
* / _ \ _ _ ___ ___| |_| _ \| __ )
4+
* | | | | | | |/ _ \/ __| __| | | | _ \
5+
* | |_| | |_| | __/\__ \ |_| |_| | |_) |
6+
* \__\_\\__,_|\___||___/\__|____/|____/
7+
*
8+
* Copyright (c) 2014-2019 Appsicle
9+
* Copyright (c) 2019-2026 QuestDB
10+
*
11+
* Licensed under the Apache License, Version 2.0 (the "License");
12+
* you may not use this file except in compliance with the License.
13+
* You may obtain a copy of the License at
14+
*
15+
* http://www.apache.org/licenses/LICENSE-2.0
16+
*
17+
* Unless required by applicable law or agreed to in writing, software
18+
* distributed under the License is distributed on an "AS IS" BASIS,
19+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
20+
* See the License for the specific language governing permissions and
21+
* limitations under the License.
22+
*
23+
******************************************************************************/
24+
25+
package io.questdb.client.test;
26+
27+
import io.questdb.client.QuestDB;
28+
import io.questdb.client.SenderConnectionEvent;
29+
import io.questdb.client.SenderConnectionListener;
30+
import io.questdb.client.SenderError;
31+
import io.questdb.client.SenderErrorHandler;
32+
import io.questdb.client.test.cutlass.qwp.client.TestPorts;
33+
import org.jetbrains.annotations.NotNull;
34+
import org.junit.Assert;
35+
import org.junit.Test;
36+
37+
import java.util.concurrent.CountDownLatch;
38+
import java.util.concurrent.TimeUnit;
39+
import java.util.concurrent.atomic.AtomicReference;
40+
41+
/**
42+
* Proves the ingest-side async callbacks exposed on the {@link QuestDB} facade
43+
* ({@link io.questdb.client.QuestDBBuilder#errorHandler}/{@code connectionListener})
44+
* actually reach the pooled {@link io.questdb.client.Sender}s -- not merely the
45+
* lower-level {@code Sender.builder()}.
46+
* <p>
47+
* Each test eagerly prewarms one ingest sender ({@code sender_pool_min=1})
48+
* pointed at a dead port in {@code initial_connect_retry=async} mode with a
49+
* tight reconnect budget: the pool's I/O thread exhausts the budget in the
50+
* background and surfaces the failure through whichever facade-wired callback is
51+
* under test. No server is required.
52+
*/
53+
public class QuestDBFacadeCallbacksTest {
54+
55+
@Test
56+
public void testFacadeConnectionListenerReceivesEvents() throws Exception {
57+
int port = TestPorts.findUnusedPort();
58+
CountDownLatch sawEvent = new CountDownLatch(1);
59+
SenderConnectionListener listener = new SenderConnectionListener() {
60+
@Override
61+
public void onEvent(@NotNull SenderConnectionEvent event) {
62+
sawEvent.countDown();
63+
}
64+
};
65+
try (QuestDB ignored = QuestDB.builder()
66+
.ingestConfig(ingestConfig(port))
67+
.queryConfig(queryConfig(port))
68+
.connectionListener(listener)
69+
.build()) {
70+
Assert.assertTrue(
71+
"facade-wired connectionListener must observe at least one connection event",
72+
sawEvent.await(5, TimeUnit.SECONDS));
73+
}
74+
}
75+
76+
@Test
77+
public void testFacadeErrorHandlerReceivesAsyncIngestError() throws Exception {
78+
int port = TestPorts.findUnusedPort();
79+
ErrorInbox inbox = new ErrorInbox();
80+
try (QuestDB ignored = QuestDB.builder()
81+
.ingestConfig(ingestConfig(port))
82+
.queryConfig(queryConfig(port))
83+
.errorHandler(inbox)
84+
.build()) {
85+
Assert.assertTrue(
86+
"facade-wired errorHandler must receive the async budget-exhaustion SenderError",
87+
inbox.await(5, TimeUnit.SECONDS));
88+
Assert.assertNotNull("a SenderError must be delivered", inbox.get());
89+
}
90+
}
91+
92+
// Eagerly prewarm one sender (sender_pool_min=1) so build() exercises the
93+
// production buildManagedSlotSender path that applies the facade callbacks.
94+
// async + tight budget -> the I/O thread fails fast against the dead port.
95+
private static String ingestConfig(int port) {
96+
return "ws::addr=localhost:" + port + ";sender_pool_min=1;sender_pool_max=1"
97+
+ ";initial_connect_retry=async;reconnect_max_duration_millis=400"
98+
+ ";reconnect_initial_backoff_millis=10;reconnect_max_backoff_millis=50"
99+
+ ";close_flush_timeout_millis=0;";
100+
}
101+
102+
// query_pool_min=0 -> the query pool never connects, so the test is isolated
103+
// to the ingest callbacks.
104+
private static String queryConfig(int port) {
105+
return "ws::addr=localhost:" + port + ";query_pool_min=0;query_pool_max=1;";
106+
}
107+
108+
private static final class ErrorInbox implements SenderErrorHandler {
109+
private final CountDownLatch latch = new CountDownLatch(1);
110+
private final AtomicReference<SenderError> first = new AtomicReference<>();
111+
112+
boolean await(long timeout, TimeUnit unit) throws InterruptedException {
113+
return latch.await(timeout, unit);
114+
}
115+
116+
SenderError get() {
117+
return first.get();
118+
}
119+
120+
@Override
121+
public void onError(@NotNull SenderError error) {
122+
first.compareAndSet(null, error);
123+
latch.countDown();
124+
}
125+
}
126+
}

0 commit comments

Comments
 (0)