Revert "MODSOURMAN-1022: migrate from DI_EDIFACT_RECORD_CREATED to DI…

…_INCOMING_EDIFACT_RECORD_PARSED event (#449)" (#451)

This reverts commit 7fc897a.

NEWS.md

@@ -1,5 +1,4 @@
 ## 5.8.0 - Unreleased
-* [MODSOURMAN-1022](https://issues.folio.org/browse/MODSOURMAN-1022) Remove step of initial saving of incoming records to SRS
 
 ## 5.7.0 - Released Poppy R2 2023
 The focus of this release was to implement Vertx SFTP client for Batch Voucher Export and implement pay against previous fiscal years

pom.xml

@@ -202,7 +202,7 @@
     <dependency>
       <groupId>org.folio</groupId>
       <artifactId>data-import-processing-core</artifactId>
-      <version>4.2.0-SNAPSHOT</version>
+      <version>4.1.0</version>
     </dependency>
     <dependency>
       <groupId>org.folio</groupId>

src/main/java/org/folio/verticles/DataImportConsumerVerticle.java

@@ -1,6 +1,6 @@
 package org.folio.verticles;
 
-import static org.folio.DataImportEventTypes.DI_INCOMING_EDIFACT_RECORD_PARSED;
+import static org.folio.DataImportEventTypes.DI_EDIFACT_RECORD_CREATED;
 
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
@@ -55,7 +55,7 @@ public void start(Promise<Void> startPromise) {
     EventManager.registerKafkaEventPublisher(kafkaConfig, vertx, maxDistributionNumber);
 
     SubscriptionDefinition subscriptionDefinition = KafkaTopicNameHelper.createSubscriptionDefinition(kafkaConfig.getEnvId(),
-      KafkaTopicNameHelper.getDefaultNameSpace(), DI_INCOMING_EDIFACT_RECORD_PARSED.value());
+      KafkaTopicNameHelper.getDefaultNameSpace(), DI_EDIFACT_RECORD_CREATED.value());
 
     consumerWrapper = KafkaConsumerWrapper.<String, String>builder()
       .context(context)

src/test/java/org/folio/dataimport/handlers/actions/CreateInvoiceEventHandlerTest.java

@@ -6,7 +6,7 @@
 import static org.folio.ApiTestSuite.KAFKA_ENV_VALUE;
 import static org.folio.ApiTestSuite.kafkaCluster;
 import static org.folio.DataImportEventTypes.DI_COMPLETED;
-import static org.folio.DataImportEventTypes.DI_INCOMING_EDIFACT_RECORD_PARSED;
+import static org.folio.DataImportEventTypes.DI_EDIFACT_RECORD_CREATED;
 import static org.folio.DataImportEventTypes.DI_ERROR;
 import static org.folio.DataImportEventTypes.DI_INVOICE_CREATED;
 import static org.folio.dataimport.handlers.actions.CreateInvoiceEventHandler.INVOICE_LINES_ERRORS_KEY;
@@ -264,7 +264,7 @@ public void shouldCreateInvoiceAndPublishDiCompletedEvent() throws InterruptedEx
     payloadContext.put(JOB_PROFILE_SNAPSHOT_ID_KEY, profileSnapshotWrapper.getId());
 
     DataImportEventPayload dataImportEventPayload = new DataImportEventPayload()
-      .withEventType(DI_INCOMING_EDIFACT_RECORD_PARSED.value())
+      .withEventType(DI_EDIFACT_RECORD_CREATED.value())
       .withTenant(DI_POST_INVOICE_LINES_SUCCESS_TENANT)
       .withOkapiUrl(OKAPI_URL)
       .withToken(TOKEN)
@@ -339,7 +339,7 @@ public void shouldMatchPoLinesByPoLineNumberAndCreateInvoiceLinesWithDescription
     payloadContext.put(DATA_IMPORT_PAYLOAD_OKAPI_USER_ID, USER_ID);
 
     DataImportEventPayload dataImportEventPayload = new DataImportEventPayload()
-      .withEventType(DI_INCOMING_EDIFACT_RECORD_PARSED.value())
+      .withEventType(DI_EDIFACT_RECORD_CREATED.value())
       .withTenant(DI_POST_INVOICE_LINES_SUCCESS_TENANT)
       .withOkapiUrl(OKAPI_URL)
       .withToken(TOKEN)
@@ -406,7 +406,7 @@ public void shouldMatchPoLinesByRefNumberAndCreateInvoiceLinesWithDescriptionFro
     payloadContext.put(JOB_PROFILE_SNAPSHOT_ID_KEY, profileSnapshotWrapper.getId());
 
     DataImportEventPayload dataImportEventPayload = new DataImportEventPayload()
-      .withEventType(DI_INCOMING_EDIFACT_RECORD_PARSED.value())
+      .withEventType(DI_EDIFACT_RECORD_CREATED.value())
       .withTenant(DI_POST_INVOICE_LINES_SUCCESS_TENANT)
       .withOkapiUrl(OKAPI_URL)
       .withToken(TOKEN)
@@ -481,7 +481,7 @@ public void shouldMatchPoLinesByPoLineNumberAndCreateInvoiceLinesWithPoLinesFund
     payloadContext.put(JOB_PROFILE_SNAPSHOT_ID_KEY, profileSnapshotWrapper.getId());
 
     DataImportEventPayload dataImportEventPayload = new DataImportEventPayload()
-      .withEventType(DI_INCOMING_EDIFACT_RECORD_PARSED.value())
+      .withEventType(DI_EDIFACT_RECORD_CREATED.value())
       .withTenant(DI_POST_INVOICE_LINES_SUCCESS_TENANT)
       .withOkapiUrl(OKAPI_URL)
       .withToken(TOKEN)
@@ -548,7 +548,7 @@ public void shouldNotLinkInvoiceLinesToPoLinesWhenMultiplePoLinesAreMatchedByRef
     payloadContext.put(JOB_PROFILE_SNAPSHOT_ID_KEY, profileSnapshotWrapper.getId());
 
     DataImportEventPayload dataImportEventPayload = new DataImportEventPayload()
-      .withEventType(DI_INCOMING_EDIFACT_RECORD_PARSED.value())
+      .withEventType(DI_EDIFACT_RECORD_CREATED.value())
       .withTenant(DI_POST_INVOICE_LINES_SUCCESS_TENANT)
       .withOkapiUrl(OKAPI_URL)
       .withToken(TOKEN)
@@ -608,7 +608,7 @@ public void shouldMatchPoLineByPoLineNumberAndLeaveEmptyInvoiceLineFundDistribut
     payloadContext.put(JOB_PROFILE_SNAPSHOT_ID_KEY, profileSnapshotWrapper.getId());
 
     DataImportEventPayload dataImportEventPayload = new DataImportEventPayload()
-      .withEventType(DI_INCOMING_EDIFACT_RECORD_PARSED.value())
+      .withEventType(DI_EDIFACT_RECORD_CREATED.value())
       .withTenant(DI_POST_INVOICE_LINES_SUCCESS_TENANT)
       .withOkapiUrl(OKAPI_URL)
       .withToken(TOKEN)
@@ -664,7 +664,7 @@ public void shouldPublishDiErrorEventWhenHasNoSourceRecord() throws InterruptedE
     addMockEntry(JOB_PROFILE_SNAPSHOTS_MOCK, profileSnapshotWrapper);
 
     DataImportEventPayload dataImportEventPayload = new DataImportEventPayload()
-      .withEventType(DI_INCOMING_EDIFACT_RECORD_PARSED.value())
+      .withEventType(DI_EDIFACT_RECORD_CREATED.value())
       .withTenant(TENANT_ID)
       .withOkapiUrl(OKAPI_URL)
       .withToken(TOKEN)
@@ -709,7 +709,7 @@ public void shouldPublishDiErrorEventWhenPostInvoiceToStorageFailed() throws Int
     payloadContext.put(JOB_PROFILE_SNAPSHOT_ID_KEY, profileSnapshotWrapper.getId());
 
     DataImportEventPayload dataImportEventPayload = new DataImportEventPayload()
-      .withEventType(DI_INCOMING_EDIFACT_RECORD_PARSED.value())
+      .withEventType(DI_EDIFACT_RECORD_CREATED.value())
       .withTenant(ERROR_TENANT)
       .withOkapiUrl(OKAPI_URL)
       .withToken(TOKEN)
@@ -762,7 +762,7 @@ public void shouldPublishDiErrorWithInvoiceLineErrorWhenOneOfInvoiceLinesCreatio
     payloadContext.put(JOB_PROFILE_SNAPSHOT_ID_KEY, profileSnapshotWrapper.getId());
 
     DataImportEventPayload dataImportEventPayload = new DataImportEventPayload()
-      .withEventType(DI_INCOMING_EDIFACT_RECORD_PARSED.value())
+      .withEventType(DI_EDIFACT_RECORD_CREATED.value())
       .withTenant(DI_POST_INVOICE_LINES_SUCCESS_TENANT)
       .withOkapiUrl(OKAPI_URL)
       .withToken(TOKEN)
@@ -820,7 +820,7 @@ public void shouldPublishDiErrorWhenMappingProfileHasInvalidMappingSyntax() thro
     payloadContext.put(JOB_PROFILE_SNAPSHOT_ID_KEY, profileSnapshotWrapper.getId());
 
     DataImportEventPayload dataImportEventPayload = new DataImportEventPayload()
-      .withEventType(DI_INCOMING_EDIFACT_RECORD_PARSED.value())
+      .withEventType(DI_EDIFACT_RECORD_CREATED.value())
       .withTenant(DI_POST_INVOICE_LINES_SUCCESS_TENANT)
       .withOkapiUrl(OKAPI_URL)
       .withToken(TOKEN)
@@ -854,7 +854,7 @@ public void shouldReturnTrueWhenHandlerIsEligibleForActionProfile() {
     // given
     ProfileSnapshotWrapper profileSnapshotWrapper = buildProfileSnapshotWrapper(jobProfile, actionProfile, mappingProfile);
     DataImportEventPayload dataImportEventPayload = new DataImportEventPayload()
-      .withEventType(DI_INCOMING_EDIFACT_RECORD_PARSED.value())
+      .withEventType(DI_EDIFACT_RECORD_CREATED.value())
       .withCurrentNode(profileSnapshotWrapper.getChildSnapshotWrappers().get(0));
 
     // when
@@ -880,7 +880,7 @@ public void shouldReturnFalseWhenHandlerIsNotEligibleForActionProfile() {
       .withContent(actionProfile);
 
     DataImportEventPayload dataImportEventPayload = new DataImportEventPayload()
-      .withEventType(DI_INCOMING_EDIFACT_RECORD_PARSED.value())
+      .withEventType(DI_EDIFACT_RECORD_CREATED.value())
       .withCurrentNode(profileSnapshotWrapper);
 
     // when

src/test/java/org/folio/verticles/DataImportConsumerVerticleTest.java

@@ -5,7 +5,7 @@
 import static org.folio.ApiTestSuite.KAFKA_ENV_VALUE;
 import static org.folio.ApiTestSuite.kafkaCluster;
 import static org.folio.DataImportEventTypes.DI_COMPLETED;
-import static org.folio.DataImportEventTypes.DI_INCOMING_EDIFACT_RECORD_PARSED;
+import static org.folio.DataImportEventTypes.DI_EDIFACT_RECORD_CREATED;
 import static org.folio.DataImportEventTypes.DI_ERROR;
 import static org.folio.dataimport.handlers.events.DataImportKafkaHandler.JOB_PROFILE_SNAPSHOT_ID_KEY;
 import static org.folio.kafka.KafkaTopicNameHelper.getDefaultNameSpace;
@@ -21,6 +21,7 @@
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
+import java.io.UnsupportedEncodingException;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
@@ -95,7 +96,7 @@ public void setUp() {
   }
 
   @Test
-  public void shouldPublishDiCompletedEventWhenProcessingCoreHandlerSucceeded() throws InterruptedException {
+  public void shouldPublishDiCompletedEventWhenProcessingCoreHandlerSucceeded() throws InterruptedException, UnsupportedEncodingException {
     // given
     EventHandler mockedEventHandler = mock(EventHandler.class);
     when(mockedEventHandler.isEligible(any(DataImportEventPayload.class))).thenReturn(true);
@@ -104,7 +105,7 @@ public void shouldPublishDiCompletedEventWhenProcessingCoreHandlerSucceeded() th
     EventManager.registerEventHandler(mockedEventHandler);
 
     DataImportEventPayload dataImportEventPayload = new DataImportEventPayload()
-      .withEventType(DI_INCOMING_EDIFACT_RECORD_PARSED.value())
+      .withEventType(DI_EDIFACT_RECORD_CREATED.value())
       .withTenant(TENANT_ID)
       .withOkapiUrl(OKAPI_URL)
       .withToken(TOKEN)
@@ -142,7 +143,7 @@ public void shouldPublishDiErrorEventWhenProcessingCoreHandlerFailed() throws In
     EventManager.registerEventHandler(mockedEventHandler);
 
     DataImportEventPayload dataImportEventPayload = new DataImportEventPayload()
-      .withEventType(DI_INCOMING_EDIFACT_RECORD_PARSED.value())
+      .withEventType(DI_EDIFACT_RECORD_CREATED.value())
       .withTenant(TENANT_ID)
       .withOkapiUrl(OKAPI_URL)
       .withToken(TOKEN)