Repository: incubator-geode Updated Branches: refs/heads/develop 274f67cae -> 10b6befa2
GEODE-478: Adding a stat for resized gateway batches Adding a stat to track when we resize a gateway sender batch if it is too large. Updating the test of resizing a batch to test that the stat is incremented. Project: http://git-wip-us.apache.org/repos/asf/incubator-geode/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-geode/commit/10b6befa Tree: http://git-wip-us.apache.org/repos/asf/incubator-geode/tree/10b6befa Diff: http://git-wip-us.apache.org/repos/asf/incubator-geode/diff/10b6befa Branch: refs/heads/develop Commit: 10b6befa2c11ea5de372953a23c56712cdf1a203 Parents: 274f67c Author: Dan Smith <[email protected]> Authored: Fri Mar 25 16:00:39 2016 -0700 Committer: Dan Smith <[email protected]> Committed: Fri Mar 25 16:00:39 2016 -0700 ---------------------------------------------------------------------- .../internal/cache/wan/GatewaySenderStats.java | 25 +++++++++++++++++++- .../wan/GatewaySenderEventRemoteDispatcher.java | 2 ++ ...arallelGatewaySenderOperationsDUnitTest.java | 5 ++++ 3 files changed, 31 insertions(+), 1 deletion(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/10b6befa/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/wan/GatewaySenderStats.java ---------------------------------------------------------------------- diff --git a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/wan/GatewaySenderStats.java b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/wan/GatewaySenderStats.java index 34e82de..15a6dee 100644 --- a/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/wan/GatewaySenderStats.java +++ b/geode-core/src/main/java/com/gemstone/gemfire/internal/cache/wan/GatewaySenderStats.java @@ -60,6 +60,8 @@ public class GatewaySenderStats { protected static final String BATCHES_DISTRIBUTED = "batchesDistributed"; /** Name of the batches redistributed statistic */ protected static final String BATCHES_REDISTRIBUTED = "batchesRedistributed"; + /** Name of the batches resized statistic */ + protected static final String BATCHES_RESIZED = "batchesResized"; /** Name of the unprocessed events added by primary statistic */ protected static final String UNPROCESSED_TOKENS_ADDED_BY_PRIMARY = "unprocessedTokensAddedByPrimary"; /** Name of the unprocessed events added by secondary statistic */ @@ -105,6 +107,8 @@ public class GatewaySenderStats { protected static int batchesDistributedId; /** Id of the batches redistributed statistic */ protected static int batchesRedistributedId; + /** Id of the batches resized statistic */ + protected static int batchesResizedId; /** Id of the unprocessed events added by primary statistic */ protected static int unprocessedTokensAddedByPrimaryId; /** Id of the unprocessed events added by secondary statistic */ @@ -191,6 +195,10 @@ public class GatewaySenderStats { "Number of batches of events removed from the event queue and resent.", "operations", false), f.createIntCounter + (BATCHES_RESIZED, + "Number of batches that were resized because they were too large", + "operations", false), + f.createIntCounter (UNPROCESSED_TOKENS_ADDED_BY_PRIMARY, "Number of tokens added to the secondary's unprocessed token map by the primary (though a listener).", "tokens"), @@ -260,6 +268,7 @@ public class GatewaySenderStats { batchDistributionTimeId = type.nameToId(BATCH_DISTRIBUTION_TIME); batchesDistributedId = type.nameToId(BATCHES_DISTRIBUTED); batchesRedistributedId = type.nameToId(BATCHES_REDISTRIBUTED); + batchesResizedId = type.nameToId(BATCHES_RESIZED); unprocessedTokensAddedByPrimaryId = type.nameToId(UNPROCESSED_TOKENS_ADDED_BY_PRIMARY); unprocessedEventsAddedBySecondaryId = type.nameToId(UNPROCESSED_EVENTS_ADDED_BY_SECONDARY); unprocessedEventsRemovedByPrimaryId = type.nameToId(UNPROCESSED_EVENTS_REMOVED_BY_PRIMARY); @@ -433,6 +442,14 @@ public class GatewaySenderStats { } /** + * Returns the current value of the batchesResized" stat. + * @return the current value of the batchesResized" stat + */ + public int getBatchesResized() { + return this.stats.getInt(batchesResizedId); + } + + /** * Increments the value of the "batchesRedistributed" stat by 1. */ public void incBatchesRedistributed() { @@ -440,6 +457,13 @@ public class GatewaySenderStats { } /** + * Increments the value of the "batchesRedistributed" stat by 1. + */ + public void incBatchesResized() { + this.stats.incInt(batchesResizedId, 1); + } + + /** * Sets the "eventQueueSize" stat. * @param size The size of the queue */ @@ -740,5 +764,4 @@ public class GatewaySenderStats { public Statistics getStats(){ return stats; } - } http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/10b6befa/geode-wan/src/main/java/com/gemstone/gemfire/internal/cache/wan/GatewaySenderEventRemoteDispatcher.java ---------------------------------------------------------------------- diff --git a/geode-wan/src/main/java/com/gemstone/gemfire/internal/cache/wan/GatewaySenderEventRemoteDispatcher.java b/geode-wan/src/main/java/com/gemstone/gemfire/internal/cache/wan/GatewaySenderEventRemoteDispatcher.java index ad2be2b..22defbc 100644 --- a/geode-wan/src/main/java/com/gemstone/gemfire/internal/cache/wan/GatewaySenderEventRemoteDispatcher.java +++ b/geode-wan/src/main/java/com/gemstone/gemfire/internal/cache/wan/GatewaySenderEventRemoteDispatcher.java @@ -203,6 +203,7 @@ public class GatewaySenderEventRemoteDispatcher implements int currentBatchId = this.processor.getBatchId(); connection = getConnection(true); int batchIdForThisConnection = this.processor.getBatchId(); + GatewaySenderStats statistics = this.sender.getStatistics(); // This means we are writing to a new connection than the previous batch. // i.e The connection has been reset. It also resets the batchId. if (currentBatchId != batchIdForThisConnection @@ -259,6 +260,7 @@ public class GatewaySenderEventRemoteDispatcher implements logger.warn(LocalizedMessage.create( LocalizedStrings.GatewaySenderEventRemoteDispatcher_MESSAGE_TOO_LARGE_EXCEPTION, new Object[] { events.size(), newBatchSize }), e); this.processor.setBatchSize(newBatchSize); + statistics.incBatchesResized(); } else { ex = e; http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/10b6befa/geode-wan/src/test/java/com/gemstone/gemfire/internal/cache/wan/parallel/ParallelGatewaySenderOperationsDUnitTest.java ---------------------------------------------------------------------- diff --git a/geode-wan/src/test/java/com/gemstone/gemfire/internal/cache/wan/parallel/ParallelGatewaySenderOperationsDUnitTest.java b/geode-wan/src/test/java/com/gemstone/gemfire/internal/cache/wan/parallel/ParallelGatewaySenderOperationsDUnitTest.java index 4e66c6f..869b050 100644 --- a/geode-wan/src/test/java/com/gemstone/gemfire/internal/cache/wan/parallel/ParallelGatewaySenderOperationsDUnitTest.java +++ b/geode-wan/src/test/java/com/gemstone/gemfire/internal/cache/wan/parallel/ParallelGatewaySenderOperationsDUnitTest.java @@ -585,6 +585,11 @@ public class ParallelGatewaySenderOperationsDUnitTest extends WANTestBase { vm2.invoke(() -> createReceiver( nyPort )); vm2.invoke(() -> createPartitionedRegion( regionName, null, 0, 100, isOffHeap() )); validateRegionSizes( regionName, numPuts, vm2 ); + + vm4.invoke(() -> { + final AbstractGatewaySender sender = (AbstractGatewaySender) cache.getGatewaySender("ln"); + assertTrue(sender.getStatistics().getBatchesResized() > 0); + }); ignoredMTLE.remove(); ignoredGIOE.remove(); }
