Revert "cleanup indexing operation listener"

This reverts commit bb785483ae.

core/src/main/java/org/elasticsearch/index/shard/IndexShard.java

@@ -536,16 +536,16 @@ public class IndexShard extends AbstractIndexShardComponent implements IndicesCl
         return new Engine.Index(uid, doc, version, versionType, origin, startTime, autoGeneratedIdTimestamp, isRetry);
     }
 
-    public Engine.IndexResult index(final Engine.Index index) {
+    public Engine.IndexResult index(Engine.Index index) {
         ensureWriteAllowed(index);
         Engine engine = getEngine();
         return index(engine, index);
     }
 
-    private Engine.IndexResult index(final Engine engine, final Engine.Index index) {
+    private Engine.IndexResult index(Engine engine, Engine.Index index) {
         active.set(true);
         final Engine.IndexResult result;
-        indexingOperationListeners.preIndex(index);
+        index = indexingOperationListeners.preIndex(index);
         try {
             if (logger.isTraceEnabled()) {
                 logger.trace("index [{}][{}]{}", index.type(), index.id(), index.docs());
@@ -585,16 +585,16 @@ public class IndexShard extends AbstractIndexShardComponent implements IndicesCl
         return new Engine.Delete(type, id, uid, version, versionType, origin, startTime);
     }
 
-    public Engine.DeleteResult delete(final Engine.Delete delete) {
+    public Engine.DeleteResult delete(Engine.Delete delete) {
         ensureWriteAllowed(delete);
         Engine engine = getEngine();
         return delete(engine, delete);
     }
 
-    private Engine.DeleteResult delete(final Engine engine, final Engine.Delete delete) {
+    private Engine.DeleteResult delete(Engine engine, Engine.Delete delete) {
         active.set(true);
         final Engine.DeleteResult result;
-        indexingOperationListeners.preDelete(delete);
+        delete = indexingOperationListeners.preDelete(delete);
         try {
             if (logger.isTraceEnabled()) {
                 logger.trace("delete [{}]", delete.uid().text());

core/src/main/java/org/elasticsearch/index/shard/IndexingOperationListener.java

@@ -30,25 +30,44 @@ import java.util.List;
  */
 public interface IndexingOperationListener {
 
-    /** Called before the indexing occurs */
+    /**
-    default void preIndex(Engine.Index operation) {}
+     * Called before the indexing occurs.
+     */
+    default Engine.Index preIndex(Engine.Index operation) {
+        return operation;
+    }
 
-    /** Called after the indexing operation occurred */
+    /**
+     * Called after the indexing operation occurred.
+     */
     default void postIndex(Engine.Index index, Engine.IndexResult result) {}
 
-    /** Called after the indexing operation occurred with exception */
+    /**
+     * Called after the indexing operation occurred with exception.
+     */
     default void postIndex(Engine.Index index, Exception ex) {}
 
-    /** Called before the delete occurs */
+    /**
-    default void preDelete(Engine.Delete delete) {}
+     * Called before the delete occurs.
+     */
+    default Engine.Delete preDelete(Engine.Delete delete) {
+        return delete;
+    }
 
-    /** Called after the delete operation occurred */
+
+    /**
+     * Called after the delete operation occurred.
+     */
     default void postDelete(Engine.Delete delete, Engine.DeleteResult result) {}
 
-    /** Called after the delete operation occurred with exception */
+    /**
+     * Called after the delete operation occurred with exception.
+     */
     default void postDelete(Engine.Delete delete, Exception ex) {}
 
-    /** A Composite listener that multiplexes calls to each of the listeners methods */
+    /**
+     * A Composite listener that multiplexes calls to each of the listeners methods.
+     */
     final class CompositeListener implements IndexingOperationListener{
         private final List<IndexingOperationListener> listeners;
         private final Logger logger;
@@ -59,7 +78,7 @@ public interface IndexingOperationListener {
         }
 
         @Override
-        public void preIndex(Engine.Index operation) {
+        public Engine.Index preIndex(Engine.Index operation) {
             assert operation != null;
             for (IndexingOperationListener listener : listeners) {
                 try {
@@ -68,11 +87,12 @@ public interface IndexingOperationListener {
                     logger.warn((Supplier<?>) () -> new ParameterizedMessage("preIndex listener [{}] failed", listener), e);
                 }
             }
+            return operation;
         }
 
         @Override
         public void postIndex(Engine.Index index, Engine.IndexResult result) {
-            assert index != null && result != null;
+            assert index != null;
             for (IndexingOperationListener listener : listeners) {
                 try {
                     listener.postIndex(index, result);
@@ -96,7 +116,7 @@ public interface IndexingOperationListener {
         }
 
         @Override
-        public void preDelete(Engine.Delete delete) {
+        public Engine.Delete preDelete(Engine.Delete delete) {
             assert delete != null;
             for (IndexingOperationListener listener : listeners) {
                 try {
@@ -105,11 +125,12 @@ public interface IndexingOperationListener {
                     logger.warn((Supplier<?>) () -> new ParameterizedMessage("preDelete listener [{}] failed", listener), e);
                 }
             }
+            return delete;
         }
 
         @Override
         public void postDelete(Engine.Delete delete, Engine.DeleteResult result) {
-            assert delete != null && result != null;
+            assert delete != null;
             for (IndexingOperationListener listener : listeners) {
                 try {
                     listener.postDelete(delete, result);

core/src/main/java/org/elasticsearch/index/shard/InternalIndexingStats.java

@@ -65,11 +65,12 @@ final class InternalIndexingStats implements IndexingOperationListener {
     }
 
     @Override
-    public void preIndex(Engine.Index operation) {
+    public Engine.Index preIndex(Engine.Index operation) {
         if (!operation.origin().isRecovery()) {
             totalStats.indexCurrent.inc();
             typeStats(operation.type()).indexCurrent.inc();
         }
+        return operation;
     }
 
     @Override
@@ -95,11 +96,13 @@ final class InternalIndexingStats implements IndexingOperationListener {
     }
 
     @Override
-    public void preDelete(Engine.Delete delete) {
+    public Engine.Delete preDelete(Engine.Delete delete) {
         if (!delete.origin().isRecovery()) {
             totalStats.deleteCurrent.inc();
             typeStats(delete.type()).deleteCurrent.inc();
         }
+        return delete;
+
     }
 
     @Override

core/src/test/java/org/elasticsearch/index/IndexModuleTests.java

@@ -233,8 +233,9 @@ public class IndexModuleTests extends ESTestCase {
         AtomicBoolean executed = new AtomicBoolean(false);
         IndexingOperationListener listener = new IndexingOperationListener() {
             @Override
-            public void preIndex(Engine.Index operation) {
+            public Engine.Index preIndex(Engine.Index operation) {
                 executed.set(true);
+                return operation;
             }
         };
         module.addIndexOperationListener(listener);

core/src/test/java/org/elasticsearch/index/shard/IndexShardTests.java

@@ -558,8 +558,9 @@ public class IndexShardTests extends IndexShardTestCase {
         shard.close("simon says", true);
         shard = reinitShard(shard, new IndexingOperationListener() {
             @Override
-            public void preIndex(Engine.Index operation) {
+            public Engine.Index preIndex(Engine.Index operation) {
                 preIndex.incrementAndGet();
+                return operation;
             }
 
             @Override
@@ -577,8 +578,9 @@ public class IndexShardTests extends IndexShardTestCase {
             }
 
             @Override
-            public void preDelete(Engine.Delete delete) {
+            public Engine.Delete preDelete(Engine.Delete delete) {
                 preDelete.incrementAndGet();
+                return delete;
             }
 
             @Override
@@ -1119,8 +1121,9 @@ public class IndexShardTests extends IndexShardTestCase {
         final AtomicInteger postDelete = new AtomicInteger();
         IndexingOperationListener listener = new IndexingOperationListener() {
             @Override
-            public void preIndex(Engine.Index operation) {
+            public Engine.Index preIndex(Engine.Index operation) {
                 preIndex.incrementAndGet();
+                return operation;
             }
 
             @Override
@@ -1129,8 +1132,9 @@ public class IndexShardTests extends IndexShardTestCase {
             }
 
             @Override
-            public void preDelete(Engine.Delete delete) {
+            public Engine.Delete preDelete(Engine.Delete delete) {
                 preDelete.incrementAndGet();
+                return delete;
             }
 
             @Override

core/src/test/java/org/elasticsearch/index/shard/IndexingOperationListenerTests.java

@@ -40,8 +40,9 @@ public class IndexingOperationListenerTests extends ESTestCase{
         AtomicInteger postDeleteException = new AtomicInteger();
         IndexingOperationListener listener = new IndexingOperationListener() {
             @Override
-            public void preIndex(Engine.Index operation) {
+            public Engine.Index preIndex(Engine.Index operation) {
                 preIndex.incrementAndGet();
+                return operation;
             }
 
             @Override
@@ -55,8 +56,9 @@ public class IndexingOperationListenerTests extends ESTestCase{
             }
 
             @Override
-            public void preDelete(Engine.Delete delete) {
+            public Engine.Delete preDelete(Engine.Delete delete) {
                 preDelete.incrementAndGet();
+                return delete;
             }
 
             @Override
@@ -72,7 +74,7 @@ public class IndexingOperationListenerTests extends ESTestCase{
 
         IndexingOperationListener throwingListener = new IndexingOperationListener() {
             @Override
-            public void preIndex(Engine.Index operation) {
+            public Engine.Index preIndex(Engine.Index operation) {
                 throw new RuntimeException();
             }
 
@@ -87,7 +89,7 @@ public class IndexingOperationListenerTests extends ESTestCase{
             }
 
             @Override
-            public void preDelete(Engine.Delete delete) {
+            public Engine.Delete preDelete(Engine.Delete delete) {
                 throw new RuntimeException();
             }
 

modules/reindex/src/test/java/org/elasticsearch/index/reindex/CancelTests.java

@@ -199,23 +199,23 @@ public class CancelTests extends ReindexTestCase {
     public static class BlockingOperationListener implements IndexingOperationListener {
 
         @Override
-        public void preIndex(Engine.Index index) {
+        public Engine.Index preIndex(Engine.Index index) {
-            preCheck(index, index.type());
+            return preCheck(index, index.type());
         }
 
         @Override
-        public void preDelete(Engine.Delete delete) {
+        public Engine.Delete preDelete(Engine.Delete delete) {
-            preCheck(delete, delete.type());
+            return preCheck(delete, delete.type());
         }
 
-        private void preCheck(Engine.Operation operation, String type) {
+        private <T extends Engine.Operation> T preCheck(T operation, String type) {
             if ((TYPE.equals(type) == false) || (operation.origin() != Origin.PRIMARY)) {
-                return;
+                return operation;
             }
 
             try {
                 if (ALLOWED_OPERATIONS.tryAcquire(30, TimeUnit.SECONDS)) {
-                    return;
+                    return operation;
                 }
             } catch (InterruptedException e) {
                 throw new RuntimeException(e);