snapshot and CDC iterator #5

rustiever · 2022-05-06T15:31:00Z

Description

Snapshot iterator is implemented
CDC iterator is implemented

Fixes # (issue)
#1

Quick checks:

I have followed the Code Guidelines.
There is no other pull request for the same update/change.
I have written unit tests.
I have made sure that the PR is of reasonable size and can be easily reviewed.

rustiever · 2022-05-06T15:31:23Z

soon i'll add unit and integration tests

.gitignore

README.md

config/config.go

source/iterator/snapshot_iterator.go

README.md

rustiever · 2022-05-23T08:22:04Z

Hi @hariso golangci-lint are failing. In summary it says,

Cut method is added in Go 1.18. i think linter works on older version of Go. Do you have any idea?

if nothing works i'll add Cut method myself in Code.

README.md

source/position/position_test.go

source/misc_utils_test.go

source/iterator/snapshot_iterator.go

hariso · 2022-05-31T14:13:46Z

source/iterator/snapshot_iterator.go

+func (s *SnapshotIterator) getLastProcessedDate(ctx context.Context, p position.Position) (time.Time, error) {
+	logger := sdk.Logger(ctx).With().Str("Method", "getInitialDate").Logger()
+	logger.Trace().Msg("Starting the getInitialDate method")
+	var date = p.CreatedAt.Add(1 * time.Second)


Something could have been created during that second.

adding a second to avoid duplicates. if we pull in same time, we may get already read records from marketo. so processing from next second. Also marketo handles data in seconds level only.

@hariso any comment on this?

if we pull in same time, we may get already read records from marketo. so processing from next second.

It feels like that's because our query is "return leads where createdAt is greater or equal than lastModified". If so, we need a query like "return leads where createdAt is greater than lastModified".

Also marketo handles data in seconds level only.

I think that's not a bad reason to to add one second, but what if Marketo changes the precision of timestamps to a millisecond? We may or may not see that soon, but if we can have logic which does not depend on timestamp precision, that would be best.

source/iterator/snapshot_iterator.go

hariso · 2022-05-31T14:19:40Z

source/iterator/cdc_iterator.go

+	for k := range leadIds {
+		keys = append(keys, k)
+	}
+	sort.Ints(keys)


Why is this needed? Also, it looks like we first collect all changed leads and then send them to Conduit?

it requires for some edge cases, especially for func (a acceptanceTest) TestSource_Open_ResumeAtPositionCDC(t *testing.T) { tests, sorting helps in choosing the last key and resuming the CDC mode

So the data from Marketo is not sorted?

a kind of no, coz we get records based on change time, it can be any record. For eg: at given time if id=1234 updated first and later id=1000 updated, then we get records in sequence 1234,1000 so on.

So why is that a problem? The position is based on the change time, so the next time CDC starts, it starts from the last processed change time.

actually without sorting, connector will work fine, but some tests in acceptance tests are failing. especially for func (a acceptanceTest) TestSource_Open_ResumeAtPositionCDC(t *testing.T) { tests, sorting helps in choosing the last key and resuming the CDC mode

So why is that a problem? The position is based on the change time, so the next time CDC starts, it starts from the last processed change time.

in this func (a acceptanceTest) TestSource_Open_ResumeAtPositionCDC(t *testing.T) { test we genrate 10 rec and ack for 5. And then expect other 5 recs. But all these 10 rec are updated in same time. so we'll get same 10 recs again. So we use lastKey to continue from where we left.

@rustiever So the keys sorted, are lead IDs. Are lead IDs actually timestamps? I might be missing something, but I can't see a connection. We use the last modified timestamp of a lead in the position, and when resuming CDC. But here, we are sorting lead IDs to make sure we can resume from a position.

we primarily use lastModiefiedTime for CDC. But sometimes this doesn't work when we resume from CDC. For eg: at 10:15 time we get 8 recs 5,6,3,2,7,1,8,4. Lets assume we processed 4(5,6,3,2) records among them. timestamp for all records are same. So when we resume CDC, We get again same set of records. we need a way to filter the processed records. So we use sorting. once the records are sorted such as 1,2,3,4,5,6,7,8, and CDC is resumed we get lasttime and lastKey. We compare fetched records with lastKey to filter which records are not processed.

marketo-client/client.go

source/misc_utils_test.go

source/acceptance_test.go

hariso · 2022-06-06T12:15:44Z

source/iterator/snapshot_iterator.go

+)
+
+var (
+	InitialDate time.Time // holds the initial date of the snapshot


Why not make it part of SnapshotIterator?

this variable was initially with SnapshotIterator, later moved as Global var for testing. if it is in SnapshotIterator then tests taking too much time. so for testing InitialDate will be current datetime

If the variable is part of SnapshotIterator, then you can use a different value for it in testing, no? Introducing a global variable only to make tests work feels like something which can be improved.

will be handled in next PR @hariso

hariso · 2022-06-06T12:20:27Z

source/iterator/cdc_iterator.go

+	for k := range leadIds {
+		keys = append(keys, k)
+	}
+	sort.Ints(keys)


So why is that a problem? The position is based on the change time, so the next time CDC starts, it starts from the last processed change time.

hariso · 2022-06-09T15:04:05Z

source/acceptance_test.go

+// AcceptanceSourceTestDriverConfig contains the configuration for
+// AcceptanceTestDriver.
+type AcceptanceSourceTestDriverConfig struct {
+	// Connector is the connector to be tested.
+	Connector sdk.Connector
+
+	// SourceConfig should be a valid config for a source connector, reading
+	// from the same location as the destination will write to.
+	SourceConfig map[string]string
+	// DestinationConfig should be a valid config for a destination connector,
+	// writing to the same location as the source will read from.
+	DestinationConfig map[string]string
+
+	// BeforeTest is executed before each acceptance test.
+	BeforeTest func(t *testing.T)
+	// AfterTest is executed after each acceptance test.
+	AfterTest func(t *testing.T)
+
+	// GoleakOptions will be applied to goleak.VerifyNone. Can be used to
+	// suppress false positive goroutine leaks.
+	GoleakOptions []goleak.Option
+}
+
+func (d AcceptanceTestDriver) DestinationConfig(*testing.T) map[string]string {
+	return map[string]string{}
+}
+func (d AcceptanceTestDriver) Connector() sdk.Connector {
+	return d.Config.Connector
+}
+
+func (d AcceptanceTestDriver) SourceConfig(*testing.T) map[string]string {
+	return d.Config.SourceConfig
+}
+
+func (d AcceptanceTestDriver) BeforeTest(t *testing.T) {
+	// before test check if the test should be skipped
+	d.Skip(t)
+
+	if d.Config.BeforeTest != nil {
+		d.Config.BeforeTest(t)
+	}
+}
+
+func (d AcceptanceTestDriver) AfterTest(t *testing.T) {
+	if d.Config.AfterTest != nil {
+		d.Config.AfterTest(t)
+	}
+}
+
+func (d AcceptanceTestDriver) GoleakOptions(_ *testing.T) []goleak.Option {
+	return d.Config.GoleakOptions
+}
+
+func (d AcceptanceTestDriver) ReadTimeout() time.Duration {
+	return time.Minute * 5
+}


It looks like we don't need these methods anymore.

when i delete these methods tests are failing

@rustiever because of timeouts? Did you update to the latest SDK?

hariso · 2022-06-09T15:04:39Z

source/iterator/cdc_iterator.go

+	logger := sdk.Logger(ctx).With().Str("Method", "flushLatestLeads").Logger()
+	logger.Trace().Msg("Starting the flushLatestLeads")
+	token, err := c.client.GetNextPageToken(c.lastModified)
+	c.lastModified = time.Now().UTC()


Maybe it's safer to change this only once we actually get the leads?

sorry @hariso, i couldn't understand, can you elaborate pls

Here, you first update c.lastModified, then get the changed and deleted leads. Is it safer, to first get the changed and deleted leads, and then update c.lastModified?

hariso · 2022-06-13T09:22:06Z

source/iterator/snapshot_iterator.go

+const (
+	// is the maximum number of days between two snapshots. If the gap between two snapshots is greater than this number,
+	// API will return an error. This is limitation of the API.
+	MaximumDaysGap = 744 // 31 days in Hours


We should then call it MaximumHoursGap.

rustiever and others added 4 commits May 6, 2022 20:36

snapshot iterator

0f895b4

Merge branch 'main' into snapshot-iterator

42e5f6b

update go.sum

b82e39f

cdc iterator

f65f536

rustiever requested review from neha-Gupta1 and WeirdMagician May 6, 2022 15:31

rustiever self-assigned this May 6, 2022

rustiever force-pushed the CDC-iterator branch 2 times, most recently from b56f29b to 66d4e8c Compare May 12, 2022 07:36

updating the CDC iterator

ed0ff33

rustiever force-pushed the CDC-iterator branch from 66d4e8c to ed0ff33 Compare May 12, 2022 08:22

rustiever requested review from hariso and maha-hajja May 12, 2022 08:25

rustiever changed the title ~~Cdc iterator~~ snapshot and CDC iterator May 12, 2022

rustiever linked an issue May 12, 2022 that may be closed by this pull request

Marketo source connector for Conduit #1

Closed

updating config keys to camel case

9e6c5dd

rustiever force-pushed the CDC-iterator branch from 5f20e6d to 9e6c5dd Compare May 13, 2022 06:25

hariso reviewed May 19, 2022

View reviewed changes

rustiever added 3 commits May 20, 2022 16:07

integration tests

b0ff0a2

Merge branch 'integration-tests' into CDC-iterator

34ef375

fixes for various review comments

5bb5f2b

fix config tests

c0ef0a1

rustiever force-pushed the CDC-iterator branch from e9c0cbb to c0ef0a1 Compare May 23, 2022 08:29

rustiever requested review from hariso and removed request for neha-Gupta1 and WeirdMagician May 23, 2022 08:29

updating readme and tests

270ed8e

rustiever force-pushed the CDC-iterator branch from 8d3e19c to 270ed8e Compare May 25, 2022 06:47

rustiever force-pushed the CDC-iterator branch from 04e9858 to d01908c Compare May 27, 2022 10:38

rustiever requested a review from hariso May 27, 2022 11:22

review comments

17ef027

rustiever force-pushed the CDC-iterator branch from d01908c to 17ef027 Compare May 27, 2022 11:57

rustiever added 2 commits May 27, 2022 17:34

Merge branch 'main' into CDC-iterator

58db1e8

fix broken links

d69379b

hariso reviewed May 30, 2022

View reviewed changes

README.md Outdated Show resolved Hide resolved

hariso reviewed May 30, 2022

View reviewed changes

source/position/position_test.go Outdated Show resolved Hide resolved

hariso reviewed May 30, 2022

View reviewed changes

source/misc_utils_test.go Show resolved Hide resolved

Update README.md

73bb3f1

hariso reviewed May 31, 2022

View reviewed changes

rustiever added 2 commits June 2, 2022 15:31

accptance test

89be830

Merge branch 'acceptance-test' into CDC-iterator

d903ee2

rustiever requested a review from hariso June 3, 2022 12:22

review comments

1a1ae5d

rustiever force-pushed the CDC-iterator branch from cdef84c to 1a1ae5d Compare June 4, 2022 06:22

hariso reviewed Jun 6, 2022

View reviewed changes

rustiever requested a review from hariso June 9, 2022 05:45

configure readTimeout in acceptance tests

f90af32

rustiever force-pushed the CDC-iterator branch from 660605f to f90af32 Compare June 9, 2022 06:37

hariso reviewed Jun 9, 2022

View reviewed changes

rustiever requested a review from hariso June 13, 2022 05:52

hariso reviewed Jun 13, 2022

View reviewed changes

rustiever force-pushed the CDC-iterator branch from 35cabb4 to 70c02a6 Compare June 13, 2022 10:21

review comments

8182f4c

rustiever force-pushed the CDC-iterator branch from 70c02a6 to 8182f4c Compare June 13, 2022 10:26

rustiever requested a review from hariso June 13, 2022 10:29

hariso approved these changes Jun 13, 2022

View reviewed changes

rustiever merged commit 6e3228a into main Jun 13, 2022

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

snapshot and CDC iterator #5

snapshot and CDC iterator #5

rustiever commented May 6, 2022 •

edited

Loading

rustiever commented May 6, 2022

rustiever commented May 23, 2022 •

edited

Loading

hariso May 31, 2022

rustiever Jun 2, 2022

rustiever Jun 11, 2022

hariso Jun 13, 2022

hariso May 31, 2022

rustiever Jun 3, 2022

hariso Jun 6, 2022

rustiever Jun 6, 2022

hariso Jun 6, 2022

rustiever Jun 9, 2022

rustiever Jun 9, 2022

hariso Jun 9, 2022

rustiever Jun 11, 2022

hariso Jun 6, 2022

rustiever Jun 9, 2022

hariso Jun 13, 2022

rustiever Jun 13, 2022

hariso Jun 6, 2022

hariso Jun 9, 2022

rustiever Jun 11, 2022

hariso Jun 13, 2022

rustiever Jun 13, 2022

hariso Jun 9, 2022

rustiever Jun 11, 2022

hariso Jun 13, 2022

rustiever Jun 13, 2022

hariso Jun 13, 2022

rustiever Jun 13, 2022

snapshot and CDC iterator #5

snapshot and CDC iterator #5

Conversation

rustiever commented May 6, 2022 • edited Loading

Description

Quick checks:

rustiever commented May 6, 2022

rustiever commented May 23, 2022 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

rustiever commented May 6, 2022 •

edited

Loading

rustiever commented May 23, 2022 •

edited

Loading