OZG-7811 OZG-8099 Use loaders in Aggregator

aggregation-manager-job/src/main/java/de/ozgcloud/aggregation/Aggregation.java

+/*
+ * Copyright (C) 2025 Das Land Schleswig-Holstein vertreten durch den
+ * Ministerpräsidenten des Landes Schleswig-Holstein
+ * Staatskanzlei
+ * Abteilung Digitalisierung und zentrales IT-Management der Landesregierung
+ *
+ * Lizenziert unter der EUPL, Version 1.2 oder - sobald
+ * diese von der Europäischen Kommission genehmigt wurden -
+ * Folgeversionen der EUPL ("Lizenz");
+ * Sie dürfen dieses Werk ausschließlich gemäß
+ * dieser Lizenz nutzen.
+ * Eine Kopie der Lizenz finden Sie hier:
+ *
+ * https://joinup.ec.europa.eu/collection/eupl/eupl-text-eupl-12
+ *
+ * Sofern nicht durch anwendbare Rechtsvorschriften
+ * gefordert oder in schriftlicher Form vereinbart, wird
+ * die unter der Lizenz verbreitete Software "so wie sie
+ * ist", OHNE JEGLICHE GEWÄHRLEISTUNG ODER BEDINGUNGEN -
+ * ausdrücklich oder stillschweigend - verbreitet.
+ * Die sprachspezifischen Genehmigungen und Beschränkungen
+ * unter der Lizenz sind dem Lizenztext zu entnehmen.
+ */
+package de.ozgcloud.aggregation;
+
+import java.util.stream.Stream;
+
+import de.ozgcloud.aggregation.warehouse.DocumentEntry;
+import lombok.Builder;
+
+@Builder
+public record Aggregation(String aggregationName, Stream<DocumentEntry> documentEntries) {
+}

aggregation-manager-job/src/main/java/de/ozgcloud/aggregation/AggregationDataLoader.java

+/*
+ * Copyright (C) 2025 Das Land Schleswig-Holstein vertreten durch den
+ * Ministerpräsidenten des Landes Schleswig-Holstein
+ * Staatskanzlei
+ * Abteilung Digitalisierung und zentrales IT-Management der Landesregierung
+ *
+ * Lizenziert unter der EUPL, Version 1.2 oder - sobald
+ * diese von der Europäischen Kommission genehmigt wurden -
+ * Folgeversionen der EUPL ("Lizenz");
+ * Sie dürfen dieses Werk ausschließlich gemäß
+ * dieser Lizenz nutzen.
+ * Eine Kopie der Lizenz finden Sie hier:
+ *
+ * https://joinup.ec.europa.eu/collection/eupl/eupl-text-eupl-12
+ *
+ * Sofern nicht durch anwendbare Rechtsvorschriften
+ * gefordert oder in schriftlicher Form vereinbart, wird
+ * die unter der Lizenz verbreitete Software "so wie sie
+ * ist", OHNE JEGLICHE GEWÄHRLEISTUNG ODER BEDINGUNGEN -
+ * ausdrücklich oder stillschweigend - verbreitet.
+ * Die sprachspezifischen Genehmigungen und Beschränkungen
+ * unter der Lizenz sind dem Lizenztext zu entnehmen.
+ */
+package de.ozgcloud.aggregation;
+
+public interface AggregationDataLoader {
+
+	void loadIntoTarget(Aggregation aggregation);
+}

aggregation-manager-job/src/main/java/de/ozgcloud/aggregation/AggregationWarehouseDataLoader.java

+/*
+ * Copyright (C) 2025 Das Land Schleswig-Holstein vertreten durch den
+ * Ministerpräsidenten des Landes Schleswig-Holstein
+ * Staatskanzlei
+ * Abteilung Digitalisierung und zentrales IT-Management der Landesregierung
+ *
+ * Lizenziert unter der EUPL, Version 1.2 oder - sobald
+ * diese von der Europäischen Kommission genehmigt wurden -
+ * Folgeversionen der EUPL ("Lizenz");
+ * Sie dürfen dieses Werk ausschließlich gemäß
+ * dieser Lizenz nutzen.
+ * Eine Kopie der Lizenz finden Sie hier:
+ *
+ * https://joinup.ec.europa.eu/collection/eupl/eupl-text-eupl-12
+ *
+ * Sofern nicht durch anwendbare Rechtsvorschriften
+ * gefordert oder in schriftlicher Form vereinbart, wird
+ * die unter der Lizenz verbreitete Software "so wie sie
+ * ist", OHNE JEGLICHE GEWÄHRLEISTUNG ODER BEDINGUNGEN -
+ * ausdrücklich oder stillschweigend - verbreitet.
+ * Die sprachspezifischen Genehmigungen und Beschränkungen
+ * unter der Lizenz sind dem Lizenztext zu entnehmen.
+ */
+package de.ozgcloud.aggregation;
+
+import org.apache.commons.lang3.StringUtils;
+import org.springframework.context.annotation.Scope;
+import org.springframework.stereotype.Component;
+
+import de.ozgcloud.aggregation.warehouse.DocumentEntry;
+import de.ozgcloud.aggregation.warehouse.WarehouseRepository;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.log4j.Log4j2;
+
+@Component
+@Scope("prototype")
+@RequiredArgsConstructor
+@Log4j2
+class AggregationWarehouseDataLoader implements AggregationDataLoader {
+
+	private final WarehouseRepository repository;
+
+	@Override
+	public void loadIntoTarget(Aggregation aggregation) {
+		var collectionName = getCollectionName(aggregation);
+		repository.clearCollection(collectionName);
+		repository.saveAllInCollection(aggregation.documentEntries(), collectionName);
+	}
+
+	String getCollectionName(Aggregation aggregation) {
+		return StringUtils.isNotBlank(aggregation.aggregationName()) ? aggregation.aggregationName() : DocumentEntry.COLLECTION;
+	}
+}

aggregation-manager-job/src/main/java/de/ozgcloud/aggregation/Aggregator.java

@@ -22,7 +22,6 @@ import de.ozgcloud.aggregation.transformation.AggregationMapping.FormIdentifier;
 import de.ozgcloud.aggregation.transformation.TransformationException;
 import de.ozgcloud.aggregation.transformation.VorgangMapper;
 import de.ozgcloud.aggregation.warehouse.DocumentEntry;
-import de.ozgcloud.aggregation.warehouse.WarehouseRepository;
 import de.ozgcloud.apilib.vorgang.OzgCloudVorgang;
 import de.ozgcloud.apilib.vorgang.OzgCloudVorgangQuery;
 import de.ozgcloud.apilib.vorgang.OzgCloudVorgangQuery.FormIdentification;
@@ -42,13 +41,13 @@ class Aggregator {
 	private static final Predicate<Batch> HAS_NEXT_BATCH = x -> !x.items.isEmpty();
 
 	private final OzgCloudVorgangService vorgangService;
-	private final WarehouseRepository repository;
 	private final VorgangMapper vorgangMapper;
 
 	private Execution execution;
 	private FormIdentifier formIdentifier;
-	private String collectionName = DocumentEntry.COLLECTION;
+	private String aggregationName;
 	private int batchSize = 100;
+	private AggregationDataLoader loader;
 
 	public Aggregator withExecution(Execution execution) {
 		this.execution = execution;
@@ -58,7 +57,7 @@ class Aggregator {
 	public Aggregator withAggregationMapping(AggregationMapping aggregationMapping) {
 		if (Objects.nonNull(aggregationMapping)) {
 			this.formIdentifier = aggregationMapping.getFormIdentifier();
-			this.collectionName = aggregationMapping.getName();
+			this.aggregationName = aggregationMapping.getName();
 		}
 		return this;
 	}
@@ -68,8 +67,13 @@ class Aggregator {
 		return this;
 	}
 
+	public Aggregator withLoader(AggregationDataLoader loader) {
+		this.loader = loader;
+		return this;
+	}
+
 	public void aggregate() {
-		loadVorgaengeIntoRepository(Stream.concat(extractBatchesOfVorgaengeFromDataSource(), extractBatchesOfDeletedVorgaengeFromDataSource()));
+		loadVorgaenge(Stream.concat(extractBatchesOfVorgaengeFromDataSource(), extractBatchesOfDeletedVorgaengeFromDataSource()));
 	}
 
 	Stream<Batch> extractBatchesOfVorgaengeFromDataSource() {
@@ -129,13 +133,12 @@ class Aggregator {
 		return UUID.randomUUID();
 	}
 
-	void loadVorgaengeIntoRepository(Stream<Batch> batches) {
-		repository.clearCollection(collectionName);
-		batches.map(this::transformBatchToDocumentEntries).forEach(this::loadDocumentEntriesIntoRepository);
+	void loadVorgaenge(Stream<Batch> batches) {
+		loader.loadIntoTarget(new Aggregation(aggregationName, batches.flatMap(this::transformBatchToDocumentEntries)));
 	}
 
-	List<DocumentEntry> transformBatchToDocumentEntries(Batch batch) {
-		return batch.items.stream().map(vorgang -> transformWithinBatch(batch, vorgang)).filter(Objects::nonNull).toList();
+	Stream<DocumentEntry> transformBatchToDocumentEntries(Batch batch) {
+		return batch.items.stream().map(vorgang -> transformWithinBatch(batch, vorgang)).filter(Objects::nonNull);
 	}
 
 	DocumentEntry transformWithinBatch(Batch batch, OzgCloudVorgang vorgang) {
@@ -149,11 +152,6 @@ class Aggregator {
 		}
 	}
 
-	void loadDocumentEntriesIntoRepository(List<DocumentEntry> entries) {
-		LOG.atDebug().log("store documents: {}", () -> entries.stream().map(DocumentEntry::getId).toList());
-		repository.saveAllInCollection(entries, collectionName);
-	}
-
 	record Batch(Execution execution, UUID id, Page page, List<OzgCloudVorgang> items) {
 	}
 

aggregation-manager-job/src/main/java/de/ozgcloud/aggregation/extern/AggregationDataRemoteService.java

@@ -31,6 +31,7 @@ import java.util.stream.Stream;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.stereotype.Service;
 
+import de.ozgcloud.aggregation.Aggregation;
 import de.ozgcloud.aggregation.AggregationManagerConfiguration;
 import de.ozgcloud.aggregation.AggregationManagerProperties;
 import de.ozgcloud.aggregation.data.AggregationDataServiceGrpc;
@@ -63,8 +64,8 @@ class AggregationDataRemoteService {
 	}
 
 	SendAggregationDataResponseObserver buildSendAggregationDataResponseObserver(Aggregation aggregation) {
-		var requestData = new RequestData(properties.getMandant(), aggregation.aggregationName,
-				toGrpcAggregationDataStream(aggregation.documentEntries).iterator());
+		var requestData = new RequestData(properties.getMandant(), aggregation.aggregationName(),
+				toGrpcAggregationDataStream(aggregation.documentEntries()).iterator());
 		return new SendAggregationDataResponseObserver(configuration.getFetchingBatchSize(), requestData);
 	}
 
@@ -139,10 +140,6 @@ class AggregationDataRemoteService {
 		}
 	}
 
-	@Builder
-	public record Aggregation(String aggregationName, Stream<DocumentEntry> documentEntries) {
-	}
-
 	@Builder
 	record RequestData(String mandant, String aggregationName, Iterator<GrpcAggregationData> aggregationDataIterator) {
 	}

aggregation-manager-job/src/main/java/de/ozgcloud/aggregation/extern/AggregationRemoteDataLoader.java

+/*
+ * Copyright (C) 2025 Das Land Schleswig-Holstein vertreten durch den
+ * Ministerpräsidenten des Landes Schleswig-Holstein
+ * Staatskanzlei
+ * Abteilung Digitalisierung und zentrales IT-Management der Landesregierung
+ *
+ * Lizenziert unter der EUPL, Version 1.2 oder - sobald
+ * diese von der Europäischen Kommission genehmigt wurden -
+ * Folgeversionen der EUPL ("Lizenz");
+ * Sie dürfen dieses Werk ausschließlich gemäß
+ * dieser Lizenz nutzen.
+ * Eine Kopie der Lizenz finden Sie hier:
+ *
+ * https://joinup.ec.europa.eu/collection/eupl/eupl-text-eupl-12
+ *
+ * Sofern nicht durch anwendbare Rechtsvorschriften
+ * gefordert oder in schriftlicher Form vereinbart, wird
+ * die unter der Lizenz verbreitete Software "so wie sie
+ * ist", OHNE JEGLICHE GEWÄHRLEISTUNG ODER BEDINGUNGEN -
+ * ausdrücklich oder stillschweigend - verbreitet.
+ * Die sprachspezifischen Genehmigungen und Beschränkungen
+ * unter der Lizenz sind dem Lizenztext zu entnehmen.
+ */
+package de.ozgcloud.aggregation.extern;
+
+import java.util.concurrent.ExecutionException;
+
+import org.springframework.context.annotation.Scope;
+import org.springframework.stereotype.Component;
+
+import de.ozgcloud.aggregation.Aggregation;
+import de.ozgcloud.aggregation.AggregationDataLoader;
+import de.ozgcloud.common.errorhandling.TechnicalException;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.log4j.Log4j2;
+
+@Component
+@Scope("prototype")
+@RequiredArgsConstructor
+@Log4j2
+class AggregationRemoteDataLoader implements AggregationDataLoader {
+
+	private final AggregationDataRemoteService service;
+
+	@Override
+	public void loadIntoTarget(Aggregation aggregation) {
+		try {
+			service.sendAggregationData(aggregation).get();
+		} catch (InterruptedException e) {
+			Thread.currentThread().interrupt();
+			throw new TechnicalException("Waiting for sending aggregation data to complete was interrupted.", e);
+		} catch (ExecutionException e) {
+			throw new TechnicalException("Error on sending aggregation data.", e);
+		}
+	}
+}

aggregation-manager-job/src/main/java/de/ozgcloud/aggregation/warehouse/CustomWarehouseRepository.java

 package de.ozgcloud.aggregation.warehouse;
 
-import java.util.List;
+import java.util.stream.Stream;
 
 interface CustomWarehouseRepository {
 
 	DocumentEntry saveInCollection(DocumentEntry documentEntry, String collectionName);
 
-	List<DocumentEntry> saveAllInCollection(Iterable<DocumentEntry> documentEntries, String collectionName);
+	void saveAllInCollection(Stream<DocumentEntry> documentEntries, String collectionName);
 
 	void clearCollection(String collectionName);
 }

aggregation-manager-job/src/main/java/de/ozgcloud/aggregation/warehouse/CustomWarehouseRepositoryImpl.java

 package de.ozgcloud.aggregation.warehouse;
 
-import java.util.List;
-import java.util.stream.StreamSupport;
+import java.util.stream.Stream;
 
 import org.springframework.data.mongodb.core.MongoTemplate;
 
@@ -13,8 +12,8 @@ class CustomWarehouseRepositoryImpl implements CustomWarehouseRepository {
 	private final MongoTemplate mongoTemplate;
 
 	@Override
-	public List<DocumentEntry> saveAllInCollection(Iterable<DocumentEntry> documentEntries, String collectionName) {
-		return StreamSupport.stream(documentEntries.spliterator(), false).map(entry -> saveInCollection(entry, collectionName)).toList();
+	public void saveAllInCollection(Stream<DocumentEntry> documentEntries, String collectionName) {
+		documentEntries.forEach(entry -> saveInCollection(entry, collectionName));
 	}
 
 	@Override

aggregation-manager-job/src/test/java/de/ozgcloud/aggregation/extern/AggregationTestFactory.java → aggregation-manager-job/src/test/java/de/ozgcloud/aggregation/AggregationTestFactory.java

@@ -21,7 +21,7 @@
  * Die sprachspezifischen Genehmigungen und Beschränkungen
  * unter der Lizenz sind dem Lizenztext zu entnehmen.
  */
-package de.ozgcloud.aggregation.extern;
+package de.ozgcloud.aggregation;
 
 import java.util.stream.Stream;
 
@@ -35,12 +35,12 @@ public class AggregationTestFactory {
 	public static final String AGGREGATION_NAME = LoremIpsum.getInstance().getWords(1);
 	public static final DocumentEntry DOCUMENT_ENTRY = DocumentEntryTestFactory.create();
 
-	public static AggregationDataRemoteService.Aggregation create() {
+	public static Aggregation create() {
 		return createBuilder().build();
 	}
 
-	public static AggregationDataRemoteService.Aggregation.AggregationBuilder createBuilder() {
-		return AggregationDataRemoteService.Aggregation.builder()
+	public static Aggregation.AggregationBuilder createBuilder() {
+		return Aggregation.builder()
 				.aggregationName(AGGREGATION_NAME)
 				.documentEntries(Stream.of(DOCUMENT_ENTRY));
 	}

aggregation-manager-job/src/test/java/de/ozgcloud/aggregation/AggregationWarehouseDataLoaderTest.java

+/*
+ * Copyright (C) 2025 Das Land Schleswig-Holstein vertreten durch den
+ * Ministerpräsidenten des Landes Schleswig-Holstein
+ * Staatskanzlei
+ * Abteilung Digitalisierung und zentrales IT-Management der Landesregierung
+ *
+ * Lizenziert unter der EUPL, Version 1.2 oder - sobald
+ * diese von der Europäischen Kommission genehmigt wurden -
+ * Folgeversionen der EUPL ("Lizenz");
+ * Sie dürfen dieses Werk ausschließlich gemäß
+ * dieser Lizenz nutzen.
+ * Eine Kopie der Lizenz finden Sie hier:
+ *
+ * https://joinup.ec.europa.eu/collection/eupl/eupl-text-eupl-12
+ *
+ * Sofern nicht durch anwendbare Rechtsvorschriften
+ * gefordert oder in schriftlicher Form vereinbart, wird
+ * die unter der Lizenz verbreitete Software "so wie sie
+ * ist", OHNE JEGLICHE GEWÄHRLEISTUNG ODER BEDINGUNGEN -
+ * ausdrücklich oder stillschweigend - verbreitet.
+ * Die sprachspezifischen Genehmigungen und Beschränkungen
+ * unter der Lizenz sind dem Lizenztext zu entnehmen.
+ */
+package de.ozgcloud.aggregation;
+
+import static org.assertj.core.api.Assertions.*;
+import static org.mockito.Mockito.*;
+
+import java.util.stream.Stream;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Nested;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.NullAndEmptySource;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Captor;
+import org.mockito.InjectMocks;
+import org.mockito.Mock;
+import org.mockito.Spy;
+
+import com.thedeanda.lorem.LoremIpsum;
+
+import de.ozgcloud.aggregation.warehouse.DocumentEntry;
+import de.ozgcloud.aggregation.warehouse.WarehouseRepository;
+
+class AggregationWarehouseDataLoaderTest {
+
+	@Mock
+	private WarehouseRepository repository;
+	@Spy
+	@InjectMocks
+	private AggregationWarehouseDataLoader loader;
+
+	@Nested
+	class TestLoadIntoTarget {
+
+		private static final String COLLECTION_NAME = LoremIpsum.getInstance().getWords(1);
+		private final Aggregation aggregation = AggregationTestFactory.create();
+
+		@Captor
+		private ArgumentCaptor<Stream<DocumentEntry>> documentEntriesCaptor;
+
+		@BeforeEach
+		void init() {
+			doReturn(COLLECTION_NAME).when(loader).getCollectionName(aggregation);
+		}
+
+		@Test
+		void shouldGetCollectionName() {
+			loader.loadIntoTarget(aggregation);
+
+			verify(loader).getCollectionName(aggregation);
+		}
+
+		@Test
+		void shouldClearCollection() {
+			loader.loadIntoTarget(aggregation);
+
+			verify(repository).clearCollection(COLLECTION_NAME);
+		}
+
+		@Test
+		void shouldSaveDocumentEntriesInCollection() {
+			loader.loadIntoTarget(aggregation);
+
+			verify(repository).saveAllInCollection(documentEntriesCaptor.capture(), eq(COLLECTION_NAME));
+			assertThat(documentEntriesCaptor.getValue()).containsExactly(AggregationTestFactory.DOCUMENT_ENTRY);
+		}
+	}
+
+	@Nested
+	class TestGetCollectionName {
+
+		@Test
+		void shouldReturnAggregationName() {
+			var collectionName = loader.getCollectionName(AggregationTestFactory.create());
+
+			assertThat(collectionName).isEqualTo(AggregationTestFactory.AGGREGATION_NAME);
+		}
+
+		@ParameterizedTest
+		@NullAndEmptySource
+		void shouldReturnCollectionNameFromDocumentEntry(String aggregationName) {
+			var collectionName = loader.getCollectionName(AggregationTestFactory.createBuilder().aggregationName(aggregationName).build());
+
+			assertThat(collectionName).isEqualTo(DocumentEntry.COLLECTION);
+		}
+	}
+}

aggregation-manager-job/src/test/java/de/ozgcloud/aggregation/AggregatorTest.java

@@ -82,7 +82,7 @@ class AggregatorTest {
 
 			private final AggregationMapping aggregationMapping = AggregationMappingTestFactory.create();
 			private final FormIdentifier formIdentifier = AggregationMappingTestFactory.FORM_IDENTIFIER;
-			private final String collectionName = AggregationMappingTestFactory.NAME;
+			private final String aggregationName = AggregationMappingTestFactory.NAME;
 
 			@Test
 			void shouldReturnSelf() {
@@ -99,10 +99,10 @@ class AggregatorTest {
 			}
 
 			@Test
-			void shouldSetCollectionName() {
+			void shouldSetAggregationName() {
 				var result = aggregator.withAggregationMapping(aggregationMapping);
 
-				assertThat(ReflectionTestUtils.getField(result, "collectionName")).isEqualTo(collectionName);
+				assertThat(ReflectionTestUtils.getField(result, "aggregationName")).isEqualTo(aggregationName);
 			}
 		}
 
@@ -124,10 +124,10 @@ class AggregatorTest {
 			}
 
 			@Test
-			void shouldNotSetCollectionName() {
+			void shouldNotSetAggregationName() {
 				var result = aggregator.withAggregationMapping(null);
 
-				assertThat(ReflectionTestUtils.getField(result, "collectionName")).isEqualTo(DocumentEntry.COLLECTION);
+				assertThat(ReflectionTestUtils.getField(result, "aggregationName")).isNull();
 			}
 		}
 	}
@@ -152,6 +152,27 @@ class AggregatorTest {
 		}
 	}
 
+	@Nested
+	class TestWithLoader {
+
+		@Mock
+		private AggregationDataLoader loader;
+
+		@Test
+		void shouldReturnSelf() {
+			var result = aggregator.withLoader(loader);
+
+			assertThat(result).isSameAs(aggregator);
+		}
+
+		@Test
+		void shouldSetLoader() {
+			aggregator.withLoader(loader);
+
+			assertThat(ReflectionTestUtils.getField(aggregator, "loader")).isSameAs(loader);
+		}
+	}
+
 	@Nested
 	class TestAggregate {
 		@Mock
@@ -165,7 +186,7 @@ class AggregatorTest {
 		void setUp() {
 			doReturn(Stream.of(batchOfVorgaenge)).when(aggregator).extractBatchesOfVorgaengeFromDataSource();
 			doReturn(Stream.of(batchOfDeletedVorgaenge)).when(aggregator).extractBatchesOfDeletedVorgaengeFromDataSource();
-			doNothing().when(aggregator).loadVorgaengeIntoRepository(any());
+			doNothing().when(aggregator).loadVorgaenge(any());
 		}
 
 		@Test
@@ -186,7 +207,7 @@ class AggregatorTest {
 		void shouldLoadVorgaengeIntoRepository() {
 			aggregator.aggregate();
 
-			verify(aggregator).loadVorgaengeIntoRepository(batchStreamCaptor.capture());
+			verify(aggregator).loadVorgaenge(batchStreamCaptor.capture());
 			assertThat(batchStreamCaptor.getValue()).containsExactly(batchOfVorgaenge, batchOfDeletedVorgaenge);
 		}
 	}
@@ -485,67 +506,55 @@ class AggregatorTest {
 	}
 
 	@Nested
-	class TestLoadVorgaengeIntoRepository {
+	class TestLoadVorgaenge {
 
 		@Mock
 		private Execution execution;
 		@Mock
 		private Batch batch;
+		@Mock
+		private AggregationDataLoader loader;
 		private final DocumentEntry documentEntry = DocumentEntryTestFactory.create();
 		private final AggregationMapping aggregationMapping = AggregationMappingTestFactory.create();
 		@Captor
-		private ArgumentCaptor<List<DocumentEntry>> documentEntriesCaptor;
+		private ArgumentCaptor<Aggregation> aggregationCaptor;
 
 		@BeforeEach
 		void init() {
-			aggregator = aggregator.withAggregationMapping(aggregationMapping);
-			doReturn(List.of(documentEntry)).when(aggregator).transformBatchToDocumentEntries(any());
-			doNothing().when(aggregator).loadDocumentEntriesIntoRepository(any());
+			aggregator = aggregator.withAggregationMapping(aggregationMapping).withLoader(loader);
 		}
 
 		@Test
-		void shouldDropCollection() {
-			loadVorgaengeIntoRepository();
+		void shouldLoadIntoTarget() {
+			loadVorgaenge();
 
-			verify(repository).clearCollection(AggregationMappingTestFactory.NAME);
+			verify(loader).loadIntoTarget(any(Aggregation.class));
 		}
 
 		@Test
 		void shouldTransform() {
-			loadVorgaengeIntoRepository();
+			doReturn(Stream.of(documentEntry)).when(aggregator).transformBatchToDocumentEntries(any());
 
+			loadVorgaenge();
+
+			getArgumentOfLoadIntoTarget().documentEntries().toList();
 			verify(aggregator).transformBatchToDocumentEntries(batch);
 		}
 
 		@Test
-		void shouldLoadIntoRepository() {
-			loadVorgaengeIntoRepository();
+		void shouldSetAggregationName() {
+			loadVorgaenge();
 
-			verify(aggregator).loadDocumentEntriesIntoRepository(documentEntriesCaptor.capture());
-			assertThat(documentEntriesCaptor.getValue()).containsExactly(documentEntry);
+			assertThat(getArgumentOfLoadIntoTarget().aggregationName()).isEqualTo(AggregationMappingTestFactory.NAME);
 		}
 
-		private void loadVorgaengeIntoRepository() {
-			aggregator.loadVorgaengeIntoRepository(Stream.of(batch));
-		}
+		private Aggregation getArgumentOfLoadIntoTarget() {
+			verify(loader).loadIntoTarget(aggregationCaptor.capture());
+			return aggregationCaptor.getValue();
 		}
 
-	@Nested
-	class TestLoadDocumentEntriesIntoRepository {
-
-		private final List<DocumentEntry> documentEntries = List.of(DocumentEntryTestFactory.create());
-		private final AggregationMapping aggregationMapping = AggregationMappingTestFactory.create();
-
-		@BeforeEach
-		void init() {
-			aggregator = aggregator.withAggregationMapping(aggregationMapping);
-		}
-
-		@Test
-		void shouldSaveDocumentEntriesInCollection() {
-			aggregator.loadDocumentEntriesIntoRepository(documentEntries);
-
-			verify(repository).saveAllInCollection(documentEntries, AggregationMappingTestFactory.NAME);
+		private void loadVorgaenge() {
+			aggregator.loadVorgaenge(Stream.of(batch));
 		}
 	}
 }

aggregation-manager-job/src/test/java/de/ozgcloud/aggregation/extern/AggregationDataRemoteServiceTest.java

@@ -45,8 +45,10 @@ import org.mockito.Spy;
 
 import com.thedeanda.lorem.LoremIpsum;
 
+import de.ozgcloud.aggregation.Aggregation;
 import de.ozgcloud.aggregation.AggregationManagerConfiguration;
 import de.ozgcloud.aggregation.AggregationManagerProperties;
+import de.ozgcloud.aggregation.AggregationTestFactory;
 import de.ozgcloud.aggregation.data.AggregationDataServiceGrpc;
 import de.ozgcloud.aggregation.data.GrpcAggregationData;
 import de.ozgcloud.aggregation.data.GrpcSendAggregationDataRequest;
@@ -73,7 +75,7 @@ class AggregationDataRemoteServiceTest {
 	@Nested
 	class TestSendAggregationData {
 
-		private final AggregationDataRemoteService.Aggregation aggregation = AggregationTestFactory.create();
+		private final Aggregation aggregation = AggregationTestFactory.create();
 
 		@Mock
 		private SendAggregationDataResponseObserver responseObserver;

aggregation-manager-job/src/test/java/de/ozgcloud/aggregation/extern/AggregationRemoteDataLoaderTest.java

+/*
+ * Copyright (C) 2025 Das Land Schleswig-Holstein vertreten durch den
+ * Ministerpräsidenten des Landes Schleswig-Holstein
+ * Staatskanzlei
+ * Abteilung Digitalisierung und zentrales IT-Management der Landesregierung
+ *
+ * Lizenziert unter der EUPL, Version 1.2 oder - sobald
+ * diese von der Europäischen Kommission genehmigt wurden -
+ * Folgeversionen der EUPL ("Lizenz");
+ * Sie dürfen dieses Werk ausschließlich gemäß
+ * dieser Lizenz nutzen.
+ * Eine Kopie der Lizenz finden Sie hier:
+ *
+ * https://joinup.ec.europa.eu/collection/eupl/eupl-text-eupl-12
+ *
+ * Sofern nicht durch anwendbare Rechtsvorschriften
+ * gefordert oder in schriftlicher Form vereinbart, wird
+ * die unter der Lizenz verbreitete Software "so wie sie
+ * ist", OHNE JEGLICHE GEWÄHRLEISTUNG ODER BEDINGUNGEN -
+ * ausdrücklich oder stillschweigend - verbreitet.
+ * Die sprachspezifischen Genehmigungen und Beschränkungen
+ * unter der Lizenz sind dem Lizenztext zu entnehmen.
+ */
+package de.ozgcloud.aggregation.extern;
+
+import static org.assertj.core.api.Assertions.*;
+import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.ArgumentMatchers.*;
+import static org.mockito.Mockito.*;
+
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Nested;
+import org.junit.jupiter.api.Test;
+import org.mockito.InjectMocks;
+import org.mockito.Mock;
+import org.mockito.Spy;
+
+import de.ozgcloud.aggregation.Aggregation;
+import de.ozgcloud.aggregation.AggregationTestFactory;
+import de.ozgcloud.common.errorhandling.TechnicalException;
+import lombok.SneakyThrows;
+
+class AggregationRemoteDataLoaderTest {
+
+	@Mock
+	private AggregationDataRemoteService remoteService;
+	@Spy
+	@InjectMocks
+	private AggregationRemoteDataLoader loader;
+
+	@Nested
+	class TestLoadIntoTarget {
+
+		private final Aggregation aggregation = AggregationTestFactory.create();
+		@Mock
+		private Future<Void> future;
+
+		@BeforeEach
+		void init() {
+			when(remoteService.sendAggregationData(any())).thenReturn(future);
+		}
+
+		@Test
+		@SneakyThrows
+		void shouldSendAggregationData() {
+			loader.loadIntoTarget(aggregation);
+
+			verify(remoteService).sendAggregationData(aggregation);
+		}
+
+		@Test
+		@SneakyThrows
+		void shouldGetFromFuture() {
+			loader.loadIntoTarget(aggregation);
+
+			verify(future).get();
+		}
+
+		@Nested
+		class TestOnInterruptedException {
+
+			private final InterruptedException exception = new InterruptedException();
+
+			@BeforeEach
+			@SneakyThrows
+			void mock() {
+				when(future.get()).thenThrow(exception);
+			}
+
+			@Test
+			void shouldThrowTechnicalException() {
+				assertThrows(TechnicalException.class, () -> loader.loadIntoTarget(aggregation));
+			}
+
+			@Test
+			void shouldInterruptThread() {
+				try {
+					loader.loadIntoTarget(aggregation);
+				} catch (TechnicalException e) {
+					// expected
+				}
+
+				assertThat(Thread.currentThread().isInterrupted()).isTrue();
+			}
+		}
+
+		@Nested
+		class TestOnExecutionException {
+
+			private final ExecutionException exception = new ExecutionException(new Exception());
+
+			@BeforeEach
+			@SneakyThrows
+			void mock() {
+				when(future.get()).thenThrow(exception);
+			}
+
+			@Test
+			void shouldThrowTechnicalException() {
+				assertThrows(TechnicalException.class, () -> loader.loadIntoTarget(aggregation));
+			}
+		}
+	}
+}

aggregation-manager-job/src/test/java/de/ozgcloud/aggregation/warehouse/CustomWarehouseRepositoryImplTest.java

 package de.ozgcloud.aggregation.warehouse;
 
 import static org.assertj.core.api.Assertions.*;
-import static org.mockito.ArgumentMatchers.*;
 import static org.mockito.Mockito.*;
 
 import java.util.List;
@@ -33,20 +32,10 @@ class CustomWarehouseRepositoryImplTest {
 
 		@Test
 		void shouldCallSaveInCollection() {
-			repository.saveAllInCollection(documentEntries, collectionName);
+			repository.saveAllInCollection(documentEntries.stream(), collectionName);
 
 			verify(repository).saveInCollection(documentEntry, collectionName);
 		}
-
-		@Test
-		void shouldReturnSavedDocumentEntry() {
-			var savedDocumentEntry = DocumentEntryTestFactory.create();
-			doReturn(savedDocumentEntry).when(repository).saveInCollection(any(), any());
-
-			var returnedDocumentEntries = repository.saveAllInCollection(documentEntries, collectionName);
-
-			assertThat(returnedDocumentEntries).containsExactly(savedDocumentEntry);
-		}
 	}
 
 	@Nested