Merge pull request #2017 from beyonnex-io/bugfix/background-sync-stuck

fix search background sync being stuck
eclipse-ditto · Sep 13, 2024 · 9baafda · 9baafda
2 parents 5d6059b + cb31b34
commit 9baafda
Show file tree

Hide file tree

Showing 5 changed files with 18 additions and 30 deletions.
diff --git a/deployment/docker/docker-compose.yml b/deployment/docker/docker-compose.yml
@@ -12,7 +12,7 @@ version: '3.4'
 
 services:
   mongodb:
-    image: docker.io/mongo:6.0
+    image: docker.io/mongo:7.0
     deploy:
       resources:
         limits:

diff --git a/deployment/docker/sandbox/docker-compose.yml b/deployment/docker/sandbox/docker-compose.yml
@@ -12,7 +12,7 @@ version: '3.4'
 
 services:
   mongodb:
-    image: docker.io/mongo:6.0
+    image: docker.io/mongo:7.0
     deploy:
       resources:
         limits:

diff --git a/deployment/kubernetes/deploymentFiles/mongodb-statefulset/mongodb-statefulset.yaml b/deployment/kubernetes/deploymentFiles/mongodb-statefulset/mongodb-statefulset.yaml
@@ -33,7 +33,7 @@ spec:
       terminationGracePeriodSeconds: 30
       containers:
         - name: mongodb
-          image: docker.io/mongo:6.0
+          image: docker.io/mongo:7.0
           command:
             - mongod
             - --bind_ip

diff --git a/deployment/kubernetes/deploymentFiles/mongodb/mongodb.yaml b/deployment/kubernetes/deploymentFiles/mongodb/mongodb.yaml
@@ -31,7 +31,7 @@ spec:
     spec:
       containers:
         - name: mongodb
-          image: docker.io/mongo:6.0
+          image: docker.io/mongo:7.0
           command:
             - mongod
             - --storageEngine

diff --git a/...g/eclipse/ditto/internal/utils/persistence/mongo/streaming/MongoTimestampPersistence.java b/...g/eclipse/ditto/internal/utils/persistence/mongo/streaming/MongoTimestampPersistence.java
@@ -19,27 +19,23 @@
 
 import javax.annotation.Nullable;
 
+import org.apache.pekko.Done;
+import org.apache.pekko.NotUsed;
+import org.apache.pekko.japi.Pair;
+import org.apache.pekko.stream.Materializer;
+import org.apache.pekko.stream.javadsl.BroadcastHub;
+import org.apache.pekko.stream.javadsl.RestartSource;
+import org.apache.pekko.stream.javadsl.Source;
 import org.bson.Document;
 import org.eclipse.ditto.internal.utils.pekko.streaming.TimestampPersistence;
 import org.eclipse.ditto.internal.utils.persistence.mongo.DittoMongoClient;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import com.mongodb.MongoCommandException;
 import com.mongodb.client.model.CreateCollectionOptions;
 import com.mongodb.reactivestreams.client.MongoCollection;
 import com.mongodb.reactivestreams.client.MongoDatabase;
 
-import org.apache.pekko.Done;
-import org.apache.pekko.NotUsed;
-import org.apache.pekko.japi.Pair;
-import org.apache.pekko.japi.pf.PFBuilder;
-import org.apache.pekko.stream.Attributes;
-import org.apache.pekko.stream.Materializer;
-import org.apache.pekko.stream.javadsl.BroadcastHub;
-import org.apache.pekko.stream.javadsl.RestartSource;
-import org.apache.pekko.stream.javadsl.Source;
-
 /**
  * MongoDB implementation of {@link TimestampPersistence}.
  */
@@ -187,21 +183,13 @@ private static Source<Done, NotUsed> repeatableCreateCappedCollectionSource(
                 .sizeInBytes(cappedCollectionSizeInBytes)
                 .maxDocuments(1);
 
-        return Source.lazySource(
-                () -> Source.fromPublisher(database.createCollection(collectionName, collectionOptions)))
-                .mapMaterializedValue(whatever -> NotUsed.getInstance())
-                .map(nullValue -> Done.done())
-                .withAttributes(Attributes.inputBuffer(1, 1))
-                .recoverWithRetries(1, new PFBuilder<Throwable, Source<Done, NotUsed>>()
-                        .match(MongoCommandException.class,
-                                MongoTimestampPersistence::isCollectionAlreadyExistsError,
-                                error -> Source.single(Done.done()))
-                        .build());
-
-    }
-
-    private static boolean isCollectionAlreadyExistsError(final MongoCommandException error) {
-        return error.getErrorCode() == COLLECTION_ALREADY_EXISTS_ERROR_CODE;
+        return Source.fromPublisher(database.listCollectionNames())
+                .filter(name -> name.equals(collectionName))
+                .map(name -> Done.done())
+                .orElse(
+                        Source.fromPublisher(database.createCollection(collectionName, collectionOptions))
+                                .map(aVoid -> Done.done())
+                );
     }
 
 }