fix: enhance error handling in extractor and loader processes; ensure proper job error propagation and logging

This commit is contained in:
2026-04-12 20:11:19 -05:00
parent 01780b4b02
commit 5633dc98d0
5 changed files with 46 additions and 14 deletions

View File

@@ -43,6 +43,7 @@ func ExtractorErrorHandler(
}
if err.Partition.RetryCounter >= maxRetryAttempts {
wgActivePartitions.Done()
jobError := JobError{
ShouldCancelJob: false,
Msg: fmt.Sprintf("Partition %v reached max retries (%d)", err.Partition.Id, maxRetryAttempts),
@@ -55,8 +56,19 @@ func ExtractorErrorHandler(
return
}
wgActivePartitions.Done()
continue
} else {
jobError := JobError{
ShouldCancelJob: false,
Msg: fmt.Sprintf("Temporal error in partition %v (retries: %d)", err.Partition.Id, err.Partition.RetryCounter),
Prev: &err,
}
select {
case chJobErrorsOut <- jobError:
case <-ctx.Done():
return
}
}
newPartition := err.Partition

View File

@@ -37,11 +37,11 @@ func JobErrorHandler(ctx context.Context, chErrorsIn <-chan JobError) error {
}
if err.ShouldCancelJob {
log.Error(err.Msg, " - ", err.Prev)
log.Errorf("(Fatal job error) - %v - %v", err.Msg, err.Prev)
return &err
}
log.Error(err.Msg, " - ", err.Prev)
log.Errorf("%v - %v", err.Msg, err.Prev)
}
}
}

View File

@@ -40,6 +40,7 @@ func LoaderErrorHandler(
}
if err.RetryCounter >= maxRetryAttempts {
wgActiveBatches.Done()
jobError := JobError{
ShouldCancelJob: false,
Msg: fmt.Sprintf("Batch %v reached max retries (%d)", err.Id, maxRetryAttempts),
@@ -52,8 +53,19 @@ func LoaderErrorHandler(
return
}
wgActiveBatches.Done()
continue
} else {
jobError := JobError{
ShouldCancelJob: false,
Msg: fmt.Sprintf("Temporal error in batch %v (retries: %d)", err.Batch.Id, err.Batch.RetryCounter),
Prev: &err,
}
select {
case chJobErrorsOut <- jobError:
case <-ctx.Done():
return
}
}
err.RetryCounter++