Preventing duplicate delivery due to unacknowledged event while completing the subscription #596

ashwing · 2019-08-17T00:24:13Z

Preventing duplicate delivery due to unacknowledged event while completing the subscription

Issue #, if available:

Description of changes:

By submitting this pull request, I confirm that you can use, modify, copy, and redistribute this contribution, under the terms of your choice.

…eting the subscription

yatins47 · 2019-08-17T01:55:54Z

...is-client/src/main/java/software/amazon/kinesis/retrieval/fanout/FanOutRecordsPublisher.java

@@ -313,8 +319,24 @@ private void errorOccurred(RecordFlow triggeringFlow, Throwable t) {
        }
    }

-    private void clearRecordsDeliveryQueue() {
+    // This method is not thread safe. This needs to be executed after acquiring lock on this.lockObject
+    private void clearRecordsDeliveryQueue(boolean isSubscriptionCompleting) {


This method seems to be no-op if the delivery q is empty. Just check that in the beginning and return and continue if not. Below seems more easier to read as we just clear the queue immediately and then wait instead of doing the opposite since setting isAwaitingFinalAckForCurrentSubscription to true has anyways stopped events from getting scheduled. Also we can update the method name to say clearRecordsDeliveryQueueAndWaitForSubscriberNotification() or something with more context

if (recordsDeliveryQueue.isEmpty()) { return; } // This will prevent further events from getting scheduled isAwaitingFinalAckForCurrentSubscription = true; recordsDeliveryQueue.clear(); if(isSubscriptionCompleting) { try { // Wait for the configured time to get a notification for already delivered event, if any. lockObject.wait(TIME_TO_WAIT_FOR_FINAL_ACK_MILLIS); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } }

// Set isAwaitingFinalAckForCurrentSubscription to default value.
isAwaitingFinalAckForCurrentSubscription = false;

Ack. Will make changes

yatins47 · 2019-08-17T02:00:35Z

...is-client/src/main/java/software/amazon/kinesis/retrieval/fanout/FanOutRecordsPublisher.java

+            try {
+                if (!recordsDeliveryQueue.isEmpty()) {
+                    // Wait for the configured time to get a notification for already delivered event, if any.
+                    lockObject.wait(TIME_TO_WAIT_FOR_FINAL_ACK_MILLIS);


Why are we not waiting if the delivery queue is empty. There might still be a event in which was already delivered?

when the delivery queue is empty then it is guaranteed that there will be no valid in-flight event.

yasemin-amzn · 2019-08-19T17:15:12Z

...is-client/src/main/java/software/amazon/kinesis/retrieval/fanout/FanOutRecordsPublisher.java

@@ -132,6 +134,10 @@ public void notify(RecordsDeliveryAck recordsDeliveryAck) {
                triggeringFlow = evictAckedEventAndScheduleNextEvent(recordsDeliveryAck);
            } catch (Throwable t) {
                errorOccurred(triggeringFlow, t);
+            } finally {
+                if(isAwaitingFinalAckForCurrentSubscription) {
+                    lockObject.notifyAll();


This is critical line, worth adding comments explaining the design.

yasemin-amzn · 2019-08-19T17:18:41Z

...is-client/src/main/java/software/amazon/kinesis/retrieval/fanout/FanOutRecordsPublisher.java

@@ -160,7 +166,7 @@ RecordFlow evictAckedEventAndScheduleNextEvent(RecordsDeliveryAck recordsDeliver
            // Update the triggering flow for post scheduling upstream request.
            flowToBeReturned = recordsRetrievedContext.getRecordFlow();
            // Try scheduling the next event in the queue, if available.
-            if (recordsDeliveryQueue.peek() != null) {
+            if (!isAwaitingFinalAckForCurrentSubscription && recordsDeliveryQueue.peek() != null) {
                subscriber.onNext(recordsDeliveryQueue.peek().getRecordsRetrieved());


given we schedule the next event under two different conditions (here and in bufferCurrentEventAndScheduleIfRequired method) , and both required isShuttingDown check, let's move out the schedule call it its own method, with the check. And add comment on why the check is necessary.

i.e. scheduleNextEvent(event) {if !pendingShutdown; then subscriber.onNext(event)}

Ya. makes sense

yasemin-amzn · 2019-08-19T17:36:52Z

...is-client/src/main/java/software/amazon/kinesis/retrieval/fanout/FanOutRecordsPublisher.java

@@ -313,8 +319,24 @@ private void errorOccurred(RecordFlow triggeringFlow, Throwable t) {
        }
    }

-    private void clearRecordsDeliveryQueue() {
+    // This method is not thread safe. This needs to be executed after acquiring lock on this.lockObject
+    private void clearRecordsDeliveryQueue(boolean isSubscriptionCompleting) {


Couple comment on this method:

Why it's necessary to introduce split-brain logic here? Do we expect queue to be full, by the time subscribeToShard call is made? That invocation seems to be backup plan. We can either pull that out to its own method, or just let the start of a new subscription wait as well.

isAwaitingFinalAck variable is tightly couple with the queue status, why not update that within the if? i.e. if !queue.isEmpty -> then set the variable, and lock.wait()

Do we expect queue to be full, by the time subscribeToShard call is made? - Ideally no, just for precaution.
We can either pull that out to its own method - would be preferred instead of waiting
isAwaitingFinalAck variable is tightly couple with the queue status, why not update that within the if? - Updated as part of yatins comment

yasemin-amzn · 2019-08-19T17:39:50Z

...is-client/src/main/java/software/amazon/kinesis/retrieval/fanout/FanOutRecordsPublisher.java

@@ -82,6 +83,7 @@
    private long availableQueueSpace = 0;

    private BlockingQueue<RecordsRetrievedContext> recordsDeliveryQueue = new LinkedBlockingQueue<>(MAX_EVENT_BURST_FROM_SERVICE);
+    private boolean isAwaitingFinalAckForCurrentSubscription = false;


What do you think about introducing this variable as a more generic status indication boolean, i.e. pendingShutdown.

Waiting to get the ACK for the last submitted batch is current design, that flips the pendingShutdown status. We can later wait to drain all events in the queue etc..

pendingShutdown might be misleading, as we are not letting know what is being shutdown. Do you think pendingCurrentSubscriptionShutdown or pendingActiveSubscriptionShutdown would be better?

sure, either one is good.

yatins47 · 2019-08-19T19:10:26Z

...is-client/src/main/java/software/amazon/kinesis/retrieval/fanout/FanOutRecordsPublisher.java

@@ -166,8 +170,8 @@ RecordFlow evictAckedEventAndScheduleNextEvent(RecordsDeliveryAck recordsDeliver
            // Update the triggering flow for post scheduling upstream request.
            flowToBeReturned = recordsRetrievedContext.getRecordFlow();
            // Try scheduling the next event in the queue, if available.
-            if (!isAwaitingFinalAckForCurrentSubscription && recordsDeliveryQueue.peek() != null) {
-                subscriber.onNext(recordsDeliveryQueue.peek().getRecordsRetrieved());
+            if (recordsDeliveryQueue.peek() != null) {


check with isEmpty() ?

yatins47 · 2019-08-19T19:13:42Z

...is-client/src/main/java/software/amazon/kinesis/retrieval/fanout/FanOutRecordsPublisher.java

@@ -212,6 +216,14 @@ void bufferCurrentEventAndScheduleIfRequired(RecordsRetrieved recordsRetrieved,
        }
    }

+    // This method is not thread-safe. You need to acquire a lock in the caller in order to execute this.
+    // Schedule the next event only when the active subscription is not pending shutdown.
+    private void scheduleNextEvent(RecordsRetrieved recordsRetrieved) {


nit : what about checkSubscriptionStatusAndScheduleNextEvent ?

Hmmm we might add more checks in future and we can't keep extending the name. Hence just mentioning the core functionality.

yatins47 · 2019-08-19T19:15:07Z

...is-client/src/main/java/software/amazon/kinesis/retrieval/fanout/FanOutRecordsPublisher.java

            try {
                // Wait for the configured time to get a notification for already delivered event, if any.
                lockObject.wait(TIME_TO_WAIT_FOR_FINAL_ACK_MILLIS);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
+            // Clear the queue to remove any remaining entries from the queue.
+            recordsDeliveryQueue.clear();


Why cant we clear the queue before waiting ?

If we clear before waiting, the ack received will be marked as stale and records will be duplicated.

yatins47

LGTM!

yasemin-amzn · 2019-08-19T21:09:19Z

...is-client/src/main/java/software/amazon/kinesis/retrieval/fanout/FanOutRecordsPublisher.java

+            recordsDeliveryQueue.clear();
+        }
+        // Set pendingActiveSubscriptionShutdown to default value.
+        pendingActiveSubscriptionShutdown = false;


Should we log warning if pendingActiveSubscriptionShutdown is found true, or there are records in the queue?

Either of these cases points to something not working as expected, and logging status under such conditions can help us for RC'ing issues.

yep, they will be informational.

yasemin-amzn

Thank you!

Preventing duplicate delivery due to unacknowledged event while compl…

8d2b501

…eting the subscription

yatins47 reviewed Aug 17, 2019

View reviewed changes

Refactored clearRecordsDeliveryQueue logic and added comments

6f24734

yasemin-amzn reviewed Aug 19, 2019

View reviewed changes

Code refactoring as per review comments

452dde1

yatins47 reviewed Aug 19, 2019

View reviewed changes

Nit fix

b5f97fa

yatins47 approved these changes Aug 19, 2019

View reviewed changes

yasemin-amzn reviewed Aug 19, 2019

View reviewed changes

Add logging to unexpected subscription state scenario

2432575

yasemin-amzn approved these changes Aug 19, 2019

View reviewed changes

micah-jaffe approved these changes Aug 19, 2019

View reviewed changes

micah-jaffe merged commit a17d145 into awslabs:master Aug 19, 2019

micah-jaffe added this to the v2.2.2 milestone Aug 19, 2019

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Preventing duplicate delivery due to unacknowledged event while completing the subscription #596

Preventing duplicate delivery due to unacknowledged event while completing the subscription #596

ashwing commented Aug 17, 2019

yatins47 Aug 17, 2019 •

edited

Loading

ashwing Aug 19, 2019

yatins47 Aug 17, 2019

ashwing Aug 19, 2019

yasemin-amzn Aug 19, 2019

ashwing Aug 19, 2019

yasemin-amzn Aug 19, 2019

ashwing Aug 19, 2019

yasemin-amzn Aug 19, 2019

ashwing Aug 19, 2019 •

edited

Loading

yasemin-amzn Aug 19, 2019

ashwing Aug 19, 2019

yasemin-amzn Aug 19, 2019

yatins47 Aug 19, 2019

yatins47 Aug 19, 2019

ashwing Aug 19, 2019

yatins47 Aug 19, 2019

ashwing Aug 19, 2019

yatins47 left a comment

yasemin-amzn Aug 19, 2019

ashwing Aug 19, 2019

yasemin-amzn left a comment

Preventing duplicate delivery due to unacknowledged event while completing the subscription #596

Preventing duplicate delivery due to unacknowledged event while completing the subscription #596

Conversation

ashwing commented Aug 17, 2019

yatins47 Aug 17, 2019 • 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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ashwing Aug 19, 2019 • 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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

yatins47 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

yasemin-amzn left a comment

Choose a reason for hiding this comment

yatins47 Aug 17, 2019 •

edited

Loading

ashwing Aug 19, 2019 •

edited

Loading