Skip to content

Commit 06379a7

Browse files
jgao54jgao54
committed
review
1 parent 8defa78 commit 06379a7

1 file changed

Lines changed: 77 additions & 77 deletions

File tree

‎flow/activities/flowable.go‎

Lines changed: 77 additions & 77 deletions
Original file line numberDiff line numberDiff line change
@@ -581,6 +581,83 @@ func (a *FlowableActivity) GetQRepPartitions(ctx context.Context,
581581
}, nil
582582
}
583583

584+
func (a *FlowableActivity) ReplicateQRepPartitions(ctx context.Context,
585+
config *protos.QRepConfig,
586+
partitions *protos.QRepPartitionBatch,
587+
runUUID string,
588+
) error {
589+
shutdown := common.HeartbeatRoutine(ctx, func() string {
590+
return "replicating partitions for job"
591+
})
592+
defer shutdown()
593+
594+
ctx = context.WithValue(ctx, shared.FlowNameKey, config.FlowJobName)
595+
logger := log.With(internal.LoggerFromCtx(ctx), slog.String(string(shared.FlowNameKey), config.FlowJobName))
596+
597+
if err := monitoring.UpdateStartTimeForQRepRun(ctx, a.CatalogPool, runUUID); err != nil {
598+
return fmt.Errorf("failed to update start time for qrep run: %w", err)
599+
}
600+
601+
numPartitions := len(partitions.Partitions)
602+
logger.Info("replicating partitions for batch",
603+
slog.String("table", config.WatermarkTable),
604+
slog.Int64("batchID", int64(partitions.BatchId)),
605+
slog.Int("totalPartitions", numPartitions))
606+
607+
qRepPullCoreConn, qRepPullCoreClose, err := connectors.GetByNameAs[connectors.QRepPullConnectorCore](
608+
ctx, config.Env, a.CatalogPool, config.SourceName)
609+
if err != nil {
610+
return a.Alerter.LogFlowError(ctx, config.FlowJobName, fmt.Errorf("failed to get qrep source connector: %w", err))
611+
}
612+
defer qRepPullCoreClose(ctx)
613+
614+
dstPeer, qRepSyncCoreConn, qRepSyncCoreClose, err := connectors.LoadPeerAndGetByNameAs[connectors.QRepSyncConnectorCore](
615+
ctx,
616+
config.Env,
617+
a.CatalogPool,
618+
config.DestinationName,
619+
)
620+
if err != nil {
621+
return a.Alerter.LogFlowError(ctx, config.FlowJobName, fmt.Errorf("failed to get qrep destination connector: %w", err))
622+
}
623+
defer qRepSyncCoreClose(ctx)
624+
625+
replicatePartitionFunc, err := initializeReplicatePartitionFunc(ctx, a, config, runUUID, dstPeer.Type, qRepPullCoreConn, qRepSyncCoreConn)
626+
if err != nil {
627+
logger.Error("failed to initialize replication method", slog.Any("error", err))
628+
return a.Alerter.LogFlowError(ctx, config.FlowJobName, err)
629+
}
630+
631+
for i, partition := range partitions.Partitions {
632+
partLogger := log.With(logger,
633+
slog.Int64("batchID", int64(partitions.BatchId)),
634+
slog.String("partitionId", partition.PartitionId),
635+
slog.String("table", config.WatermarkTable),
636+
slog.Int("partitionNum", i+1),
637+
slog.Int("totalPartitions", numPartitions))
638+
639+
startTime := time.Now()
640+
partLogger.Info(fmt.Sprintf("start replicating partition %d/%d of table %s", i+1, numPartitions, config.WatermarkTable))
641+
642+
if err := replicatePartitionFunc(partition); err != nil {
643+
partLogger.Error(fmt.Sprintf("failed to replicate partition %d/%d of table %s", i+1, numPartitions, config.WatermarkTable),
644+
slog.Any("error", err))
645+
return a.Alerter.LogFlowError(ctx, config.FlowJobName, err)
646+
}
647+
648+
partLogger.Info(fmt.Sprintf("finished replicating partition %d/%d of table %s", i+1, numPartitions, config.WatermarkTable),
649+
slog.Time("startTime", startTime),
650+
slog.Time("finishTime", time.Now()))
651+
}
652+
653+
a.Alerter.LogFlowInfo(
654+
ctx,
655+
config.FlowJobName,
656+
fmt.Sprintf("replicated %d partitions to destination for table %s", numPartitions, config.DestinationTableIdentifier),
657+
)
658+
return nil
659+
}
660+
584661
func initializeReplicatePartitionFunc(
585662
ctx context.Context,
586663
a *FlowableActivity,
@@ -667,83 +744,6 @@ func initializeReplicatePartitionFunc(
667744
}
668745
}
669746

670-
func (a *FlowableActivity) ReplicateQRepPartitions(ctx context.Context,
671-
config *protos.QRepConfig,
672-
partitions *protos.QRepPartitionBatch,
673-
runUUID string,
674-
) error {
675-
shutdown := common.HeartbeatRoutine(ctx, func() string {
676-
return "replicating partitions for job"
677-
})
678-
defer shutdown()
679-
680-
ctx = context.WithValue(ctx, shared.FlowNameKey, config.FlowJobName)
681-
logger := log.With(internal.LoggerFromCtx(ctx), slog.String(string(shared.FlowNameKey), config.FlowJobName))
682-
683-
if err := monitoring.UpdateStartTimeForQRepRun(ctx, a.CatalogPool, runUUID); err != nil {
684-
return fmt.Errorf("failed to update start time for qrep run: %w", err)
685-
}
686-
687-
numPartitions := len(partitions.Partitions)
688-
logger.Info("replicating partitions for batch",
689-
slog.String("table", config.WatermarkTable),
690-
slog.Int64("batchID", int64(partitions.BatchId)),
691-
slog.Int("totalPartitions", numPartitions))
692-
693-
qRepPullCoreConn, qRepPullCoreClose, err := connectors.GetByNameAs[connectors.QRepPullConnectorCore](
694-
ctx, config.Env, a.CatalogPool, config.SourceName)
695-
if err != nil {
696-
return a.Alerter.LogFlowError(ctx, config.FlowJobName, fmt.Errorf("failed to get qrep source connector: %w", err))
697-
}
698-
defer qRepPullCoreClose(ctx)
699-
700-
dstPeer, qRepSyncCoreConn, qRepSyncCoreClose, err := connectors.LoadPeerAndGetByNameAs[connectors.QRepSyncConnectorCore](
701-
ctx,
702-
config.Env,
703-
a.CatalogPool,
704-
config.DestinationName,
705-
)
706-
if err != nil {
707-
return a.Alerter.LogFlowError(ctx, config.FlowJobName, fmt.Errorf("failed to get qrep destination connector: %w", err))
708-
}
709-
defer qRepSyncCoreClose(ctx)
710-
711-
replicatePartitionFunc, err := initializeReplicatePartitionFunc(ctx, a, config, runUUID, dstPeer.Type, qRepPullCoreConn, qRepSyncCoreConn)
712-
if err != nil {
713-
logger.Error("failed to initialize replication method", slog.Any("error", err))
714-
return a.Alerter.LogFlowError(ctx, config.FlowJobName, err)
715-
}
716-
717-
for i, partition := range partitions.Partitions {
718-
partLogger := log.With(logger,
719-
slog.Int64("batchID", int64(partitions.BatchId)),
720-
slog.String("partitionId", partition.PartitionId),
721-
slog.String("table", config.WatermarkTable),
722-
slog.Int("partitionNum", i+1),
723-
slog.Int("totalPartitions", numPartitions))
724-
725-
startTime := time.Now()
726-
partLogger.Info(fmt.Sprintf("start replicating partition %d/%d of table %s", i+1, numPartitions, config.WatermarkTable))
727-
728-
if err := replicatePartitionFunc(partition); err != nil {
729-
partLogger.Error(fmt.Sprintf("failed to replicate partition %d/%d of table %s", i+1, numPartitions, config.WatermarkTable),
730-
slog.Any("error", err))
731-
return a.Alerter.LogFlowError(ctx, config.FlowJobName, err)
732-
}
733-
734-
partLogger.Info(fmt.Sprintf("finished replicating partition %d/%d of table %s", i+1, numPartitions, config.WatermarkTable),
735-
slog.Time("startTime", startTime),
736-
slog.Time("finishTime", time.Now()))
737-
}
738-
739-
a.Alerter.LogFlowInfo(
740-
ctx,
741-
config.FlowJobName,
742-
fmt.Sprintf("replicated %d partitions to destination for table %s", numPartitions, config.DestinationTableIdentifier),
743-
)
744-
return nil
745-
}
746-
747747
func (a *FlowableActivity) ConsolidateQRepPartitions(ctx context.Context, config *protos.QRepConfig,
748748
runUUID string,
749749
) error {

0 commit comments

Comments
 (0)