Skip to content

returns snapshot incomplete error when snapshot is interrupted #27

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 17 commits into from
Apr 8, 2022
Merged

returns snapshot incomplete error when snapshot is interrupted #27

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
17 commits
Select commit Hold shift + click to select a range
53be2c2
snapshot iterator
dylanlott Apr 1, 2022
ee5ea8b
adds lifecycle test for proper cleanup
dylanlott Apr 5, 2022
7aa9acd
linter fixes
dylanlott Apr 5, 2022
17517bf
Merge branch 'main' of github.com:ConduitIO/conduit-connector-postgre…
dylanlott Apr 5, 2022
4242433
refactors tx handling to use pgx.Tx
dylanlott Apr 5, 2022
a549d9f
use is instead of i for package reference
dylanlott Apr 5, 2022
3292e8c
clean up and review feedback
dylanlott Apr 6, 2022
3f20c27
includes table name in snapshot positions
dylanlott Apr 6, 2022
0c597bf
use pooled connection in tests
dylanlott Apr 6, 2022
b9e2a1d
checks for context errors in Next
dylanlott Apr 7, 2022
19ca9e0
handle rollback error in loadRows
dylanlott Apr 7, 2022
4c9630a
use query row instead of query in test setup
dylanlott Apr 7, 2022
b46f343
returns snapshot incomplete error when snapshot is interrupted
dylanlott Apr 6, 2022
17902c3
linter fix
dylanlott Apr 6, 2022
b544e42
adds log or return behavior to Teardown
dylanlott Apr 7, 2022
c1add03
Merge branch 'main' of github.com:ConduitIO/conduit-connector-postgre…
dylanlott Apr 8, 2022
22e5c1a
use logOrReturn in Teardown
dylanlott Apr 8, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
34 changes: 30 additions & 4 deletions source/logrepl/snapshot.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@ package logrepl
import (
"context"
"database/sql"
"errors"
"fmt"
"strconv"
"time"
Expand All @@ -29,6 +30,12 @@ import (
"github.com/jackc/pgx/v4"
)

// ErrSnapshotComplete is returned by Next when a snapshot is finished
var ErrSnapshotComplete = errors.New("snapshot complete")

// ErrSnapshotInterrupt is returned by Teardown when a snapshot is interrupted
var ErrSnapshotInterrupt = errors.New("snapshot interrupted")

var psql = sq.StatementBuilder.PlaceholderFormat(sq.Dollar)

const actionSnapshot string = "snapshot"
Expand All @@ -46,6 +53,7 @@ type SnapshotIterator struct {
tx pgx.Tx
rows pgx.Rows

complete bool
internalPos int64
}

Expand Down Expand Up @@ -120,8 +128,8 @@ func (s *SnapshotIterator) Next(ctx context.Context) (sdk.Record, error) {
if err := s.rows.Err(); err != nil {
return sdk.Record{}, fmt.Errorf("rows error: %w", err)
}

return sdk.Record{}, sdk.ErrBackoffRetry
s.complete = true
return sdk.Record{}, ErrSnapshotComplete
}

return s.buildRecord(ctx)
Expand All @@ -135,10 +143,19 @@ func (s *SnapshotIterator) Ack(ctx context.Context, pos sdk.Position) error {
// Teardown attempts to gracefully teardown the iterator.
func (s *SnapshotIterator) Teardown(ctx context.Context) error {
s.rows.Close()
var err error
if commitErr := s.tx.Commit(ctx); commitErr != nil {
Copy link
Contributor

@hariso hariso Apr 7, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Might be a dumb question, but why do we call s.tx.Commit() here at all?

Copy link
Contributor

@lovromazgon lovromazgon Apr 7, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Simply to end the transaction.

Copy link
Contributor

@hariso hariso Apr 7, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ah, I think I need a break.:D I was thinking this is the source teardown, but it's the iterator teardown, and actually a different thing in mind because of that.

sdk.Logger(ctx).Err(commitErr).Msg("teardown commit failed")
err = logOrReturnError(ctx, err, commitErr, "teardown commit failed")
}
if rowsErr := s.rows.Err(); rowsErr != nil {
err = logOrReturnError(ctx, err, rowsErr, "rows returned an error")
}

if !s.complete {
err = logOrReturnError(ctx, err, ErrSnapshotInterrupt, "snapshot interrupted")
}
return s.rows.Err()

return err
}

func (s *SnapshotIterator) buildRecord(ctx context.Context) (sdk.Record, error) {
Expand Down Expand Up @@ -216,3 +233,12 @@ func oidToScannerValue(oid pgtype.OID) scannerValue {
}
return t
}

// logOrReturn
func logOrReturnError(ctx context.Context, oldErr, newErr error, msg string) error {
if oldErr == nil {
return fmt.Errorf(msg+": %w", newErr)
}
sdk.Logger(ctx).Err(newErr).Msg(msg)
return oldErr
}
3 changes: 2 additions & 1 deletion source/logrepl/snapshot_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@ package logrepl

import (
"context"
"errors"
"fmt"
"testing"
"time"
Expand Down Expand Up @@ -70,7 +71,7 @@ func TestLifecycle(t *testing.T) {
"table": table,
},
})
is.NoErr(s.Teardown(ctx)) // TODO: Should return an error
is.True(errors.Is(s.Teardown(ctx), ErrSnapshotInterrupt))
}

// createTestSnapshot starts a transaction that stays open while a snapshot test
Expand Down