Fix double notify bug

Author: jasontedor

server/src/main/java/org/elasticsearch/index/shard/GlobalCheckpointListeners.java

@@ -92,10 +92,12 @@ public interface GlobalCheckpointListener {
     synchronized void add(final long currentGlobalCheckpoint, final GlobalCheckpointListener listener) {
         if (closed) {
             executor.execute(() -> listener.accept(UNASSIGNED_SEQ_NO, new IndexShardClosedException(shardId)));
+            return;
         }
         if (lastKnownGlobalCheckpoint > currentGlobalCheckpoint) {
             // notify directly
-            executor.execute(() -> notifyListener(listener, lastKnownGlobalCheckpoint, null));
+            executor.execute(() -> listener.accept(lastKnownGlobalCheckpoint, null));
+            return;
         } else {
             if (listeners == null) {
                 listeners = new ArrayList<>();
@@ -112,6 +114,10 @@ public void close() throws IOException {
         notifyListeners(UNASSIGNED_SEQ_NO, new IndexShardClosedException(shardId));
     }
 
+    synchronized int pendingListeners() {
+        return listeners == null ? 0 : listeners.size();
+    }
+
     /**
      * Invoke to notify all registered listeners of an updated global checkpoint.
      *

server/src/test/java/org/elasticsearch/index/shard/GlobalCheckpointListenersTests.java

@@ -26,6 +26,7 @@
 import org.mockito.ArgumentCaptor;
 
 import java.io.IOException;
+import java.io.UncheckedIOException;
 import java.util.List;
 import java.util.concurrent.BrokenBarrierException;
 import java.util.concurrent.CopyOnWriteArrayList;
@@ -350,11 +351,21 @@ public void testConcurrency() throws BrokenBarrierException, InterruptedExceptio
         globalCheckpointListeners.globalCheckpointUpdated(globalCheckpoint.get());
         final CyclicBarrier barrier = new CyclicBarrier(3);
         final int numberOfIterations = randomIntBetween(1, 1024);
-
+        final AtomicBoolean closed = new AtomicBoolean();
         final Thread updatingThread = new Thread(() -> {
             awaitQuietly(barrier);
             for (int i = 0; i < numberOfIterations; i++) {
-                globalCheckpointListeners.globalCheckpointUpdated(globalCheckpoint.incrementAndGet());
+                if (rarely() && closed.get() == false) {
+                    closed.set(true);
+                    try {
+                        globalCheckpointListeners.close();
+                    } catch (final IOException e) {
+                        throw new UncheckedIOException(e);
+                    }
+                }
+                if (closed.get() == false) {
+                    globalCheckpointListeners.globalCheckpointUpdated(globalCheckpoint.incrementAndGet());
+                }
             }
             awaitQuietly(barrier);
         });
@@ -381,7 +392,10 @@ public void testConcurrency() throws BrokenBarrierException, InterruptedExceptio
         barrier.await();
         barrier.await();
         // one last update to ensure all listeners are notified
-        globalCheckpointListeners.globalCheckpointUpdated(globalCheckpoint.incrementAndGet());
+        if (closed.get() == false) {
+            globalCheckpointListeners.globalCheckpointUpdated(globalCheckpoint.incrementAndGet());
+        }
+        assertThat(globalCheckpointListeners.pendingListeners(), equalTo(0));
         executor.shutdown();
         executor.awaitTermination(Long.MAX_VALUE, TimeUnit.SECONDS);
         for (final AtomicBoolean invocation : invocations) {