Compare commits
2 Commits
30170d85ac
...
a74b220ed7
| Author | SHA1 | Date | |
|---|---|---|---|
| a74b220ed7 | |||
| 99f492d5eb |
@@ -12,7 +12,10 @@
|
||||
"WebFetch(domain:medium.com)",
|
||||
"WebFetch(domain:quarkus.io)",
|
||||
"WebFetch(domain:github.com)",
|
||||
"WebFetch(domain:walidhajeri.hashnode.dev)"
|
||||
"WebFetch(domain:walidhajeri.hashnode.dev)",
|
||||
"Bash(find \"C:\\\\\\\\src\\\\\\\\Galabau\\\\\\\\glb-spielwiese\\\\\\\\automaton\" -name \"FileProcessingPipeline.java\")",
|
||||
"PowerShell(Get-ChildItem -Path \"C:\\\\src\\\\Galabau\\\\glb-spielwiese\\\\quarkus-automaton\\\\src\\\\main\\\\java\" -Recurse | Where-Object { !$_.PSIsContainer } | Select-Object FullName)",
|
||||
"PowerShell(cmd /c \"dir /s /b C:\\\\src\\\\Galabau\\\\glb-spielwiese\\\\quarkus-automaton\\\\src\\\\main\\\\java\")"
|
||||
]
|
||||
}
|
||||
}
|
||||
|
||||
@@ -55,7 +55,7 @@ create or replace package body pck_auto_import as
|
||||
pck_log.p_info(
|
||||
i_module => c_log_module
|
||||
,i_action => l_log_action
|
||||
,i_message => 'Verarbeitung offener OCI-Batches gestartet'
|
||||
,i_message => 'Verarbeitung offener BA-Korrespondenz-Dateien aus OCI gestartet'
|
||||
);
|
||||
|
||||
p_process_incoming_ba_data;
|
||||
@@ -63,7 +63,7 @@ create or replace package body pck_auto_import as
|
||||
pck_log.p_info(
|
||||
i_module => c_log_module
|
||||
,i_action => l_log_action
|
||||
,i_message => 'Verarbeitung offener OCI-Batches abgeschlossen'
|
||||
,i_message => 'Verarbeitung offener BA-Korrespondenz-Dateien aus OCI abgeschlossen'
|
||||
);
|
||||
|
||||
-- Quarkus anstoßen — Fehler werden geloggt, nicht eskaliert
|
||||
@@ -264,7 +264,7 @@ create or replace package body pck_auto_import as
|
||||
pck_log.p_info(
|
||||
i_module => c_log_module
|
||||
,i_action => l_log_action
|
||||
,i_message => 'Verarbeitung von eingehen BA Korrespondenzen gestartet'
|
||||
,i_message => 'BA-KORRESPONDENZEN IMPORT-START: Verarbeitung von eingehen BA Korrespondenzen gestartet'
|
||||
);
|
||||
|
||||
-- Zielordner Name zusammenstellen
|
||||
@@ -314,7 +314,7 @@ create or replace package body pck_auto_import as
|
||||
pck_log.p_info(
|
||||
i_module => c_log_module
|
||||
,i_action => l_log_action
|
||||
,i_message => 'Kein DB-Verarbeitungsmarker vorhanden — Batch wird übersprungen (Upload noch nicht abgeschlossen)'
|
||||
,i_message => 'Kein DB-Verarbeitungsmarker vorhanden — Entpackter ZIP-Ordner wird übersprungen (Upload noch nicht abgeschlossen)'
|
||||
,i_object_ref => rec_folder.object_key
|
||||
);
|
||||
continue;
|
||||
@@ -333,7 +333,7 @@ create or replace package body pck_auto_import as
|
||||
pck_log.p_info(
|
||||
i_module => c_log_module
|
||||
,i_action => l_log_action
|
||||
,i_message => 'Batch-Verarbeitung gestartet — Zielordner: "' || l_target_folder || '"'
|
||||
,i_message => 'ZIP-START: Verarbeitung von entpacktem ZIP-Ordner gestartet — Zielordner: "' || l_target_folder || '"'
|
||||
,i_object_ref => rec_folder.object_key
|
||||
);
|
||||
|
||||
@@ -415,7 +415,7 @@ create or replace package body pck_auto_import as
|
||||
pck_log.p_warn(
|
||||
i_module => c_log_module
|
||||
,i_action => l_log_action
|
||||
,i_message => 'Batch mit Fehlern abgeschlossen — mind. eine Datei konnte nicht importiert werden, SB-Marker wird hochgeladen...'
|
||||
,i_message => 'ZIP-ENDE: Entpackter ZIP-Ordner mit Fehlern abgeschlossen — mind. eine Datei konnte nicht importiert werden, SB-Marker wird hochgeladen...'
|
||||
,i_object_ref => rec_folder.object_key
|
||||
);
|
||||
|
||||
@@ -431,11 +431,19 @@ create or replace package body pck_auto_import as
|
||||
pck_log.p_info(
|
||||
i_module => c_log_module
|
||||
,i_action => l_log_action
|
||||
,i_message => 'Batch abgeschlossen, alle Dateien erfolgreich importiert'
|
||||
,i_message => 'ZIP-ENDE: Entpackter ZIP-Ordner abgeschlossen, alle Dateien erfolgreich importiert'
|
||||
,i_object_ref => rec_folder.object_key
|
||||
);
|
||||
end if;
|
||||
end loop;
|
||||
|
||||
|
||||
pck_log.p_info(
|
||||
i_module => c_log_module
|
||||
,i_action => l_log_action
|
||||
,i_message => 'BA-KORRESPONDENZEN IMPORT-ENDE: Verarbeitung von eingehen BA Korrespondenzen abgeschlossen'
|
||||
);
|
||||
|
||||
end p_process_incoming_ba_data;
|
||||
|
||||
end pck_auto_import;
|
||||
|
||||
@@ -12,8 +12,8 @@ import java.util.UUID;
|
||||
*/
|
||||
public class ProcessingContext {
|
||||
|
||||
/** Eindeutige Lauf-ID — wird als MDC-Feld {@code runId} gesetzt. */
|
||||
public final UUID runId;
|
||||
/** Eindeutige Datei-ID — wird als MDC-Feld {@code fileId} gesetzt. */
|
||||
public final UUID fileId;
|
||||
|
||||
/** Originaler ZIP-Dateiname auf dem SFTP-Server, z.B. {@code export_2026-04-08.zip}. */
|
||||
public final String zipFilename;
|
||||
@@ -39,8 +39,8 @@ public class ProcessingContext {
|
||||
/** Aktueller Verarbeitungsstatus. */
|
||||
public ProcessingStatus status = ProcessingStatus.PENDING;
|
||||
|
||||
public ProcessingContext(UUID runId, String zipFilename) {
|
||||
this.runId = runId;
|
||||
public ProcessingContext(UUID fileId, String zipFilename) {
|
||||
this.fileId = fileId;
|
||||
this.zipFilename = zipFilename;
|
||||
this.zipNameWithoutExt = zipFilename.endsWith(".zip")
|
||||
? zipFilename.substring(0, zipFilename.length() - 4)
|
||||
|
||||
@@ -2,7 +2,6 @@ package de.galabau.dateieingang.ords;
|
||||
|
||||
import de.galabau.dateieingang.config.OrdsConfig;
|
||||
import de.galabau.dateieingang.exception.OrdsException;
|
||||
import de.galabau.dateieingang.model.ProcessingContext;
|
||||
import io.quarkus.logging.Log;
|
||||
import jakarta.enterprise.context.ApplicationScoped;
|
||||
import jakarta.inject.Inject;
|
||||
@@ -35,28 +34,25 @@ public class OrdsNotificationService {
|
||||
* Wird bei transienten Fehlern bis zu 3-mal wiederholt (1s Backoff, 10s Timeout je Versuch).
|
||||
* Maximale Wartezeit: ca. 33 Sekunden (3 × 10s + 3 × 1s Backoff).
|
||||
*
|
||||
* @param context enthält {@code zipNameWithoutExt} für das Log
|
||||
* @throws OrdsException wenn der ORDS-Aufruf nach allen Retries fehlschlägt
|
||||
*/
|
||||
@Retry(maxRetries = 3, delay = 1000, delayUnit = ChronoUnit.MILLIS,
|
||||
retryOn = OrdsException.class)
|
||||
@Timeout(value = 10, unit = ChronoUnit.SECONDS)
|
||||
public void triggerDbProcessing(ProcessingContext context) throws OrdsException {
|
||||
Log.infof("Rufe ORDS-Endpunkt auf für '%s'", context.zipNameWithoutExt);
|
||||
public void triggerDbProcessing() throws OrdsException {
|
||||
Log.info("Rufe ORDS-Endpunkt auf");
|
||||
Response response;
|
||||
try {
|
||||
response = ordsClient.processIncomingBaData(config.apiKey());
|
||||
} catch (Exception e) {
|
||||
throw new OrdsException("ORDS-Verbindung fehlgeschlagen für '"
|
||||
+ context.zipNameWithoutExt + "'", e);
|
||||
throw new OrdsException("ORDS-Verbindung fehlgeschlagen", e);
|
||||
}
|
||||
|
||||
int status = response.getStatus();
|
||||
if (status >= 400) {
|
||||
throw new OrdsException("ORDS antwortete mit HTTP " + status
|
||||
+ " für '" + context.zipNameWithoutExt + "'");
|
||||
throw new OrdsException("ORDS antwortete mit HTTP " + status);
|
||||
}
|
||||
|
||||
Log.infof("ORDS-Endpunkt aufgerufen, HTTP %d für '%s'", status, context.zipNameWithoutExt);
|
||||
Log.infof("ORDS-Endpunkt aufgerufen, HTTP %d", status);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -79,8 +79,11 @@ public class FileProcessingPipeline {
|
||||
}
|
||||
|
||||
private void processAll() {
|
||||
Log.info("Pipeline-Lauf gestartet");
|
||||
UUID pipelineRunId = UUID.randomUUID();
|
||||
MDC.put("pipelineRunId", pipelineRunId.toString());
|
||||
Log.infof("Pipeline-Lauf gestartet [pipelineRunId=%s]", pipelineRunId);
|
||||
|
||||
try {
|
||||
preProcessingCleanup();
|
||||
|
||||
List<String> zipFiles;
|
||||
@@ -93,24 +96,32 @@ public class FileProcessingPipeline {
|
||||
|
||||
if (zipFiles.isEmpty()) {
|
||||
Log.info("Keine neuen ZIP-Dateien auf dem SFTP-Server gefunden");
|
||||
Log.info("Pipeline-Lauf abgeschlossen");
|
||||
return;
|
||||
}
|
||||
|
||||
Log.infof("%d neue ZIP-Datei(en) auf dem SFTP-Server gefunden", zipFiles.size());
|
||||
|
||||
for (String zipFilename : zipFiles) {
|
||||
Log.infof("Datei gefunden: %s", zipFilename);
|
||||
Log.infof("ZIP-Datei gefunden: %s", zipFilename);
|
||||
processZip(zipFilename);
|
||||
}
|
||||
|
||||
Log.info("Pipeline-Lauf abgeschlossen");
|
||||
MDC.put("step", "ords-notify");
|
||||
try {
|
||||
ordsNotificationService.triggerDbProcessing();
|
||||
} catch (OrdsException e) {
|
||||
Log.errorf(e, "ORDS-Benachrichtigung fehlgeschlagen — DB-Verarbeitung wird beim nächsten Lauf ausgelöst");
|
||||
}
|
||||
} finally {
|
||||
Log.infof("Pipeline-Lauf abgeschlossen [pipelineRunId=%s]", pipelineRunId);
|
||||
MDC.clear();
|
||||
}
|
||||
}
|
||||
|
||||
private void processZip(String zipFilename) {
|
||||
ProcessingContext context = new ProcessingContext(UUID.randomUUID(), zipFilename);
|
||||
MDC.put("runId", context.runId.toString());
|
||||
Log.infof("Starte Verarbeitung von '%s' [runId=%s]", zipFilename, context.runId);
|
||||
MDC.put("fileId", context.fileId.toString());
|
||||
Log.infof("Starte Verarbeitung von '%s' [fileId=%s]", zipFilename, context.fileId);
|
||||
|
||||
try {
|
||||
// --- Download ---
|
||||
@@ -142,19 +153,13 @@ public class FileProcessingPipeline {
|
||||
MDC.put("step", "oci-marker");
|
||||
ociUploadService.uploadMarker(context);
|
||||
context.status = ProcessingStatus.MARKER_UPLOADED;
|
||||
|
||||
// --- ORDS Notify ---
|
||||
MDC.put("step", "ords-notify");
|
||||
ordsNotificationService.triggerDbProcessing(context);
|
||||
|
||||
context.status = ProcessingStatus.ORDS_NOTIFIED;
|
||||
Log.infof("Verarbeitung erfolgreich abgeschlossen: '%s'", zipFilename);
|
||||
|
||||
} catch (ZipException e) {
|
||||
Log.errorf(e, "Ungültige ZIP-Datei '%s' — wird zu .error umbenannt", zipFilename);
|
||||
context.status = ProcessingStatus.FAILED;
|
||||
tryRenameToError(zipFilename);
|
||||
} catch (SftpException | OciException | OrdsException e) {
|
||||
} catch (SftpException | OciException e) {
|
||||
Log.errorf(e, "Verarbeitung von '%s' fehlgeschlagen (Infrastruktur): %s", zipFilename, e.getMessage());
|
||||
context.status = ProcessingStatus.FAILED;
|
||||
} catch (IOException e) {
|
||||
@@ -166,10 +171,11 @@ public class FileProcessingPipeline {
|
||||
} finally {
|
||||
postProcessingCleanup(context);
|
||||
long duration = Duration.between(context.startTime, LocalDateTime.now()).toMillis();
|
||||
Log.infof("Lauf %s für Datei %s abgeschlossen — Status: %s, Dauer: %dms",
|
||||
context.runId, zipFilename, context.status, duration);
|
||||
Log.infof("Datei %s abgeschlossen — Status: %s, Dauer: %dms [fileId=%s]",
|
||||
zipFilename, context.status, duration, context.fileId);
|
||||
Log.info("-----------------------------------------------------------------------------------------------------");
|
||||
MDC.clear();
|
||||
MDC.remove("fileId");
|
||||
MDC.remove("step");
|
||||
}
|
||||
}
|
||||
|
||||
@@ -234,7 +240,7 @@ public class FileProcessingPipeline {
|
||||
*/
|
||||
private void postProcessingCleanup(ProcessingContext context) {
|
||||
MDC.put("step", "post-cleanup");
|
||||
Log.infof("Cleanup gestartet für Lauf %s", context.runId);
|
||||
Log.infof("Cleanup gestartet für Datei '%s'", context.zipFilename);
|
||||
try {
|
||||
if (context.localZipPath != null) {
|
||||
Files.deleteIfExists(context.localZipPath);
|
||||
@@ -245,8 +251,8 @@ public class FileProcessingPipeline {
|
||||
Log.infof("Lokales Entpack-Verzeichnis gelöscht: %s", context.localExtractDir);
|
||||
}
|
||||
} catch (IOException e) {
|
||||
Log.warnf(e, "Cleanup für Lauf %s fehlgeschlagen — lokale Dateien verbleiben ggf. in %s",
|
||||
context.runId,
|
||||
Log.warnf(e, "Cleanup für '%s' fehlgeschlagen — lokale Dateien verbleiben ggf. in %s",
|
||||
context.zipFilename,
|
||||
context.localZipPath != null ? context.localZipPath.getParent() : "unbekannt");
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user