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 756ffbed7d2..6b2798cf6ca 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 void resetStats(){ } } + @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 82fc9ca1899..34d7d4b198e 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 a1ad16f8dc0..b6e61dbf009 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 @@ -219,4 +219,29 @@ public void setRemoteUserName(String remoteUserName) { 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 12fc22a2d95..08b3c76f74c 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 @@ -94,4 +94,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 762c6eeffdb..c8bb1386c5f 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.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 void start() throws Exception { 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 void onException(IOException error) { 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 void run() { startedLatch.countDown(); localStartedLatch.countDown(); staticDestinationsLatch.countDown(); + startedTimestamp.set(0L); ss.throwFirstException(); } @@ -650,12 +665,21 @@ protected void startRemoteBridge() throws Exception { @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 void serviceLocalException(Throwable error) { 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 void serviceLocalException(MessageDispatch messageDispatch, Throwable err 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 long getEnqueueCounter() { 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 3b9696cf925..5532f3f98a5 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 void onServiceAdd(DiscoveryEvent event) { 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 void onServiceAdd(DiscoveryEvent event) { 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 void onServiceAdd(DiscoveryEvent event) { } 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 fb9e3d9f086..42d0cd38f86 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 a9157dcbd1a..721003a4a14 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 @@ -16,6 +16,8 @@ */ package org.apache.activemq.network; +import java.util.Set; + import org.apache.activemq.management.CountStatisticImpl; import org.apache.activemq.management.StatsImpl; @@ -29,13 +31,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."); - addStatistics(Set.of(enqueues, dequeues, receivedCount)); + addStatistics(Set.of(enqueues, dequeues, receivedCount, localExceptionCount, remoteExceptionCount)); } /** @@ -68,6 +74,26 @@ public CountStatisticImpl getReceivedCount() { 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()) { @@ -75,6 +101,8 @@ public void reset() { enqueues.reset(); dequeues.reset(); receivedCount.reset(); + localExceptionCount.reset(); + remoteExceptionCount.reset(); } } @@ -84,6 +112,8 @@ public void setEnabled(boolean enabled) { enqueues.setEnabled(enabled); dequeues.setEnabled(enabled); receivedCount.setEnabled(enabled); + localExceptionCount.setEnabled(enabled); + remoteExceptionCount.setEnabled(enabled); } public void setParent(NetworkBridgeStatistics parent) { @@ -91,10 +121,14 @@ public void setParent(NetworkBridgeStatistics parent) { 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 a51345f3d6c..2852b0ffa8e 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.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.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 bridges = new ConcurrentHashMap(); + 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() { @@ -164,11 +174,15 @@ public static ActiveMQDestination[] getDurableTopicDestinations(final Set 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 b2bddbd89c5..296426fa3cb 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 @@ -332,6 +332,21 @@ public long getDequeueCounter() { 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 a7bbacfc542..add9c5ea54b 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.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 @@ protected void doSetUp(boolean deleteAllMessages) throws Exception { 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,8 +111,59 @@ protected String getLocalBrokerURI() { return "org/apache/activemq/network/localBroker-advancedNetworkStatistics.xml"; } - //Added for AMQ-9437 test advancedStatistics for networkEnqueue and networkDequeue + //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 { // create a remote durable consumer to create demand @@ -207,7 +260,7 @@ public boolean isSatisified() throws Exception { 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) { @@ -331,4 +384,35 @@ public boolean isSatisified() throws Exception { })); } -} \ No newline at end of file + 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 f17fa9bcc53..f6438afd748 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"> @@ -56,10 +60,33 @@ + + + + + + + + + + + + + + + + + + + + + + + - + 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 d2b3c0f33df..85d99e508e0 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 @@ - - - + + + + + + + + + + + + + + + + + + + + + + + +