Skip to content

Commit

Permalink
Reduce CDC timeout for PG (#406)
Browse files Browse the repository at this point in the history
  • Loading branch information
iskakaushik authored Sep 21, 2023
1 parent 6126e64 commit 06a53ed
Show file tree
Hide file tree
Showing 2 changed files with 19 additions and 2 deletions.
19 changes: 18 additions & 1 deletion flow/activities/flowable.go
Original file line number Diff line number Diff line change
Expand Up @@ -157,6 +157,7 @@ func (a *FlowableActivity) CreateNormalizedTable(
// StartFlow implements StartFlow.
func (a *FlowableActivity) StartFlow(ctx context.Context,
input *protos.StartFlowInput) (*model.SyncResponse, error) {
activity.RecordHeartbeat(ctx, "starting flow...")
conn := input.FlowConnectionConfigs

ctx = context.WithValue(ctx, shared.EnableMetricsKey, a.EnableMetrics)
Expand All @@ -180,6 +181,7 @@ func (a *FlowableActivity) StartFlow(ctx context.Context,
if err != nil {
return nil, fmt.Errorf("failed to initialize table schema: %w", err)
}
activity.RecordHeartbeat(ctx, "initialized table schema")

log.WithFields(log.Fields{
"flowName": input.FlowConnectionConfigs.FlowJobName,
Expand All @@ -202,6 +204,10 @@ func (a *FlowableActivity) StartFlow(ctx context.Context,
return nil, fmt.Errorf("failed to pull records: %w", err)
}
recordBatch := recordsWithTableSchemaDelta.RecordBatch

pullRecordWithCount := fmt.Sprintf("pulled %d records", len(recordBatch.Records))
activity.RecordHeartbeat(ctx, pullRecordWithCount)

if a.CatalogMirrorMonitor.IsActive() && len(recordBatch.Records) > 0 {
syncBatchID, err := dstConn.GetLastSyncBatchID(input.FlowConnectionConfigs.FlowJobName)
if err != nil && conn.Destination.Type != protos.DBType_EVENTHUB {
Expand Down Expand Up @@ -238,6 +244,15 @@ func (a *FlowableActivity) StartFlow(ctx context.Context,
}, nil
}

shutdown := utils.HeartbeatRoutine(ctx, 10*time.Second, func() string {
jobName := input.FlowConnectionConfigs.FlowJobName
return fmt.Sprintf("pushing records for job - %s", jobName)
})

defer func() {
shutdown <- true
}()

syncStartTime := time.Now()
res, err := dstConn.SyncRecords(&model.SyncRecordsRequest{
Records: recordBatch,
Expand Down Expand Up @@ -275,7 +290,9 @@ func (a *FlowableActivity) StartFlow(ctx context.Context,
}
res.TableSchemaDelta = recordsWithTableSchemaDelta.TableSchemaDelta
res.RelationMessageMapping = recordsWithTableSchemaDelta.RelationMessageMapping
activity.RecordHeartbeat(ctx, "pushed records")

pushedRecordsWithCount := fmt.Sprintf("pushed %d records", numRecords)
activity.RecordHeartbeat(ctx, pushedRecordsWithCount)

return res, nil
}
Expand Down
2 changes: 1 addition & 1 deletion flow/workflows/sync_flow.go
Original file line number Diff line number Diff line change
Expand Up @@ -85,7 +85,7 @@ func (s *SyncFlowExecution) executeSyncFlow(
StartToCloseTimeout: 24 * time.Hour,
// TODO: activity needs to call heartbeat.
// see https://github.com/PeerDB-io/nexus/issues/216
HeartbeatTimeout: 5 * time.Minute,
HeartbeatTimeout: 30 * time.Second,
})

// execute StartFlow on the peers to start the flow
Expand Down

0 comments on commit 06a53ed

Please sign in to comment.