feat: Added ODPEventManager implementation #487

zashraf1985 · 2022-09-01T06:22:21Z

Summary

Added ODPEventManager implementation.

Test plan

Unit tests pending.

Jira

OASIS-8386

jaeopt

It looks good overall. A few suggestions.

jaeopt · 2022-09-01T16:07:31Z

core-api/src/main/java/com/optimizely/ab/odp/ODPEventManager.java

+    private static final Logger logger = LoggerFactory.getLogger(ODPEventManager.class);
+    private static final int DEFAULT_BATCH_SIZE = 10;
+    private static final int DEFAULT_QUEUE_SIZE = 10000;
+    private static final int FLUSH_INTERVAL = 1000;


Let's make the default interval to 10secs for all SDKs consistent.

jaeopt · 2022-09-01T16:09:24Z

core-api/src/main/java/com/optimizely/ab/odp/ODPEventManager.java

+    private final List<ODPEvent> currentBatch = new ArrayList<>();
+    private final BlockingQueue<ODPEvent> eventQueue = new LinkedBlockingQueue<>();


Thread-safety for these - currentBatch and eventQueue?

Moved currentBatch to the thread instance itself which runs in a loop which means there will be NO simultaneous access on this at all.

eventQueue is an instance of LinkedBlockingQueue which itself is thread safe

jaeopt · 2022-09-01T16:24:29Z

core-api/src/main/java/com/optimizely/ab/odp/ODPEventManager.java

+        public void run() {
+            while (true) {
+                try {
+                    ODPEvent nextEvent = eventQueue.poll(FLUSH_INTERVAL, TimeUnit.MILLISECONDS);


If we use this fixed interval between events (instead of overrall interval between flush), the max gap between flushes can be increased up to (batchSize * FLUSH_INTERVAL). Is this what we have for logx event batching?
Some ODP events can be time-sensitive (relatively more than logx events), so it'll be good if we have more tight control of the extra delay.

Fixed this whole thing. Made this timeout dynamic based on how much time elapsed since last flush.

jaeopt · 2022-09-01T21:35:13Z

core-api/src/main/java/com/optimizely/ab/odp/ODPEventManager.java

+    }
+
+    public void updateSettings(ODPConfig odpConfig) {
+        this.odpConfig = odpConfig;


Is this replacement thread-safe?

Made odpConfig volatile to make it thread safe.

jaeopt · 2022-09-01T21:45:32Z

core-api/src/main/java/com/optimizely/ab/odp/ODPEventManager.java

+                    numAttempts ++;
+                } while (numAttempts < MAX_RETRIES && statusCode != null && (statusCode == 0 || statusCode >= 500));
+            } else {
+                logger.warn("ODPConfig not ready, discarding event batch");


Shouldn't this be debug level? When ODP is not integrated, we do not want dump many warning messages.

jaeopt · 2022-09-01T21:49:48Z

core-api/src/main/java/com/optimizely/ab/odp/ODPEventManager.java

+        }
+
+        if (!eventQueue.offer(event)) {
+            logger.warn("Failed to Process ODP Event. Event Queue is not accepting any more events");


In what case, event is discarded here? It should be logger.error in this case.

The usual case would be when queue is at capacity but we are already checking that before this. This check is just there because offer can return false in case of an unexpected error. Changing log level to error

jaeopt · 2022-09-01T21:50:12Z

core-api/src/main/java/com/optimizely/ab/odp/ODPEventManager.java

+        return data;
+    }
+
+    private void processEvent(ODPEvent event) {


We can consider filtering out silently events when apiKey/host is null. I see you discards them when flushing below. They don't need to be queued and discarded if ODP is not integrated.

zashraf1985 · 2022-09-13T01:44:41Z

@jaeopt I incorporated all the changes you requested. Unit tests are pending. Take a look when you have time.

zashraf1985 · 2022-09-14T02:00:04Z

@jaeopt I incorporated all the feedback and added unit tests. Please take a look when you can.

jaeopt

All the changes and new tests look good.
I found one more potential issue with the event queue batch interval. See my comments.
Also a few more tests suggested.

jaeopt · 2022-09-14T16:23:29Z

core-api/src/main/java/com/optimizely/ab/odp/ODPEventManager.java


    private Boolean isRunning = false;
-    private ODPConfig odpConfig;
+    private volatile ODPConfig odpConfig;


Can we add a comment why this should be volatile?

jaeopt · 2022-09-14T16:28:24Z

core-api/src/main/java/com/optimizely/ab/odp/ODPEvent.java

@@ -26,8 +27,8 @@ public class ODPEvent {
    public ODPEvent(String type, String action, Map<String, String> identifiers, Map<String, Object> data) {


What about adding @Nullable, @Nonnull annotations to all params?

jaeopt · 2022-09-14T16:28:53Z

core-api/src/main/java/com/optimizely/ab/odp/ODPEventManager.java

-        this(odpConfig, apiManager);
-        this.batchSize = batchSize;
-        this.queueSize = queueSize;
+    public ODPEventManager(ODPConfig odpConfig, ODPApiManager apiManager, Integer batchSize, Integer queueSize, Integer flushInterval) {


Same here. @Nonnull, @Nullable will be helpful.

jaeopt · 2022-09-14T16:38:17Z

core-api/src/main/java/com/optimizely/ab/odp/ODPEventManager.java

-        for (ODPEvent event: events) {
-            sendEvent(event);
-        }
+    public void identifyUser(String vuid, String userId) {


vuid should be @Nullable, good for Android-SDK support only. We need to filter it out for other server-sdk usage.

jaeopt · 2022-09-14T16:50:19Z

core-api/src/main/java/com/optimizely/ab/odp/ODPEventManager.java

    private EventDispatcherThread eventDispatcherThread;

    private final ODPApiManager apiManager;
-    private final List<ODPEvent> currentBatch = new ArrayList<>();
    private final BlockingQueue<ODPEvent> eventQueue = new LinkedBlockingQueue<>();


A comment about thready-safety requirement will be helpful.

jaeopt · 2022-09-14T17:30:25Z

core-api/src/main/java/com/optimizely/ab/odp/ODPEventManager.java

        @Override
        public void run() {
            while (true) {
                try {
-                    ODPEvent nextEvent = eventQueue.poll(FLUSH_INTERVAL, TimeUnit.MILLISECONDS);
+                    long nextFlushMillis = Math.max(0, flushInterval - (new Date().getTime() - lastFlushTime));
+                    ODPEvent nextEvent = eventQueue.poll(nextFlushMillis, TimeUnit.MILLISECONDS);


If we do not get events for a while (after 1sec), this will make it busy-waiting all the time instead of blocking.

Added a check to see if the batch has events, then adding a timeout otherwise doing a no argument poll which indefinitely blocks till the next event arrives.

jaeopt · 2022-09-14T17:42:26Z

core-api/src/main/java/com/optimizely/ab/odp/ODPEventManager.java

        @Override
        public void run() {
            while (true) {
                try {
-                    ODPEvent nextEvent = eventQueue.poll(FLUSH_INTERVAL, TimeUnit.MILLISECONDS);
+                    long nextFlushMillis = Math.max(0, flushInterval - (new Date().getTime() - lastFlushTime));


If nextFlushTime depends on the lastFlushTime, the event arriving after a gap can be fired without waiting enough for batching. Example -
(Event1) (Event2) (flush. lastFlushTime updated) (990ms gap) (Event3) (10ms gap) (flush on nextFlushTime)

Fixed it! Now the nextFlushTime is determined when a new batch is started.

jaeopt · 2022-09-14T22:02:23Z

core-api/src/test/java/com/optimizely/ab/odp/ODPEventManagerTest.java

+        eventManager.start();
+        ODPEvent event = new ODPEvent("test-type", "test-action", Collections.emptyMap(), Collections.emptyMap());
+        eventManager.sendEvent(event);
+        logbackVerifier.expectMessage(Level.DEBUG, "Unable to Process Event. ODPConfig is not ready.");


Suggested change

logbackVerifier.expectMessage(Level.DEBUG, "Unable to Process Event. ODPConfig is not ready.");

logbackVerifier.expectMessage(Level.DEBUG, "Unable to Process ODP Event. ODPConfig is not ready.");

jaeopt

LGTM

added ODPEventManager implementation

089ff22

zashraf1985 requested a review from a team as a code owner September 1, 2022 06:22

optimizely-codereview-bot assigned zashraf1985 Sep 1, 2022

jaeopt requested changes Sep 1, 2022

View reviewed changes

incorporated review feedback. Unit tests are still pending.

432db4d

Added unit tests

6d3bb18

zashraf1985 removed their assignment Sep 14, 2022

optimizely-codereview-bot requested a review from jaeopt September 14, 2022 02:00

optimizely-codereview-bot added the 🤖 4 hrs label Sep 14, 2022

jaeopt requested changes Sep 14, 2022

View reviewed changes

optimizely-codereview-bot removed the 🤖 4 hrs label Sep 14, 2022

optimizely-codereview-bot assigned zashraf1985 Sep 14, 2022

zashraf1985 added 2 commits September 14, 2022 22:54

Incorporated review feedback

b7883eb

Added a comment on an instance variable

8caa93e

zashraf1985 removed their assignment Sep 15, 2022

optimizely-codereview-bot requested a review from jaeopt September 15, 2022 05:59

optimizely-codereview-bot added 🤖 4 hrs 🤖 8 hrs and removed 🤖 4 hrs labels Sep 15, 2022

jaeopt approved these changes Sep 16, 2022

View reviewed changes

zashraf1985 merged commit 2a38338 into master Sep 16, 2022

zashraf1985 deleted the zeeshan/ats-event-manager branch September 16, 2022 17:24

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat: Added ODPEventManager implementation #487

feat: Added ODPEventManager implementation #487

zashraf1985 commented Sep 1, 2022

jaeopt left a comment

jaeopt Sep 1, 2022

zashraf1985 Sep 13, 2022

jaeopt Sep 1, 2022

zashraf1985 Sep 12, 2022

jaeopt Sep 1, 2022

zashraf1985 Sep 13, 2022

jaeopt Sep 1, 2022

zashraf1985 Sep 12, 2022

jaeopt Sep 1, 2022

zashraf1985 Sep 13, 2022

jaeopt Sep 1, 2022

zashraf1985 Sep 12, 2022

jaeopt Sep 1, 2022

zashraf1985 Sep 13, 2022

zashraf1985 commented Sep 13, 2022

zashraf1985 commented Sep 14, 2022

jaeopt left a comment

jaeopt Sep 14, 2022

jaeopt Sep 14, 2022

jaeopt Sep 14, 2022

jaeopt Sep 14, 2022

jaeopt Sep 14, 2022

zashraf1985 Sep 15, 2022 •

edited

Loading

jaeopt Sep 14, 2022

zashraf1985 Sep 15, 2022

jaeopt Sep 14, 2022

zashraf1985 Sep 15, 2022

jaeopt Sep 14, 2022

jaeopt left a comment

		private final List<ODPEvent> currentBatch = new ArrayList<>();
		private final BlockingQueue<ODPEvent> eventQueue = new LinkedBlockingQueue<>();

		@@ -26,8 +27,8 @@ public class ODPEvent {
		public ODPEvent(String type, String action, Map<String, String> identifiers, Map<String, Object> data) {

	logbackVerifier.expectMessage(Level.DEBUG, "Unable to Process Event. ODPConfig is not ready.");
	logbackVerifier.expectMessage(Level.DEBUG, "Unable to Process ODP Event. ODPConfig is not ready.");

feat: Added ODPEventManager implementation #487

feat: Added ODPEventManager implementation #487

Conversation

zashraf1985 commented Sep 1, 2022

Summary

Test plan

Jira

jaeopt left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

zashraf1985 commented Sep 13, 2022

zashraf1985 commented Sep 14, 2022

jaeopt left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

zashraf1985 Sep 15, 2022 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jaeopt left a comment

Choose a reason for hiding this comment

zashraf1985 Sep 15, 2022 •

edited

Loading