Source multiple topics #120
Conversation
source_pubsub/client.go
Outdated
| c.startCDC(ctx, topic) | ||
| return nil |
There was a problem hiding this comment.
@anna-cross is it intended to not return the result of c.startCDC()? if so, what's the context?
My understanding is that this is anyways within the func() passed in above, so it shouldn't break the for loop
There was a problem hiding this comment.
Woops yes! I'll move it back into return
source_pubsub/source_test.go
Outdated
| @@ -45,7 +45,7 @@ func Test_Read(t *testing.T) { | |||
| ClientID: "test-client-id", | |||
| ClientSecret: "test-client-secret", | |||
| OAuthEndpoint: "https://somewhere", | |||
| TopicName: "/events/TestEvent__e", | |||
| TopicNames: []string{"/events/TestEvent__e"}, | |||
There was a problem hiding this comment.
nice. would be great to also add a test case here with 2 topics - thought, you might already be on this since it's in draft mode
source_pubsub/source.go
Outdated
| // setting topic name as collection | ||
| r.Metadata.SetCollection(s.config.TopicName) | ||
| // r.Metadata.SetCollection(s.config.TopicName) |
There was a problem hiding this comment.
should we remove this?
anna-cross
force-pushed
the
source_multiple_topics
branch
from
9b5e666
to
c59c6e2
Compare
source_pubsub/position/position.go
Outdated
| topicEvent := p.Topics[topic] | ||
| topicEvent.ReplayID = replayID | ||
| topicEvent.ReadTime = time.Now() | ||
| p.Topics[topic] = topicEvent |
There was a problem hiding this comment.
| p.Topics[topic] = topicEvent | |
| p.Topics[topic] = TopicPosition{ | |
| ReplayID: replayID, | |
| ReadTime: time.Now().UTC(), | |
| } |
Is the equivalent.
source_pubsub/position/position.go
Outdated
| p.Topics[topic] = topicEvent | ||
| } else { | ||
| // should never be even reaching this point, something went wrong if we do | ||
| panic(fmt.Errorf("attempting to set replay id - %b on topic %s, topic doesn't exist on position", replayID, topic)) |
There was a problem hiding this comment.
Return an error here, panic is reserved for some rare cases.
source_pubsub/source.go
Outdated
| if err != nil { | ||
| // if using old config and the position isnt parsable | ||
| // assume its in the old position format | ||
| if len(s.config.TopicName) > 0 { |
There was a problem hiding this comment.
the alternative is if s.config.TopicName == "" { return err.. }
source_pubsub/source.go
Outdated
| Str("topics", strings.Join(s.config.TopicNames, ",")). | ||
| Msg("Open Source Connector") | ||
|
|
||
| parsedPositions, err = position.ParseSDKPosition(sdkPos) |
There was a problem hiding this comment.
Can you push this logic in ParseSDKPosition such that position.ParseSDKPosition(sdkPos, s.config.TopicNames) does the conversion internally.
anna-cross
force-pushed
the
source_multiple_topics
branch
from
2eeadfc
to
1235b14
Compare
source_pubsub/source.go
Outdated
| @@ -78,18 +78,10 @@ func (s *Source) Open(ctx context.Context, sdkPos sdk.Position) error { | |||
| Str("topics", strings.Join(s.config.TopicNames, ",")). | |||
| Msg("Open Source Connector") | |||
|
|
|||
| parsedPositions, err = position.ParseSDKPosition(sdkPos) | |||
| parsedPositions, err = position.ParseSDKPosition(sdkPos, s.config.TopicName) | |||
There was a problem hiding this comment.
Do you want to pass all the topics here?
There was a problem hiding this comment.
We only need to pass the old config to process the old config scenario, if multiple topics are being used we parse as normal
There was a problem hiding this comment.
@anna-cross Oh right, but what if you have new config with old position?
There was a problem hiding this comment.
You mean if they are updating the config to use new one? I dont believe we allow that at the moment, if a person is using old config and the pipeline is deployed they cant switch over to new one
There was a problem hiding this comment.
Yes, I'm referring to this in general. If you get the new config with an old position this may not work as expected?
The position which is passed to Open(..) is the last position committed to Conduit, if you start the connector and the position is upgraded but not committed, next time it will come up with the same position. While this may not happen with mdpx right now, it can turn into an edge case later is what I'm saying ;-)
| @@ -65,24 +66,37 @@ type PubSubClient struct { | |||
| codecCache map[string]*goavro.Codec | |||
| unionFields map[string]map[string]struct{} | |||
|
|
|||
| buffer chan sdk.Record | |||
| buffer chan ConnectResponseEvent | |||
| ticker *time.Ticker | |||
There was a problem hiding this comment.
If each topic gets its own consumer, then each will need its own ticker. Since each ticker provides a channel and the channel is drained once per tick, it is unclear how many of the CDC consumers will actually execute a fetch.
Description
We want to add a new connector feature on salesforce source, allowing client to subscribe to multiple topics at once.
The topics subscribe and pull data parallel to each other, but push to same destination. We don't enforce data structure on the source topics, so the destination should be able to accept data with different record structure (this works for redpanda destination but will not work for something like snowflake destination)
Fixes # (issue)
https://github.com/meroxa/mdpx/issues/2515
Quick checks: