refactor: standardize job error channel size; update batch size for transformed batches
This commit is contained in:
@@ -20,6 +20,8 @@ import (
|
||||
"golang.org/x/sync/errgroup"
|
||||
)
|
||||
|
||||
const jobErrorsChannelSize int = 100
|
||||
|
||||
func buildTruncateQuery(targetDbType, schema, table, truncateMethod string) string {
|
||||
if truncateMethod == "DELETE" {
|
||||
if targetDbType == "postgres" {
|
||||
@@ -110,11 +112,11 @@ func processMigrationJob(
|
||||
log.Error("Unexpected error calculating batch ranges: ", err)
|
||||
}
|
||||
|
||||
chJobErrors := make(chan custom_errors.JobError, job.ExtractorQueueSize)
|
||||
chJobErrors := make(chan custom_errors.JobError, jobErrorsChannelSize)
|
||||
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)
|
||||
chBatchesTransformed := make(chan models.Batch, job.TransformerQueueSize)
|
||||
|
||||
var wgActivePartitions sync.WaitGroup
|
||||
var wgActiveBatches sync.WaitGroup
|
||||
|
||||
Reference in New Issue
Block a user