Setting up size to events Queue

Author: ldecheverz-split

client/src/main/java/io/split/client/EventClientImpl.java

@@ -54,23 +54,15 @@ public class EventClientImpl implements EventClient {
     private final TelemetryRuntimeProducer _telemetryRuntimeProducer;
 
     ThreadFactory eventClientThreadFactory(final String name) {
-        return new ThreadFactory() {
-            @Override
-            public Thread newThread(final Runnable r) {
-                return new Thread(new Runnable() {
-                    @Override
-                    public void run() {
-                        Thread.currentThread().setPriority(MIN_PRIORITY);
-                        r.run();
-                    }
-                }, name);
-            }
-        };
+        return r -> new Thread(() -> {
+            Thread.currentThread().setPriority(MIN_PRIORITY);
+            r.run();
+        }, name);
     }
 
 
     public static EventClientImpl create(CloseableHttpClient httpclient, URI eventsRootTarget, int maxQueueSize, long flushIntervalMillis, int waitBeforeShutdown, TelemetryRuntimeProducer telemetryRuntimeProducer) throws URISyntaxException {
-        return new EventClientImpl(new LinkedBlockingQueue<WrappedEvent>(),
+        return new EventClientImpl(new LinkedBlockingQueue<>(maxQueueSize),
                 httpclient,
                 Utils.appendPath(eventsRootTarget, "api/events/bulk"),
                 maxQueueSize,
@@ -131,10 +123,14 @@ public boolean track(Event event, int eventSize) {
             if (event == null) {
                 return false;
             }
-            _eventQueue.put(new WrappedEvent(event, eventSize));
-            _telemetryRuntimeProducer.recordEventStats(EventsDataRecordsEnum.EVENTS_QUEUED, 1);
+            if(_eventQueue.offer(new WrappedEvent(event, eventSize))) {
+                _telemetryRuntimeProducer.recordEventStats(EventsDataRecordsEnum.EVENTS_QUEUED, 1);
+            }
+            else {
+                _telemetryRuntimeProducer.recordEventStats(EventsDataRecordsEnum.EVENTS_DROPPED, 1);
+            }
 
-        } catch (ClassCastException | NullPointerException | InterruptedException e) {
+        } catch (ClassCastException | NullPointerException | IllegalArgumentException e) {
             _telemetryRuntimeProducer.recordEventStats(EventsDataRecordsEnum.EVENTS_DROPPED, 1);
             _log.warn("Interruption when adding event withed while adding message %s.", event);
             return false;
@@ -164,7 +160,7 @@ public void run() {
             List<Event> events = new ArrayList<>();
             long accumulated = 0;
             try {
-                while (true) {
+                while (!Thread.currentThread().isInterrupted()) {
                     WrappedEvent data = _eventQueue.take();
                     Event event = data.event();
                     Long size = data.size();

client/src/test/java/io/split/client/EventsClientImplTest.java

@@ -1,6 +1,8 @@
 package io.split.client;
 
 import io.split.client.dtos.Event;
+import io.split.engine.segments.SegmentSynchronizationTaskImp;
+import io.split.telemetry.domain.enums.EventsDataRecordsEnum;
 import io.split.telemetry.storage.InMemoryTelemetryStorage;
 import io.split.telemetry.storage.TelemetryStorage;
 import org.apache.hc.client5.http.classic.methods.HttpUriRequest;
@@ -12,8 +14,12 @@
 import org.mockito.Mockito;
 
 import java.io.IOException;
+import java.lang.reflect.Field;
+import java.lang.reflect.Modifier;
 import java.net.URI;
 import java.net.URISyntaxException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
 import java.util.concurrent.LinkedBlockingQueue;
 
 public class EventsClientImplTest {
@@ -74,4 +80,34 @@ public void testEventsFlushedWhenSizeLimitReached() throws URISyntaxException, I
         Thread.sleep(2000);
         Mockito.verify(client, Mockito.times(1)).execute((HttpUriRequest) Mockito.any());
     }
+
+    @Test
+    public void testEventDropped() throws URISyntaxException, NoSuchFieldException, IllegalAccessException {
+        CloseableHttpClient client = Mockito.mock(CloseableHttpClient.class);
+        Field eventDroppedConsumer = EventClientImpl.class.getDeclaredField("_consumerExecutor");
+        ExecutorService executorService = Executors.newFixedThreadPool(2);
+        eventDroppedConsumer.setAccessible(true);
+
+        Field modifiersField = Field.class.getDeclaredField("modifiers");
+        modifiersField.setAccessible(true);
+        modifiersField.setInt(eventDroppedConsumer, eventDroppedConsumer.getModifiers() & ~Modifier.FINAL);
+        executorService.execute(() -> {
+
+        });
+        EventClientImpl eventClient = new EventClientImpl(new LinkedBlockingQueue<>(2),
+                client,
+                URI.create("https://kubernetesturl.com/split"),
+                10000, // Long queue so it doesn't flush by # of events
+                100000, // Long period so it doesn't flush by timeout expiration.
+                0, TELEMETRY_STORAGE);
+        eventClient.close();
+        eventDroppedConsumer.set(eventClient, executorService); // 1ms
+        for (int i = 0; i < 3; ++i) {
+            Event event = new Event();
+            eventClient.track(event, 1);
+        }
+
+        Mockito.verify(TELEMETRY_STORAGE, Mockito.times(2)).recordEventStats(EventsDataRecordsEnum.EVENTS_QUEUED, 1);
+        Mockito.verify(TELEMETRY_STORAGE, Mockito.times(1)).recordEventStats(EventsDataRecordsEnum.EVENTS_DROPPED, 1);
+    }
 }

client/src/test/java/io/split/engine/segments/SegmentSynchronizationTaskImpTest.java

@@ -100,14 +100,13 @@ public void testFetchAllAsynchronousAndGetFalse() throws NoSuchFieldException, I
         Mockito.doNothing().when(segmentFetcher).callLoopRun(Mockito.anyBoolean(),Mockito.anyBoolean());
 
         // Before executing, we'll update the map of segmentFecthers via reflection.
-        Field backoffBase = SegmentSynchronizationTaskImp.class.getDeclaredField("_segmentFetchers");
-        backoffBase.setAccessible(true);
+        Field segmentFetchersForced = SegmentSynchronizationTaskImp.class.getDeclaredField("_segmentFetchers");
+        segmentFetchersForced.setAccessible(true);
         Field modifiersField = Field.class.getDeclaredField("modifiers");
         modifiersField.setAccessible(true);
-        modifiersField.setInt(backoffBase, backoffBase.getModifiers() & ~Modifier.FINAL);
+        modifiersField.setInt(segmentFetchersForced, segmentFetchersForced.getModifiers() & ~Modifier.FINAL);
 
-        backoffBase.set(fetchers, _segmentFetchers); // 1ms
-        fetcher1.set(segmentFetcher);
+        segmentFetchersForced.set(fetchers, _segmentFetchers); // 1ms
         boolean fetch = fetchers.fetchAllSynchronous();
         Assert.assertEquals(false, fetch);
     }
@@ -117,16 +116,18 @@ public void testFetchAllAsynchronousAndGetTrue() throws NoSuchFieldException, Il
         SDKReadinessGates gates = new SDKReadinessGates();
         SegmentCache segmentCache = Mockito.mock(SegmentCache.class);
 
+        ConcurrentMap<String, SegmentFetcher> _segmentFetchers = Maps.newConcurrentMap();
         SegmentChangeFetcher segmentChangeFetcher = Mockito.mock(SegmentChangeFetcher.class);
         SegmentFetcherImp segmentFetcher = Mockito.mock(SegmentFetcherImp.class);
         final SegmentSynchronizationTaskImp fetchers = new SegmentSynchronizationTaskImp(segmentChangeFetcher, 1L, 1, gates, segmentCache, TELEMETRY_STORAGE);
 
         // Before executing, we'll update the map of segmentFecthers via reflection.
-        Field backoffBase = SegmentSynchronizationTaskImp.class.getDeclaredField("_segmentFetchers");
-        backoffBase.setAccessible(true);
+        Field segmentFetchersForced = SegmentSynchronizationTaskImp.class.getDeclaredField("_segmentFetchers");
+        segmentFetchersForced.setAccessible(true);
         Field modifiersField = Field.class.getDeclaredField("modifiers");
         modifiersField.setAccessible(true);
-        modifiersField.setInt(backoffBase, backoffBase.getModifiers() & ~Modifier.FINAL);
+        modifiersField.setInt(segmentFetchersForced, segmentFetchersForced.getModifiers() & ~Modifier.FINAL);
+        segmentFetchersForced.set(fetchers, _segmentFetchers); // 1ms
         Mockito.doNothing().when(segmentFetcher).callLoopRun(Mockito.anyBoolean(),Mockito.anyBoolean());
         Mockito.when(segmentFetcher.runWhitCacheHeader()).thenReturn(true);
         Mockito.when(segmentFetcher.fetchAndUpdate(Mockito.anyBoolean())).thenReturn(true);