refactor: generalize watermark fetching as an interface of ISB service. Fixes #252 #263
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Signed-off-by: Keran Yang <[email protected]>
KeranYang
changed the title
Signed-off-by: Keran Yang <[email protected]>
Signed-off-by: Keran Yang <[email protected]>
Signed-off-by: Keran Yang <[email protected]>
|
On issue #252 , there is also the following request
@whynowy Hey Derek, regarding the request above, is it already resolved by https://github.com/numaproj/numaflow/pull/238/files? |
whynowy
reviewed
Oct 25, 2022
pkg/isbsvc/jetstream_service.go
Outdated
| @@ -21,6 +25,21 @@ type jetStreamSvc struct { | |||
| js *jsclient.JetStreamContext | |||
| } | |||
|
|
|||
| func (jss *jetStreamSvc) CreateWatermarkFetcher(ctx context.Context, pipelineName string, bufferName string) (fetch.Fetcher, error) { | |||
There was a problem hiding this comment.
pipelineName is already a property of the struct, not needed in the function signature.
pkg/isbsvc/jetstream_service.go
Outdated
| @@ -21,6 +25,21 @@ type jetStreamSvc struct { | |||
| js *jsclient.JetStreamContext | |||
| } | |||
|
|
|||
| func (jss *jetStreamSvc) CreateWatermarkFetcher(ctx context.Context, pipelineName string, bufferName string) (fetch.Fetcher, error) { | |||
| hbBucketName := JetStreamProcessorBucket(pipelineName, bufferName) | |||
| hbWatch, err := jetstream.NewKVJetStreamKVWatch(ctx, pipelineName, hbBucketName, jsclient.NewInClusterJetStreamClient()) | |||
There was a problem hiding this comment.
Use the jetstream client in the struct.
pkg/isbsvc/jetstream_service.go
Outdated
| return nil, err | ||
| } | ||
| otBucketName := JetStreamOTBucket(pipelineName, bufferName) | ||
| otWatch, err := jetstream.NewKVJetStreamKVWatch(ctx, pipelineName, otBucketName, jsclient.NewInClusterJetStreamClient()) |
vigith
changed the title
Signed-off-by: Keran Yang <[email protected]>
…aflow into generic-watermark-fetching
vigith
previously approved these changes
Oct 25, 2022
vigith
reviewed
Oct 25, 2022
pkg/isbsvc/jetstream_service.go
Outdated
| hbWatch, err := jetstream.NewKVJetStreamKVWatch(ctx, pipelineName, hbBucketName, jsclient.NewInClusterJetStreamClient()) | ||
| func (jss *jetStreamSvc) CreateWatermarkFetcher(ctx context.Context, bufferName string) (fetch.Fetcher, error) { | ||
| hbBucketName := JetStreamProcessorBucket(jss.pipelineName, bufferName) | ||
| hbWatch, err := jetstream.NewKVJetStreamKVWatch(ctx, jss.pipelineName, hbBucketName, jsclient.NewInClusterJetStreamClient()) |
There was a problem hiding this comment.
why do you need a new client, can't we get it from jss.jsClient
There was a problem hiding this comment.
Good catch. There are two types of jsClient - defaultJetStreamClient and inClusterJetStreamClient. Searching across the code base I see defaultJetStreamClient is only used in unit test files so I think it's safe to get jsClient from jss.jsClient. Will address.
Signed-off-by: Keran Yang <[email protected]>
whynowy
reviewed
Oct 25, 2022
pkg/isbsvc/jetstream_service.go
Outdated
| @@ -21,6 +25,21 @@ type jetStreamSvc struct { | |||
| js *jsclient.JetStreamContext | |||
| } | |||
|
|
|||
| func (jss *jetStreamSvc) CreateWatermarkFetcher(ctx context.Context, bufferName string) (fetch.Fetcher, error) { | |||
There was a problem hiding this comment.
Nit: move this function to the place after NewISBJetStreamSvc(, let it sit together with other interface functions.
Signed-off-by: Keran Yang <[email protected]>
vigith
approved these changes
Oct 25, 2022
whynowy
pushed a commit
that referenced
this pull request
Oct 27, 2022
Fixes #252 (#263) ### Description Fixes #252 This change is to NOT hardcode JetStream as ISB service when fetching watermarks. When the ISB service is Redis, we use noop KV watchers(for both heartbeat and offset timeline) for fetching watermarks. Signed-off-by: Keran Yang <[email protected]>
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Description
Fixes #252 This change is to NOT hardcode JetStream as ISB service when fetching watermarks. When the ISB service is Redis, we use noop KV watchers(for both heartbeat and offset timeline) for fetching watermarks.
Testing
Before the change, if we use Redis as ISB service, when creating a simple-pipeline, the
simple-pipeline-daemon-xxxpod crashes with following error:And numaflow UI shows empty page for simple-pipeline.
After the change, when using Redis as ISB, we can start the
simple-pipeline-daemon-xxxpod successfully. On the UI, we show default watermark -1 across ALL vertices.Signed-off-by: Keran Yang [email protected]
Explain what this PR does.