This is an automated email from the ASF dual-hosted git repository.
mattrpav pushed a commit to branch activemq-6.2.x
in repository https://gitbox.apache.org/repos/asf/activemq.git
The following commit(s) were added to refs/heads/activemq-6.2.x by this push:
new 05242aab21 [AMQ-9687] Add metrics about error and reconnect counts to
network connector and network bridges (#2239)
05242aab21 is described below
commit 05242aab216ea2e91dd8190d38e588cbdec8112d
Author: Matt Pavlovich <[email protected]>
AuthorDate: Thu Jul 23 15:01:12 2026 -0500
[AMQ-9687] Add metrics about error and reconnect counts to network
connector and network bridges (#2239)
- NetworkConnector statistics now parent the bridge statistics
- Use a CAS AtomicBoolean to ensure only one of local exception or remote
exception is counted in a bridge
- Bridge statistics guard against shutdown leading to miscounts
- More tests to affirm local and remote counted separately
(cherry picked from commit cea5c0917beea41b934d9cbd5df773097a80fda3)
---
.../activemq/broker/jmx/NetworkBridgeView.java | 15 ++++
.../broker/jmx/NetworkBridgeViewMBean.java | 5 ++
.../activemq/broker/jmx/NetworkConnectorView.java | 25 ++++++
.../broker/jmx/NetworkConnectorViewMBean.java | 10 +++
.../network/DemandForwardingBridgeSupport.java | 50 +++++++++++-
.../network/DiscoveryNetworkConnector.java | 3 +
.../org/apache/activemq/network/NetworkBridge.java | 6 ++
.../activemq/network/NetworkBridgeStatistics.java | 38 ++++++++-
.../apache/activemq/network/NetworkConnector.java | 55 +++++++++++++
.../java/org/apache/activemq/bugs/AMQ4160Test.java | 15 ++++
.../network/NetworkAdvancedStatisticsTest.java | 90 +++++++++++++++++++++-
.../localBroker-advancedNetworkStatistics.xml | 31 +++++++-
.../remoteBroker-advancedNetworkStatistics.xml | 27 ++++++-
13 files changed, 358 insertions(+), 12 deletions(-)
diff --git
a/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/NetworkBridgeView.java
b/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/NetworkBridgeView.java
index 756ffbed7d..6b2798cf6c 100644
---
a/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/NetworkBridgeView.java
+++
b/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/NetworkBridgeView.java
@@ -99,6 +99,21 @@ public class NetworkBridgeView implements
NetworkBridgeViewMBean {
}
}
+ @Override
+ public long getStartedTimestamp() {
+ return bridge.getStartedTimestamp();
+ }
+
+ @Override
+ public long getLocalExceptionCount() {
+ return bridge.getLocalExceptionCount();
+ }
+
+ @Override
+ public long getRemoteExceptionCount() {
+ return bridge.getRemoteExceptionCount();
+ }
+
public void addNetworkDestinationView(NetworkDestinationView
networkDestinationView){
networkDestinationViewList.add(networkDestinationView);
}
diff --git
a/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/NetworkBridgeViewMBean.java
b/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/NetworkBridgeViewMBean.java
index 82fc9ca189..34d7d4b198 100644
---
a/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/NetworkBridgeViewMBean.java
+++
b/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/NetworkBridgeViewMBean.java
@@ -40,4 +40,9 @@ public interface NetworkBridgeViewMBean extends Service {
void resetStats();
+ long getStartedTimestamp();
+
+ long getLocalExceptionCount();
+
+ long getRemoteExceptionCount();
}
diff --git
a/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/NetworkConnectorView.java
b/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/NetworkConnectorView.java
index 05e6747dd7..09e7da0755 100644
---
a/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/NetworkConnectorView.java
+++
b/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/NetworkConnectorView.java
@@ -203,4 +203,29 @@ public class NetworkConnectorView implements
NetworkConnectorViewMBean {
public boolean isAutoStart() {
return connector.isAutoStart();
}
+
+ @Override
+ public long getStartedTimestamp() {
+ return connector.getStartedTimestamp();
+ }
+
+ @Override
+ public long getStoppedTimestamp() {
+ return connector.getStoppedTimestamp();
+ }
+
+ @Override
+ public long getBridgeExceptionCount() {
+ return connector.getBridgeExceptionCounter().getCount();
+ }
+
+ @Override
+ public long getLocalExceptionCount() {
+ return connector.getLocalExceptionCounter().getCount();
+ }
+
+ @Override
+ public long getRemoteExceptionCount() {
+ return connector.getRemoteExceptionCounter().getCount();
+ }
}
diff --git
a/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/NetworkConnectorViewMBean.java
b/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/NetworkConnectorViewMBean.java
index e51b9a16ca..47a8da6c07 100644
---
a/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/NetworkConnectorViewMBean.java
+++
b/activemq-broker/src/main/java/org/apache/activemq/broker/jmx/NetworkConnectorViewMBean.java
@@ -90,4 +90,14 @@ public interface NetworkConnectorViewMBean extends Service {
void setRemotePassword(String remotePassword);
boolean isAutoStart();
+
+ long getStartedTimestamp();
+
+ long getStoppedTimestamp();
+
+ long getBridgeExceptionCount();
+
+ long getLocalExceptionCount();
+
+ long getRemoteExceptionCount();
}
diff --git
a/activemq-broker/src/main/java/org/apache/activemq/network/DemandForwardingBridgeSupport.java
b/activemq-broker/src/main/java/org/apache/activemq/network/DemandForwardingBridgeSupport.java
index 762c6eeffd..c8bb1386c5 100644
---
a/activemq-broker/src/main/java/org/apache/activemq/network/DemandForwardingBridgeSupport.java
+++
b/activemq-broker/src/main/java/org/apache/activemq/network/DemandForwardingBridgeSupport.java
@@ -37,6 +37,7 @@ import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
import java.util.regex.Pattern;
import javax.management.ObjectName;
@@ -129,6 +130,8 @@ public abstract class DemandForwardingBridgeSupport
implements NetworkBridge, Br
protected final AtomicBoolean localBridgeStarted = new
AtomicBoolean(false);
protected final AtomicBoolean remoteBridgeStarted = new
AtomicBoolean(false);
protected final AtomicBoolean bridgeFailed = new AtomicBoolean();
+ // Ensures a single bridge failure counts exactly one local/remote
exception
+ protected final AtomicBoolean bridgeExceptionCounted = new AtomicBoolean();
protected final AtomicBoolean disposed = new AtomicBoolean();
protected BrokerId localBrokerId;
protected ActiveMQDestination[] excludedDestinations;
@@ -171,6 +174,7 @@ public abstract class DemandForwardingBridgeSupport
implements NetworkBridge, Br
private final ExecutorService syncExecutor =
Executors.newSingleThreadExecutor();
private Transport duplexInboundLocalBroker = null;
private ProducerInfo duplexInboundLocalProducerInfo;
+ private final AtomicLong startedTimestamp = new AtomicLong(0L);
public DemandForwardingBridgeSupport(NetworkBridgeConfiguration
configuration, Transport localBroker, Transport remoteBroker) {
this.configuration = configuration;
@@ -194,7 +198,15 @@ public abstract class DemandForwardingBridgeSupport
implements NetworkBridge, Br
throw new IllegalArgumentException("BrokerService is null on "
+ this);
}
-
networkBridgeStatistics.setEnabled(brokerService.isEnableStatistics());
+ if (brokerService.isEnableStatistics()) {
+ networkBridgeStatistics.setEnabled(true);
+ // Bridges are ephemeral - parent this bridge's statistics to
the owning
+ // connector so its counts (including local/remote exceptions)
roll up
+ // automatically and survive the bridge being torn down and
recreated.
+ if (configuration instanceof NetworkConnector) {
+ networkBridgeStatistics.setParent(((NetworkConnector)
configuration).getNetworkBridgeStatistics());
+ }
+ }
if (isDuplex()) {
duplexInboundLocalBroker =
NetworkBridgeFactory.createLocalAsyncTransport(brokerService.getBroker().getVmConnectorURI());
@@ -260,6 +272,8 @@ public abstract class DemandForwardingBridgeSupport
implements NetworkBridge, Br
triggerStartAsyncNetworkBridgeCreation();
} catch (IOException e) {
LOG.warn("Caught exception from remote start", e);
+ } finally {
+ startedTimestamp.set(System.currentTimeMillis());
}
} else {
LOG.warn("Bridge was disposed before the start() method was
fully executed.");
@@ -336,6 +350,7 @@ public abstract class DemandForwardingBridgeSupport
implements NetworkBridge, Br
startedLatch.countDown();
localStartedLatch.countDown();
staticDestinationsLatch.countDown();
+ startedTimestamp.set(0L);
ss.throwFirstException();
}
@@ -650,12 +665,21 @@ public abstract class DemandForwardingBridgeSupport
implements NetworkBridge, Br
@Override
public void serviceRemoteException(Throwable error) {
if (!disposed.get()) {
+ // Count the single genuine failure that trips disposal, not the
peer exceptions
+ // fired during teardown nor a second callback racing in before
disposal completes.
+ // This rolls up to the owning connector automatically via the
parent statistics
+ // wired in start().
+ if (bridgeExceptionCounted.compareAndSet(false, true)) {
+ networkBridgeStatistics.getRemoteExceptionCount().increment();
+ }
+
if (error instanceof SecurityException || error instanceof
GeneralSecurityException) {
LOG.error("Network connection between {} and {} shutdown due
to a remote error: {}", localBroker, remoteBroker, error.toString());
} else {
LOG.warn("Network connection between {} and {} shutdown due to
a remote error: {}", localBroker, remoteBroker, error.toString());
}
LOG.debug("The remote Exception was: {}", error, error);
+
brokerService.getTaskRunnerFactory().execute(new Runnable() {
@Override
public void run() {
@@ -1114,6 +1138,7 @@ public abstract class DemandForwardingBridgeSupport
implements NetworkBridge, Br
public void serviceLocalException(MessageDispatch messageDispatch,
Throwable error) {
LOG.trace("serviceLocalException: disposed {} ex", disposed.get(),
error);
+
if (!disposed.get()) {
if (error instanceof DestinationDoesNotExistException &&
((DestinationDoesNotExistException) error).isTemporary()) {
// not a reason to terminate the bridge - temps can disappear
with
@@ -1134,6 +1159,14 @@ public abstract class DemandForwardingBridgeSupport
implements NetworkBridge, Br
return;
}
+ // Count the single genuine failure that trips disposal, not the
peer exceptions
+ // fired during teardown, a second callback racing in before
disposal completes,
+ // nor the ignorable temporary-destination errors handled above.
This rolls up to
+ // the owning connector automatically via the parent statistics
wired in start().
+ if (bridgeExceptionCounted.compareAndSet(false, true)) {
+ networkBridgeStatistics.getLocalExceptionCount().increment();
+ }
+
LOG.info("Network connection between {} and {} shutdown due to a
local error: {}", localBroker, remoteBroker, error);
LOG.debug("The local Exception was: {}", error, error);
@@ -1920,6 +1953,21 @@ public abstract class DemandForwardingBridgeSupport
implements NetworkBridge, Br
return networkBridgeStatistics.getEnqueues().getCount();
}
+ @Override
+ public long getStartedTimestamp() {
+ return startedTimestamp.get();
+ }
+
+ @Override
+ public long getLocalExceptionCount() {
+ return networkBridgeStatistics.getLocalExceptionCount().getCount();
+ }
+
+ @Override
+ public long getRemoteExceptionCount() {
+ return networkBridgeStatistics.getRemoteExceptionCount().getCount();
+ }
+
@Override
public NetworkBridgeStatistics getNetworkBridgeStatistics() {
return networkBridgeStatistics;
diff --git
a/activemq-broker/src/main/java/org/apache/activemq/network/DiscoveryNetworkConnector.java
b/activemq-broker/src/main/java/org/apache/activemq/network/DiscoveryNetworkConnector.java
index 3b9696cf92..5532f3f98a 100644
---
a/activemq-broker/src/main/java/org/apache/activemq/network/DiscoveryNetworkConnector.java
+++
b/activemq-broker/src/main/java/org/apache/activemq/network/DiscoveryNetworkConnector.java
@@ -131,6 +131,7 @@ public class DiscoveryNetworkConnector extends
NetworkConnector implements Disco
try {
remoteTransport = TransportFactory.connect(connectUri);
} catch (Exception e) {
+
networkBridgeStatistics.getRemoteExceptionCount().increment();
LOG.warn("Could not connect to remote URI: {}: {}",
connectUri, e.getMessage());
LOG.debug("Connection failure exception: ", e);
try {
@@ -143,6 +144,7 @@ public class DiscoveryNetworkConnector extends
NetworkConnector implements Disco
try {
localTransport = createLocalTransport();
} catch (Exception e) {
+
networkBridgeStatistics.getLocalExceptionCount().increment();
ServiceSupport.dispose(remoteTransport);
LOG.warn("Could not connect to local URI: {}: {}",
localURI, e.getMessage());
LOG.debug("Connection failure exception: ", e);
@@ -164,6 +166,7 @@ public class DiscoveryNetworkConnector extends
NetworkConnector implements Disco
}
bridge.start();
} catch (Exception e) {
+ bridgeExceptionCounter.increment();
ServiceSupport.dispose(localTransport);
ServiceSupport.dispose(remoteTransport);
LOG.warn("Could not start network bridge between: {} and: {}
due to: {}", localURI, uri, e.getMessage());
diff --git
a/activemq-broker/src/main/java/org/apache/activemq/network/NetworkBridge.java
b/activemq-broker/src/main/java/org/apache/activemq/network/NetworkBridge.java
index fb9e3d9f08..42d0cd38f8 100644
---
a/activemq-broker/src/main/java/org/apache/activemq/network/NetworkBridge.java
+++
b/activemq-broker/src/main/java/org/apache/activemq/network/NetworkBridge.java
@@ -95,4 +95,10 @@ public interface NetworkBridge extends Service {
ObjectName getMbeanObjectName();
void resetStats();
+
+ long getStartedTimestamp();
+
+ long getLocalExceptionCount();
+
+ long getRemoteExceptionCount();
}
diff --git
a/activemq-broker/src/main/java/org/apache/activemq/network/NetworkBridgeStatistics.java
b/activemq-broker/src/main/java/org/apache/activemq/network/NetworkBridgeStatistics.java
index 50c3951279..8b52db8194 100644
---
a/activemq-broker/src/main/java/org/apache/activemq/network/NetworkBridgeStatistics.java
+++
b/activemq-broker/src/main/java/org/apache/activemq/network/NetworkBridgeStatistics.java
@@ -17,6 +17,8 @@
package org.apache.activemq.network;
+import java.util.Set;
+
import org.apache.activemq.management.CountStatisticImpl;
import org.apache.activemq.management.StatsImpl;
@@ -28,15 +30,17 @@ public class NetworkBridgeStatistics extends StatsImpl {
protected CountStatisticImpl enqueues;
protected CountStatisticImpl dequeues;
protected CountStatisticImpl receivedCount;
+ protected CountStatisticImpl localExceptionCount;
+ protected CountStatisticImpl remoteExceptionCount;
public NetworkBridgeStatistics() {
enqueues = new CountStatisticImpl("enqueues", "The current number of
enqueues this bridge has, which is the number of potential messages to be
forwarded.");
dequeues = new CountStatisticImpl("dequeues", "The current number of
dequeues this bridge has, which is the number of messages received by the
remote broker.");
receivedCount = new CountStatisticImpl("receivedCount", "The number of
messages that have been received by the NetworkBridge from the remote broker.
Only applies for Duplex bridges.");
+ localExceptionCount = new CountStatisticImpl("localExceptionCount",
"The number of exceptions that have been received by the NetworkBridge from the
local broker.");
+ remoteExceptionCount = new CountStatisticImpl("remoteExceptionCount",
"The number of exceptions that have been received by the NetworkBridge from the
remote broker.");
- addStatistic("enqueues", enqueues);
- addStatistic("dequeues", dequeues);
- addStatistic("receivedCount", receivedCount);
+ addStatistics(Set.of(enqueues, dequeues, receivedCount,
localExceptionCount, remoteExceptionCount));
}
/**
@@ -69,6 +73,26 @@ public class NetworkBridgeStatistics extends StatsImpl {
return receivedCount;
}
+ /**
+ * The current number of exceptions this bridge has, which is the number of
+ * exceptions received from the remote broker.
+ *
+ * @return
+ */
+ public CountStatisticImpl getLocalExceptionCount() {
+ return localExceptionCount;
+ }
+
+ /**
+ * The current number of exceptions this bridge has, which is the number of
+ * exceptions received from the remote broker.
+ *
+ * @return
+ */
+ public CountStatisticImpl getRemoteExceptionCount() {
+ return remoteExceptionCount;
+ }
+
@Override
public void reset() {
if (this.isDoReset()) {
@@ -76,6 +100,8 @@ public class NetworkBridgeStatistics extends StatsImpl {
enqueues.reset();
dequeues.reset();
receivedCount.reset();
+ localExceptionCount.reset();
+ remoteExceptionCount.reset();
}
}
@@ -85,6 +111,8 @@ public class NetworkBridgeStatistics extends StatsImpl {
enqueues.setEnabled(enabled);
dequeues.setEnabled(enabled);
receivedCount.setEnabled(enabled);
+ localExceptionCount.setEnabled(enabled);
+ remoteExceptionCount.setEnabled(enabled);
}
public void setParent(NetworkBridgeStatistics parent) {
@@ -92,10 +120,14 @@ public class NetworkBridgeStatistics extends StatsImpl {
enqueues.setParent(parent.enqueues);
dequeues.setParent(parent.dequeues);
receivedCount.setParent(parent.receivedCount);
+ localExceptionCount.setParent(parent.localExceptionCount);
+ remoteExceptionCount.setParent(parent.remoteExceptionCount);
} else {
enqueues.setParent(null);
dequeues.setParent(null);
receivedCount.setParent(null);
+ localExceptionCount.setParent(null);
+ remoteExceptionCount.setParent(null);
}
}
diff --git
a/activemq-broker/src/main/java/org/apache/activemq/network/NetworkConnector.java
b/activemq-broker/src/main/java/org/apache/activemq/network/NetworkConnector.java
index 1bbc483d5c..b08e03bab1 100644
---
a/activemq-broker/src/main/java/org/apache/activemq/network/NetworkConnector.java
+++
b/activemq-broker/src/main/java/org/apache/activemq/network/NetworkConnector.java
@@ -24,6 +24,7 @@ import java.util.List;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.atomic.AtomicLong;
import javax.management.MalformedObjectNameException;
import javax.management.ObjectName;
@@ -36,6 +37,8 @@ import org.apache.activemq.broker.jmx.NetworkBridgeView;
import org.apache.activemq.broker.jmx.NetworkBridgeViewMBean;
import org.apache.activemq.command.ActiveMQDestination;
import org.apache.activemq.command.ConsumerId;
+import org.apache.activemq.management.CountStatistic;
+import org.apache.activemq.management.CountStatisticImpl;
import org.apache.activemq.transport.Transport;
import org.apache.activemq.util.ServiceStopper;
import org.apache.activemq.util.ServiceSupport;
@@ -51,6 +54,13 @@ public abstract class NetworkConnector extends
NetworkBridgeConfiguration implem
protected URI localURI;
protected ConnectionFilter connectionFilter;
protected ConcurrentMap<URI, NetworkBridge> bridges = new
ConcurrentHashMap<URI, NetworkBridge>();
+ protected final AtomicLong startedTimestamp = new AtomicLong(0L);
+ protected final AtomicLong stoppedTimestamp = new AtomicLong(0L);
+ protected final CountStatisticImpl bridgeExceptionCounter = new
CountStatisticImpl("bridgeExceptionCount", "Count of exceptions when
establishing network bridge.");
+ // Aggregates the local/remote exception counts (and message flow) of
every bridge this
+ // connector creates. Each bridge sets this as the parent of its own
statistics, so the
+ // counts roll up automatically and survive the ephemeral bridges being
recreated.
+ protected final NetworkBridgeStatistics networkBridgeStatistics = new
NetworkBridgeStatistics();
protected ServiceSupport serviceSupport = new ServiceSupport() {
@@ -162,11 +172,15 @@ public abstract class NetworkConnector extends
NetworkBridgeConfiguration implem
@Override
public void start() throws Exception {
serviceSupport.start();
+ startedTimestamp.set(System.currentTimeMillis());
+ stoppedTimestamp.set(0L);
}
@Override
public void stop() throws Exception {
serviceSupport.stop();
+ stoppedTimestamp.set(System.currentTimeMillis());
+ startedTimestamp.set(0L);
}
protected void handleStart() throws Exception {
@@ -174,10 +188,22 @@ public abstract class NetworkConnector extends
NetworkBridgeConfiguration implem
throw new IllegalStateException("You must configure the 'localURI'
property");
}
LOG.info("Network Connector {} started", this);
+
+ if (brokerService != null && brokerService.isEnableStatistics()) {
+ bridgeExceptionCounter.setEnabled(true);
+ bridgeExceptionCounter.setCount(0L);
+ networkBridgeStatistics.setEnabled(true);
+ networkBridgeStatistics.reset();
+ }
}
protected void handleStop(ServiceStopper stopper) throws Exception {
LOG.info("Network Connector {} stopped", this);
+
+ if (brokerService != null && brokerService.isEnableStatistics()) {
+ bridgeExceptionCounter.reset();
+ networkBridgeStatistics.reset();
+ }
}
public boolean isStarted() {
@@ -255,4 +281,33 @@ public abstract class NetworkConnector extends
NetworkBridgeConfiguration implem
public Collection<NetworkBridge> activeBridges() {
return bridges.values();
}
+
+ public long getStartedTimestamp() {
+ return startedTimestamp.get();
+ }
+
+ public long getStoppedTimestamp() {
+ return stoppedTimestamp.get();
+ }
+
+ public CountStatistic getBridgeExceptionCounter() {
+ return bridgeExceptionCounter;
+ }
+
+ public CountStatistic getLocalExceptionCounter() {
+ return networkBridgeStatistics.getLocalExceptionCount();
+ }
+
+ public CountStatistic getRemoteExceptionCounter() {
+ return networkBridgeStatistics.getRemoteExceptionCount();
+ }
+
+ /**
+ * @return the connector-level statistics that aggregate every bridge's
statistics.
+ * A bridge sets this as the parent of its own {@link
NetworkBridgeStatistics} so the
+ * counts roll up automatically.
+ */
+ public NetworkBridgeStatistics getNetworkBridgeStatistics() {
+ return networkBridgeStatistics;
+ }
}
diff --git
a/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ4160Test.java
b/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ4160Test.java
index 5563ded869..7ecfc8564b 100644
---
a/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ4160Test.java
+++
b/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ4160Test.java
@@ -329,6 +329,21 @@ public class AMQ4160Test extends
JmsMultipleBrokersTestSupport {
return next.getDequeueCounter();
}
+ @Override
+ public long getLocalExceptionCount() {
+ return next.getLocalExceptionCount();
+ }
+
+ @Override
+ public long getRemoteExceptionCount() {
+ return next.getRemoteExceptionCount();
+ }
+
+ @Override
+ public long getStartedTimestamp() {
+ return next.getStartedTimestamp();
+ }
+
@Override
public NetworkBridgeStatistics
getNetworkBridgeStatistics() {
return next.getNetworkBridgeStatistics();
diff --git
a/activemq-unit-tests/src/test/java/org/apache/activemq/network/NetworkAdvancedStatisticsTest.java
b/activemq-unit-tests/src/test/java/org/apache/activemq/network/NetworkAdvancedStatisticsTest.java
index e18a66c3b0..8df2f0ee10 100644
---
a/activemq-unit-tests/src/test/java/org/apache/activemq/network/NetworkAdvancedStatisticsTest.java
+++
b/activemq-unit-tests/src/test/java/org/apache/activemq/network/NetworkAdvancedStatisticsTest.java
@@ -17,6 +17,7 @@
package org.apache.activemq.network;
import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
@@ -40,6 +41,7 @@ import org.apache.activemq.command.ActiveMQDestination;
import org.apache.activemq.command.ActiveMQQueue;
import org.apache.activemq.command.ActiveMQTopic;
import org.apache.activemq.management.MessageFlowStats;
+import org.apache.activemq.transport.discovery.simple.SimpleDiscoveryAgent;
import org.apache.activemq.util.Wait;
import org.apache.activemq.util.Wait.Condition;
import org.junit.Test;
@@ -87,12 +89,12 @@ public class NetworkAdvancedStatisticsTest extends
BaseNetworkTest {
ActiveMQConnectionFactory fac = new
ActiveMQConnectionFactory(localURI);
fac.setAlwaysSyncSend(true);
fac.setDispatchAsync(false);
- localConnection = fac.createConnection();
+ localConnection = fac.createConnection("localAdmin", "passwordA");
localConnection.setClientID("localClientId");
URI remoteURI = remoteBroker.getVmConnectorURI();
fac = new ActiveMQConnectionFactory(remoteURI);
- remoteConnection = fac.createConnection();
+ remoteConnection = fac.createConnection("remoteAdmin", "passwordB");
remoteConnection.setClientID("remoteClientId");
localSession = localConnection.createSession(false,
Session.AUTO_ACKNOWLEDGE);
@@ -109,6 +111,57 @@ public class NetworkAdvancedStatisticsTest extends
BaseNetworkTest {
return
"org/apache/activemq/network/localBroker-advancedNetworkStatistics.xml";
}
+ //Added for AMQ-9687 test network bridge local/remote exception stats
rolled up to the connector
+ @Test(timeout = 120 * 1000)
+ public void testNetworkAdvancedStatisticsErrors() throws Exception {
+ final NetworkConnector localNetworkConnector =
localBroker.getNetworkConnectorByName("local-to-remote");
+ assertNotNull(localNetworkConnector);
+
+ final String originalLocalPassword =
localNetworkConnector.getPassword();
+ final String originalRemotePassword =
localNetworkConnector.getRemotePassword();
+
+ assertTrue(localNetworkConnector.isAutoStart());
+ assertTrue(localNetworkConnector.isStarted());
+
+ // Use a non-retrying discovery agent so each induced fault produces a
single,
+ // definitive exception count instead of an ever-growing series of
reconnects.
+ ((SimpleDiscoveryAgent) ((DiscoveryNetworkConnector)
localNetworkConnector).getDiscoveryAgent())
+ .setMaxReconnectAttempts(1);
+
+ // Healthy bridge: no exceptions rolled up to the connector.
+ assertNetworkStats(false, false, 0L, 0L, localNetworkConnector);
+
+ try {
+ // Remote error only: a bad remote password is rejected by the
remote broker and
+ // surfaces as a remote exception on the bridge, rolled up to the
connector.
+ restartConnector(localNetworkConnector, originalLocalPassword,
"wrongRemotePassword");
+ assertNetworkStats(false, false, 0L, 1L, localNetworkConnector);
+
+ // Local error only: a bad local password is rejected by the local
broker and
+ // surfaces as a local exception on the bridge, rolled up to the
connector.
+ restartConnector(localNetworkConnector, "wrongLocalPassword",
originalRemotePassword);
+ assertNetworkStats(false, false, 1L, 0L, localNetworkConnector);
+ } finally {
+ restartConnector(localNetworkConnector, originalLocalPassword,
originalRemotePassword);
+ }
+ }
+
+ /**
+ * Stops the connector, applies the given local/remote credentials, and
starts it again,
+ * waiting for each transition to complete. A stop/start cycle resets the
connector's
+ * exception counters, so each scenario starts from a clean baseline.
+ */
+ private static void restartConnector(final NetworkConnector
networkConnector, final String localPassword, final String remotePassword)
throws Exception {
+ networkConnector.stop();
+ assertTrue(Wait.waitFor(networkConnector::isStopped));
+
+ networkConnector.setPassword(localPassword);
+ networkConnector.setRemotePassword(remotePassword);
+
+ networkConnector.start();
+ assertTrue(Wait.waitFor(networkConnector::isStarted));
+ }
+
//Added for AMQ-9437 test advancedStatistics for networkEnqueue and
networkDequeue
@Test(timeout = 60 * 1000)
public void testNetworkAdvancedStatistics() throws Exception {
@@ -191,7 +244,7 @@ public class NetworkAdvancedStatisticsTest extends
BaseNetworkTest {
assertEquals(lastIncludedSentMessageID,
localBrokerIncludedMessageFlowStats.getDequeuedMessageID().getValue());
assertNotNull(localBrokerIncludedMessageFlowStats.getDequeuedMessageBrokerInTime().getValue());
assertNotNull(localBrokerIncludedMessageFlowStats.getDequeuedMessageBrokerOutTime().getValue());
-
assertTrue(localBrokerIncludedMessageFlowStats.getDequeuedMessageClientID().getValue().startsWith("networkConnector"));
+
assertTrue(localBrokerIncludedMessageFlowStats.getDequeuedMessageClientID().getValue().startsWith("local-to-remote"));
assertNotNull(localBrokerIncludedMessageFlowStats.getDequeuedMessageTimestamp().getValue());
if(includedDestination.isTopic() && !durable) {
@@ -265,4 +318,35 @@ public class NetworkAdvancedStatisticsTest extends
BaseNetworkTest {
}));
}
+ protected static void assertNetworkStats(final boolean
connectorStartTimestampZeroExpected, final boolean
bridgeStartTimestampZeroExpected, final long localExceptionCount, final long
remoteExceptionCount, final NetworkConnector networkConnector) throws Exception
{
+ if(connectorStartTimestampZeroExpected) {
+ assertEquals(Long.valueOf(0L),
Long.valueOf(networkConnector.getStartedTimestamp()));
+ } else {
+ assertNotEquals(Long.valueOf(0L),
Long.valueOf(networkConnector.getStartedTimestamp()));
+ }
+
+ for(var networkBridge : networkConnector.activeBridges()) {
+ if(bridgeStartTimestampZeroExpected) {
+ assertEquals(Long.valueOf(0L),
Long.valueOf(networkBridge.getStartedTimestamp()));
+ } else {
+ assertNotEquals(Long.valueOf(0L),
Long.valueOf(networkBridge.getStartedTimestamp()));
+ }
+ }
+
+ // Bridges are ephemeral, so exception counts are asserted at the
connector level where
+ // they are rolled up. Failures arrive asynchronously, so first wait
for the counters to
+ // reach the expected values, then assert they are exact - the
connector uses a
+ // non-retrying discovery agent, so each fault yields a single,
definitive count.
+ assertTrue("Timed out waiting for connector exception counts local>="
+ localExceptionCount
+ + " remote>=" + remoteExceptionCount,
+ Wait.waitFor(new Wait.Condition() {
+ @Override
+ public boolean isSatisified() throws Exception {
+ return
networkConnector.getLocalExceptionCounter().getCount() >= localExceptionCount &&
+
networkConnector.getRemoteExceptionCounter().getCount() >= remoteExceptionCount;
+ }
+ }));
+ assertEquals(localExceptionCount,
networkConnector.getLocalExceptionCounter().getCount());
+ assertEquals(remoteExceptionCount,
networkConnector.getRemoteExceptionCounter().getCount());
+ }
}
diff --git
a/activemq-unit-tests/src/test/resources/org/apache/activemq/network/localBroker-advancedNetworkStatistics.xml
b/activemq-unit-tests/src/test/resources/org/apache/activemq/network/localBroker-advancedNetworkStatistics.xml
index f17fa9bcc5..f6438afd74 100644
---
a/activemq-unit-tests/src/test/resources/org/apache/activemq/network/localBroker-advancedNetworkStatistics.xml
+++
b/activemq-unit-tests/src/test/resources/org/apache/activemq/network/localBroker-advancedNetworkStatistics.xml
@@ -42,7 +42,11 @@
dynamicOnly = "false"
conduitSubscriptions = "true"
decreaseNetworkConsumerPriority = "false"
- name="networkConnector">
+ name="local-to-remote"
+ userName="localAdmin"
+ password="passwordA"
+ remoteUserName="remoteAdmin"
+ remotePassword="passwordB">
<dynamicallyIncludedDestinations>
<queue physicalName="include.test.foo"/>
<topic physicalName="include.test.durable"/>
@@ -56,10 +60,33 @@
</networkConnector>
</networkConnectors>
+ <plugins>
+ <authorizationPlugin>
+ <map>
+ <authorizationMap>
+ <authorizationEntries>
+ <authorizationEntry queue=">" read="admins" write="admins"
admin="admins" />
+ <authorizationEntry topic=">" read="admins" write="admins"
admin="admins" />
+ <authorizationEntry topic="ActiveMQ.Advisory.>" read="*"
write="*" admin="*"/>
+ </authorizationEntries>
+ <tempDestinationAuthorizationEntry>
+ <tempDestinationAuthorizationEntry read="admins" write="admins"
admin="admins"/>
+ </tempDestinationAuthorizationEntry>
+ </authorizationMap>
+ </map>
+ </authorizationPlugin>
+ <simpleAuthenticationPlugin>
+ <users>
+ <authenticationUser username="localAdmin" password="passwordA"
groups="admins"/>
+ <authenticationUser username="remoteAdmin" password="passwordB"
groups="admins"/>
+ </users>
+ </simpleAuthenticationPlugin>
+ </plugins>
+
<transportConnectors>
<transportConnector uri="tcp://localhost:61616"/>
</transportConnectors>
-
+
</broker>
</beans>
diff --git
a/activemq-unit-tests/src/test/resources/org/apache/activemq/network/remoteBroker-advancedNetworkStatistics.xml
b/activemq-unit-tests/src/test/resources/org/apache/activemq/network/remoteBroker-advancedNetworkStatistics.xml
index d2b3c0f33d..85d99e508e 100644
---
a/activemq-unit-tests/src/test/resources/org/apache/activemq/network/remoteBroker-advancedNetworkStatistics.xml
+++
b/activemq-unit-tests/src/test/resources/org/apache/activemq/network/remoteBroker-advancedNetworkStatistics.xml
@@ -37,9 +37,30 @@
</policyMap>
</destinationPolicy>
<networkConnectors>
- <networkConnector uri="static:(tcp://localhost:61616)" />
- </networkConnectors>
-
+ <networkConnector name="remote-to-local"
uri="static:(tcp://localhost:61616)" userName="localAdmin" password="passwordA"
remoteUserName="remoteAdmin" remotePassword="passwordB"/>
+ </networkConnectors>
+ <plugins>
+ <authorizationPlugin>
+ <map>
+ <authorizationMap>
+ <authorizationEntries>
+ <authorizationEntry queue=">" read="admins" write="admins"
admin="admins" />
+ <authorizationEntry topic=">" read="admins" write="admins"
admin="admins" />
+ <authorizationEntry topic="ActiveMQ.Advisory.>" read="*"
write="*" admin="*"/>
+ </authorizationEntries>
+ <tempDestinationAuthorizationEntry>
+ <tempDestinationAuthorizationEntry read="admins" write="admins"
admin="admins"/>
+ </tempDestinationAuthorizationEntry>
+ </authorizationMap>
+ </map>
+ </authorizationPlugin>
+ <simpleAuthenticationPlugin>
+ <users>
+ <authenticationUser username="localAdmin" password="passwordA"
groups="admins"/>
+ <authenticationUser username="remoteAdmin" password="passwordB"
groups="admins"/>
+ </users>
+ </simpleAuthenticationPlugin>
+ </plugins>
<transportConnectors>
<transportConnector uri="tcp://localhost:61617"/>
</transportConnectors>
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
For further information, visit: https://activemq.apache.org/contact
