fix rebase collateral

DataDog · Oct 13, 2020 · 183d368 · 183d368
1 parent 784f099
commit 183d368
Show file tree

Hide file tree

Showing 4 changed files with 27 additions and 20 deletions.
diff --git a/src/main/java/com/timgroup/statsd/StatsDBlockingProcessor.java b/src/main/java/com/timgroup/statsd/StatsDBlockingProcessor.java
@@ -16,6 +16,10 @@ public class StatsDBlockingProcessor extends StatsDProcessor {
 
     private class ProcessingTask extends StatsDProcessor.ProcessingTask {
 
+        public ProcessingTask(int id) {
+            super(id);
+        }
+
         @Override
         public void run() {
             ByteBuffer sendBuffer;
@@ -76,7 +80,7 @@ public void run() {
                             writeBuilderToSendBuffer(sendBuffer);
                         }
 
-                        if (null == messages.peek()) {
+                        if (null == processorWorkQueue[this.processorQueueId].peek()) {
                             outboundQueue.put(sendBuffer);
                             sendBuffer = bufferPool.borrow();
                         }
@@ -96,7 +100,6 @@ public void run() {
             aggregator.stop();
             endSignal.countDown();
         }
-
     }
 
     StatsDBlockingProcessor(final int queueSize, final StatsDClientErrorHandler handler,
@@ -115,8 +118,8 @@ public void run() {
     }
 
     @Override
-    protected ProcessingTask createProcessingTask() {
-        return new ProcessingTask();
+    protected ProcessingTask createProcessingTask(int id) {
+        return new ProcessingTask(id);
     }
 
     StatsDBlockingProcessor(final StatsDBlockingProcessor processor)

diff --git a/src/main/java/com/timgroup/statsd/StatsDNonBlockingProcessor.java b/src/main/java/com/timgroup/statsd/StatsDNonBlockingProcessor.java
@@ -20,8 +20,7 @@ private class ProcessingTask extends StatsDProcessor.ProcessingTask {
         private final int processorQueueId;
 
         public ProcessingTask(int id) {
-            super();
-            this.processorQueueId = id;
+            super(id);
         }
 
         @Override
@@ -144,8 +143,8 @@ public void run() {
     }
 
     @Override
-    protected ProcessingTask createProcessingTask() {
-        return new ProcessingTask();
+    protected ProcessingTask createProcessingTask(int id) {
+        return new ProcessingTask(id);
     }
 
     StatsDNonBlockingProcessor(final StatsDNonBlockingProcessor processor) throws Exception {

diff --git a/src/main/java/com/timgroup/statsd/StatsDProcessor.java b/src/main/java/com/timgroup/statsd/StatsDProcessor.java
@@ -137,7 +137,7 @@ public Thread newThread(final Runnable runnable) {
                 processor.getAggregator().getFlushInterval());
     }
 
-    protected abstract ProcessingTask createProcessingTask();
+    protected abstract ProcessingTask createProcessingTask(int id);
 
     protected abstract boolean send(final Message message);
 
@@ -172,7 +172,7 @@ public static int getThreadId() {
     public void run() {
 
         for (int i = 0 ; i < workers ; i++) {
-            executor.submit(createProcessingTask());
+            executor.submit(createProcessingTask(i));
         }
 
         boolean done = false;

diff --git a/src/test/java/com/timgroup/statsd/TelemetryTest.java b/src/test/java/com/timgroup/statsd/TelemetryTest.java
@@ -31,11 +31,16 @@ public static class FakeProcessor extends StatsDProcessor {
         public final List<Message> messages = new ArrayList<>();
 
         FakeProcessor(final StatsDClientErrorHandler handler) throws Exception {
-            super(0, handler, 0, 1, 1, 0, 0);
+            super(0, handler, 0, 1, 1, 1, 0, 0);
         }
 
 
         private class FakeProcessingTask extends StatsDProcessor.ProcessingTask {
+
+            public FakeProcessingTask(int id) {
+                super(id);
+            }
+
             @Override
             public void run() {}
         }
@@ -50,8 +55,8 @@ public boolean send(final Message msg) {
         public void run(){}
 
         @Override
-        protected ProcessingTask createProcessingTask() {
-            return new FakeProcessingTask();
+        protected ProcessingTask createProcessingTask(int id) {
+            return new FakeProcessingTask(id);
         }
 
         public List<Message> getMessages() {
@@ -82,12 +87,12 @@ public StatsDNonBlockingTelemetry(final String prefix, final int queueSize, Stri
                                        final StatsDClientErrorHandler errorHandler, Callable<SocketAddress> addressLookup,
                                        final int timeout, final int bufferSize, final int maxPacketSizeBytes,
                                        String entityID, final int poolSize, final int processorWorkers,
-                                       final int senderWorkers, boolean blocking, final boolean enableTelemetry,
-                                       final int telemetryFlushInterval)
+                                       final int senderWorkers, final int lockShardGrain, boolean blocking,
+                                       final boolean enableTelemetry, final int telemetryFlushInterval)
                 throws StatsDClientException {
                 super(prefix, queueSize, constantTags, errorHandler, addressLookup, addressLookup, timeout,
                         bufferSize, maxPacketSizeBytes, entityID, poolSize, processorWorkers, senderWorkers,
-                        blocking, enableTelemetry, telemetryFlushInterval, 0, 0);
+                        lockShardGrain, blocking, enableTelemetry, telemetryFlushInterval, 0, 0);
         }
     };
 
@@ -98,13 +103,13 @@ public StatsDNonBlockingTelemetry build() throws StatsDClientException {
             if (addressLookup != null) {
                 return new StatsDNonBlockingTelemetry(prefix, queueSize, constantTags, errorHandler,
                         addressLookup, timeout, socketBufferSize, maxPacketSizeBytes, entityID,
-                        bufferPoolSize, processorWorkers, senderWorkers, blocking, enableTelemetry,
-                        telemetryFlushInterval);
+                        bufferPoolSize, processorWorkers, senderWorkers, lockShardGrain, blocking,
+                        enableTelemetry, telemetryFlushInterval);
             } else {
                 return new StatsDNonBlockingTelemetry(prefix, queueSize, constantTags, errorHandler,
                         staticStatsDAddressResolution(hostname, port), timeout, socketBufferSize, maxPacketSizeBytes,
-                        entityID, bufferPoolSize, processorWorkers, senderWorkers, blocking, enableTelemetry,
-                        telemetryFlushInterval);
+                        entityID, bufferPoolSize, processorWorkers, senderWorkers, lockShardGrain, blocking,
+                        enableTelemetry, telemetryFlushInterval);
             }
         }
     }