[sampling] sampling on counts to be disabled when aggregation is enabled.

src/main/java/com/timgroup/statsd/NonBlockingStatsDClient.java

@@ -1145,6 +1145,11 @@ private void send(final Message message) {
 
     // send double with sample rate
     private void send(String aspect, final double value, Message.Type type, double sampleRate, String[] tags) {
+        if (statsDProcessor.getAggregator().getFlushInterval() != 0
+                && type == Message.Type.COUNT && !Double.isNaN(sampleRate)) {
+            sampleRate = Double.NaN;
+        }
+
         if (Double.isNaN(sampleRate) || !isInvalidSample(sampleRate)) {
 
             sendMetric(new StatsDMessage<Double>(aspect, type, Double.valueOf(value), sampleRate, tags) {
@@ -1162,6 +1167,11 @@ private void send(String aspect, final double value, Message.Type type, String[]
 
     // send long with sample rate
     private void send(String aspect, final long value, Message.Type type, double sampleRate, String[] tags) {
+        if (statsDProcessor.getAggregator().getFlushInterval() != 0
+                && type == Message.Type.COUNT && !Double.isNaN(sampleRate)) {
+            sampleRate = Double.NaN;
+        }
+
         if (Double.isNaN(sampleRate) || !isInvalidSample(sampleRate)) {
             sendMetric(new StatsDMessage<Long>(aspect, type, value, sampleRate, tags) {
                 @Override protected void writeValue(StringBuilder builder) {

src/test/java/com/timgroup/statsd/NonBlockingStatsDClientTest.java

@@ -1114,6 +1114,41 @@ public void testBasicCountAggregation() throws Exception {
         }
     }
 
+    @Test(timeout=5000L)
+    public void testSampledCountAggregation() throws Exception {
+        final RecordingErrorHandler errorHandler = new RecordingErrorHandler();
+        final NonBlockingStatsDClient testClient = new NonBlockingStatsDClientBuilder()
+            .prefix("my.prefix")
+            .hostname("localhost")
+            .port(STATSD_SERVER_PORT)
+            .enableTelemetry(false)  // don't want additional packets
+            .enableAggregation(true)
+            .aggregationFlushInterval(3000)
+            .errorHandler(errorHandler)
+            .build();
+
+        try {
+            for (int i=0 ; i<10 ; i++) {
+                // NOTE: because aggregation is enabled, sampling is disabled, so all
+                // counts should be accounted for, as if sample rate were 1.0.
+                testClient.count("top.level.count", i, 0.1);
+            }
+            for (int i=0 ; i<10 ; i++) {
+                testClient.increment("top.level.count");
+            }
+
+            server.waitForMessage("my.prefix");
+
+            List<String> messages = server.messagesReceived();
+
+            assertThat(messages.size(), comparesEqualTo(1));
+            assertThat(messages, hasItem(comparesEqualTo("my.prefix.top.level.count:55|c")));
+
+        } finally {
+            testClient.stop();
+        }
+    }
+
     @Test(timeout=5000L)
     public void testBasicSetAggregation() throws Exception {
         final RecordingErrorHandler errorHandler = new RecordingErrorHandler();