refactor: consolidate error handling and retry configuration

Centralize the error channel size, retry policy and max-failed batches/partition limits in the job config.

cmd/go_migrate/process.go

@@ -110,12 +110,11 @@ func processMigrationJob(
 		log.Error("Unexpected error calculating batch ranges: ", err)
 	}
 
-	chJobErrors := make(chan custom_errors.JobError, job.QueueSize)
-	chExtractorErrors := make(chan custom_errors.ExtractorError, job.QueueSize)
-	chLoadersErrors := make(chan custom_errors.LoaderError, job.QueueSize)
-	chPartitions := make(chan models.Partition, job.QueueSize)
-	chBatchesRaw := make(chan models.Batch, job.QueueSize)
-	chBatchesTransformed := make(chan models.Batch, job.QueueSize)
+	chJobErrors := make(chan custom_errors.JobError, job.ExtractorQueueSize)
+	chLoadersErrors := make(chan custom_errors.LoaderError, job.ExtractorQueueSize)
+	chPartitions := make(chan models.Partition, job.ExtractorQueueSize)
+	chBatchesRaw := make(chan models.Batch, job.ExtractorQueueSize)
+	chBatchesTransformed := make(chan models.Batch, job.ExtractorQueueSize)
 
 	var wgActivePartitions sync.WaitGroup
 	var wgActiveBatches sync.WaitGroup
@@ -131,19 +130,10 @@ func processMigrationJob(
 		}
 	}()
 
-	go custom_errors.ExtractorErrorHandler(
-		localCtx,
-		job.Retry,
-		job.MaxPartitionErrrors,
-		chExtractorErrors,
-		chPartitions,
-		chJobErrors,
-		&wgActivePartitions,
-	)
 	go custom_errors.LoaderErrorHandler(
 		localCtx,
 		job.Retry,
-		job.MaxChunkErrors,
+		job.MaxExtractorBatchErrors,
 		chLoadersErrors,
 		chBatchesTransformed,
 		chJobErrors,
@@ -159,10 +149,10 @@ func processMigrationJob(
 				localCtx,
 				job.SourceTable,
 				sourceColTypes,
-				job.BatchSize,
+				job.ExtractorBatchSize,
+				job.Retry,
 				chPartitions,
 				chBatchesRaw,
-				chExtractorErrors,
 				chJobErrors,
 				&wgActivePartitions,
 				&rowsRead,
@@ -216,8 +206,6 @@ func processMigrationJob(
 		log.Debugf("wgActivePartitions is empty (%v)", job.Name)
 		close(chPartitions)
 		log.Debugf("chPartitions is closed (%v)", job.Name)
-		close(chExtractorErrors)
-		log.Debugf("chExtractorErrors is closed (%v)", job.Name)
 
 		wgExtractors.Wait()
 		log.Debugf("wgExtractors is empty (%v)", job.Name)