Skip to content

Commit c54da83

Browse files
committed
Track and expose node capabilities per connection
AxonServerConnection.supports(capability) reports whether the node this connection currently reaches has advertised a given capability, so callers can check upfront instead of calling an RPC and reacting to UNIMPLEMENTED. Captured fresh from NodeInfo on every (re)connect in AxonServerManagedChannel, so it reflects the actual connected node even across reconnects to a different one.
1 parent 92bd45e commit c54da83

4 files changed

Lines changed: 39 additions & 1 deletion

File tree

‎pom.xml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -44,7 +44,7 @@
4444
<sonar.organization>axoniq</sonar.organization>
4545
<sonar.host.url>https://sonarcloud.io</sonar.host.url>
4646

47-
<axonserver.api.version>2026.1.0</axonserver.api.version>
47+
<axonserver.api.version>2026.2.0-SNAPSHOT</axonserver.api.version>
4848

4949
<grpc.version>1.79.0</grpc.version>
5050
<protobuf.version>4.36.2</protobuf.version>

‎src/main/java/io/axoniq/axonserver/connector/AxonServerConnection.java‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -55,6 +55,18 @@ public interface AxonServerConnection {
5555
*/
5656
boolean isConnected();
5757

58+
/**
59+
* Indicates whether the Axon Server node this connection currently reaches supports the given {@code
60+
* capability}. The result reflects the node reached by the most recent (re)connect, so it may change over time -
61+
* for example when this connection moves to a different node in the cluster, or that node gets upgraded.
62+
*
63+
* @param capability the capability identifier to check, e.g. {@code "dcb:snapshot-aware-sourcing"}
64+
* @return {@code true} if the currently connected node has advertised the given capability, {@code false} if it
65+
* either does not support it, or has not advertised any capabilities at all (e.g. because it predates capability
66+
* negotiation)
67+
*/
68+
boolean supports(String capability);
69+
5870
/**
5971
* Disconnects all communication channels and terminates any active network connections to AxonServer. No more
6072
* operations can be performed on this connection after invoking <em>{@code disconnect()}</em>.

‎src/main/java/io/axoniq/axonserver/connector/impl/AxonServerManagedChannel.java‎

Lines changed: 21 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -34,6 +34,7 @@
3434
import java.util.Collections;
3535
import java.util.List;
3636
import java.util.Queue;
37+
import java.util.Set;
3738
import java.util.concurrent.LinkedBlockingQueue;
3839
import java.util.concurrent.RejectedExecutionException;
3940
import java.util.concurrent.ScheduledExecutorService;
@@ -63,6 +64,7 @@ public class AxonServerManagedChannel extends ManagedChannel {
6364
private final long connectTimeout;
6465

6566
private final AtomicReference<ManagedChannel> activeChannel = new AtomicReference<>();
67+
private final AtomicReference<Set<String>> capabilities = new AtomicReference<>();
6668
private final AtomicBoolean shutdown = new AtomicBoolean();
6769
private final AtomicBoolean suppressErrors = new AtomicBoolean();
6870
private final Queue<Runnable> connectListeners = new LinkedBlockingQueue<>();
@@ -143,6 +145,9 @@ private ManagedChannel connectChannel() {
143145
new ServerAddress(primaryClusterInfo.getHostName(), primaryClusterInfo.getGrpcPort());
144146
connection = connectionFactory.apply(serverAddress, context);
145147
}
148+
capabilities.set(primaryClusterInfo.hasCapabilities()
149+
? Set.copyOf(primaryClusterInfo.getCapabilities().getCapabilityList())
150+
: null);
146151
suppressErrors.set(false);
147152
lastConnectException.set(null);
148153
break;
@@ -414,6 +419,22 @@ public boolean isReady() {
414419
return getState(false) == ConnectivityState.READY;
415420
}
416421

422+
/**
423+
* Indicates whether the node this channel is currently connected to supports the given {@code capability}.
424+
* <p>
425+
* The result reflects the node reached by the most recent (re)connect, so it may change across reconnects -
426+
* for example after this channel moves to a different node in the cluster, or after that node is upgraded.
427+
*
428+
* @param capability the capability identifier to check, e.g. {@code "dcb:snapshot-aware-sourcing"}
429+
* @return {@code true} if the currently connected node has advertised the given capability, {@code false} if it
430+
* either does not support it, or has not advertised any capabilities at all (e.g. because it predates capability
431+
* negotiation)
432+
*/
433+
public boolean supports(String capability) {
434+
Set<String> currentCapabilities = capabilities.get();
435+
return currentCapabilities != null && currentCapabilities.contains(capability);
436+
}
437+
417438
private static class FailingCall<REQ, RESP> extends ClientCall<REQ, RESP> {
418439

419440
@Override

‎src/main/java/io/axoniq/axonserver/connector/impl/ContextConnection.java‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -143,6 +143,11 @@ public boolean isConnected() {
143143
return connection.getState(false) == ConnectivityState.READY;
144144
}
145145

146+
@Override
147+
public boolean supports(String capability) {
148+
return connection.supports(capability);
149+
}
150+
146151
@Override
147152
public void disconnect() {
148153
doIfNotNull(controlChannel, ControlChannelImpl::disconnect);

0 commit comments

Comments
 (0)