Make SearchUpdateMapper asynchronous.

Signed-off-by: Yufei Cai <yufei.cai@bosch.io>
eclipse-ditto · Oct 25, 2021 · b0bce92 · b0bce92
1 parent 9afd80c
commit b0bce92
Show file tree

Hide file tree

Showing 3 changed files with 16 additions and 14 deletions.
diff --git a/...ipse/ditto/thingsearch/service/persistence/write/streaming/DefaultSearchUpdateMapper.java b/...ipse/ditto/thingsearch/service/persistence/write/streaming/DefaultSearchUpdateMapper.java
@@ -16,7 +16,9 @@
 
 import org.eclipse.ditto.thingsearch.service.persistence.write.model.AbstractWriteModel;
 
+import akka.NotUsed;
 import akka.actor.ActorSystem;
+import akka.stream.javadsl.Source;
 
 /**
  * Default {@code SearchUpdateMapper} for custom search update processing.
@@ -32,8 +34,8 @@ protected DefaultSearchUpdateMapper(final ActorSystem actorSystem) {
     }
 
     @Override
-    public List<AbstractWriteModel> processWriteModels(final List<AbstractWriteModel> writeModels) {
-        return writeModels;
+    public Source<List<AbstractWriteModel>, NotUsed> processWriteModels(final List<AbstractWriteModel> writeModels) {
+        return Source.single(writeModels);
     }
 
 }
diff --git a/...eclipse/ditto/thingsearch/service/persistence/write/streaming/MongoSearchUpdaterFlow.java b/...eclipse/ditto/thingsearch/service/persistence/write/streaming/MongoSearchUpdaterFlow.java
@@ -29,7 +29,6 @@
 import org.eclipse.ditto.thingsearch.service.persistence.write.model.WriteResultAndErrors;
 
 import com.mongodb.MongoBulkWriteException;
-import com.mongodb.bulk.BulkWriteResult;
 import com.mongodb.client.model.BulkWriteOptions;
 import com.mongodb.client.model.WriteModel;
 import com.mongodb.reactivestreams.client.MongoCollection;
@@ -110,16 +109,15 @@ public Flow<Source<AbstractWriteModel, NotUsed>, WriteResultAndErrors, NotUsed>
 
         final Flow<List<AbstractWriteModel>, WriteResultAndErrors, NotUsed> writeFlow =
                 Flow.<List<AbstractWriteModel>>create()
-                        .map(writeModels -> {
-                            try {
-                                return searchUpdateMapper.processWriteModels(writeModels);
-                            } catch (final Exception e) {
-                                LOGGER.error(
-                                        "Skipping mapping of search update write models because an unexpected error occurred.",
-                                        e);
-                                return writeModels;
-                            }
-                        })
+                        .flatMapConcat(writeModels -> searchUpdateMapper.processWriteModels(writeModels)
+                                .recoverWithRetries(1,
+                                        new PFBuilder<Throwable, Source<List<AbstractWriteModel>, NotUsed>>()
+                                                .matchAny(e -> {
+                                                    LOGGER.error("Skipping mapping of search update write models " +
+                                                            "because an unexpected error occurred.", e);
+                                                    return Source.single(writeModels);
+                                                })
+                                                .build()))
                         .mapAsync(1, MongoSearchUpdaterFlow::toIncrementalMongo)
                         .flatMapMerge(parallelism, writeModels ->
                                 executeBulkWrite(shouldAcknowledge, writeModels).async(DISPATCHER_NAME, parallelism));

diff --git a/...org/eclipse/ditto/thingsearch/service/persistence/write/streaming/SearchUpdateMapper.java b/...org/eclipse/ditto/thingsearch/service/persistence/write/streaming/SearchUpdateMapper.java
@@ -20,10 +20,12 @@
 import org.eclipse.ditto.thingsearch.service.common.config.SearchConfig;
 import org.eclipse.ditto.thingsearch.service.persistence.write.model.AbstractWriteModel;
 
+import akka.NotUsed;
 import akka.actor.AbstractExtensionId;
 import akka.actor.ActorSystem;
 import akka.actor.ExtendedActorSystem;
 import akka.actor.Extension;
+import akka.stream.javadsl.Source;
 
 /**
  * Search Update Mapper to be loaded by reflection.
@@ -48,7 +50,7 @@ protected SearchUpdateMapper(final ActorSystem actorSystem) {
      * Should not throw an exception. If a exception is thrown, the mapping is ignored.
      * If no search update should be executed an empty list can be returned.
      */
-    public abstract List<AbstractWriteModel> processWriteModels(final List<AbstractWriteModel> writeModels);
+    public abstract Source<List<AbstractWriteModel>, NotUsed> processWriteModels(final List<AbstractWriteModel> writeModels);
 
     /**
      * Load a {@code SearchUpdateListener} dynamically according to the search configuration.