diff --git a/flow/activities/flowable.go b/flow/activities/flowable.go index cc09bae0d7..8a65f5dded 100644 --- a/flow/activities/flowable.go +++ b/flow/activities/flowable.go @@ -287,11 +287,13 @@ func (a *FlowableActivity) MaintainPull( ctx = context.WithValue(ctx, shared.FlowNameKey, config.FlowJobName) srcConn, err := connectors.GetByNameAs[connectors.CDCPullConnector](ctx, config.Env, a.CatalogPool, config.SourceName) if err != nil { + a.Alerter.LogFlowError(ctx, config.FlowJobName, err) return err } defer connectors.CloseConnector(ctx, srcConn) if err := srcConn.SetupReplConn(ctx); err != nil { + a.Alerter.LogFlowError(ctx, config.FlowJobName, err) return err } diff --git a/flow/alerting/alerting.go b/flow/alerting/alerting.go index 5f05005d14..69282330d6 100644 --- a/flow/alerting/alerting.go +++ b/flow/alerting/alerting.go @@ -356,7 +356,7 @@ func (a *Alerter) checkAndAddAlertToCatalog(ctx context.Context, alertConfigId i return true } - logger.Info(fmt.Sprintf("Skipped sending alerts: last alert was sent at %s, which was >=%s ago", createdTimestamp.String(), dur.String())) + logger.Info(fmt.Sprintf("Skipped sending alerts: last alert was sent at %s, which was <=%s ago", createdTimestamp.String(), dur.String())) return false } diff --git a/ui/app/peers/[peerName]/lagGraph.tsx b/ui/app/peers/[peerName]/lagGraph.tsx index 87b90fa8c8..d971bee8f0 100644 --- a/ui/app/peers/[peerName]/lagGraph.tsx +++ b/ui/app/peers/[peerName]/lagGraph.tsx @@ -21,9 +21,10 @@ type LagGraphProps = { function parseLSN(lsn: string): number { if (!lsn) return 0; const [lsn1, lsn2] = lsn.split('/'); - return Number( - (BigInt(parseInt(lsn1, 16)) << BigInt(32)) | BigInt(parseInt(lsn2, 16)) - ); + const parsedLsn1 = parseInt(lsn1, 16); + const parsedLsn2 = parseInt(lsn2, 16); + if (isNaN(parsedLsn1) || isNaN(parsedLsn2)) return 0; + return Number((BigInt(parsedLsn1) << BigInt(32)) | BigInt(parsedLsn2)); } export default function LagGraph({ peerName }: LagGraphProps) {