issue #1083: does not continue in the work-stage when document could …

…be skipped
smclab · Oct 15, 2024 · 246e9fc · 246e9fc
1 parent a75c7bc
commit 246e9fc
Show file tree

Hide file tree

Showing 2 changed files with 9 additions and 3 deletions.
diff --git a/core/app/datasource/src/main/java/io/openk9/datasource/pipeline/actor/EnrichPipeline.java b/core/app/datasource/src/main/java/io/openk9/datasource/pipeline/actor/EnrichPipeline.java
@@ -315,9 +315,7 @@ yield initPipeline(
 						dataPayload.getContentId()
 					);
 
-					var buffer = Json.encodeToBuffer(dataPayload);
-
-					replyTo.tell(new Processor.Success(buffer.getBytes(), heldMessage));
+					replyTo.tell(new Skip(heldMessage));
 
 					return Behaviors.stopped();
 				}
@@ -400,4 +398,6 @@ private record InternalResponseWrapper(byte[] jsonObject) implements Processor.C
 
 	private record InternalError(DataProcessException exception) implements Processor.Command {}
 
+	public record Skip(HeldMessage heldMessage) implements Processor.Response {}
+
 }
diff --git a/.../app/datasource/src/main/java/io/openk9/datasource/pipeline/stages/working/WorkStage.java b/.../app/datasource/src/main/java/io/openk9/datasource/pipeline/stages/working/WorkStage.java
@@ -30,6 +30,7 @@
 import io.openk9.common.util.ShardingKey;
 import io.openk9.common.util.ingestion.PayloadType;
 import io.openk9.datasource.pipeline.actor.DataProcessException;
+import io.openk9.datasource.pipeline.actor.EnrichPipeline;
 import io.openk9.datasource.pipeline.actor.Scheduling;
 import io.openk9.datasource.pipeline.actor.WorkStageException;
 import io.openk9.datasource.pipeline.actor.common.AggregateBehavior;
@@ -252,6 +253,11 @@ private Behavior<Command> onPostProcess(PostProcess postProcess) {
 				heldMessage
 			));
 
+		}
+		else if (response instanceof EnrichPipeline.Skip) {
+
+			this.replyTo.tell(new Done(heldMessage));
+
 		}
 		else if (response instanceof Processor.Failure failure) {