chore: fix watermark components ownership #963

yhl25 · 2023-08-18T02:57:04Z

related to #900

publisher shouldn't close the store, it should just delete its key from the store, since it doesn't own it and in some cases, stores will be shared across publishers(for example Kafka source)
for source, publishers should be closed inside the forwarder, since creation also happens inside the forwarder
for the map, reduce and sink publishers and fetchers should be closed outside the forwarder because they don't own them and they are shared when there are multiple partitions
introduced BuildProcessorManagers, BuildToVertexWatermarkStores and BuildPublishersFromStores helper functions
removed io.Closer from Fetcher and UXFetcher interface.

also fixes #962

Signed-off-by: Yashash H L <yashashhl25@gmail.com>

whynowy · 2023-08-18T06:15:44Z

pkg/sinks/sink.go

+	// close the wm stores, since the publisher and fetcher are closed
+	// since we created the stores, we can close them
+	for _, wmStore := range wmStores {
+		wmStore.HeartbeatStore().Close()


Add a Close() function to WatermarkStore interface ?

whynowy · 2023-08-18T06:16:45Z

pkg/sinks/sink.go

+			names := u.VertexInstance.Vertex.GetToBuffers()
+			if u.VertexInstance.Vertex.IsASink() {
+				// Sink has no to buffers, we use the vertex name as the buffer writer name.
+				names = append(names, u.VertexInstance.Vertex.Spec.Name)


Already know it's a sink, then the line above is not needed?

whynowy · 2023-08-18T06:25:57Z

pkg/udf/reduce_udf.go

+	log.Info("SIGTERM, exiting...")
+	wg.Wait()
+
+	// close the watermark fetcher and publisher since we created them
 	err = fetchWatermark.Close()
 	if err != nil {
 		log.Error("Failed to close the watermark fetcher", zap.Error(err))


log.Errorw()

whynowy · 2023-08-18T06:26:24Z

pkg/udf/map_udf.go

@@ -214,18 +245,29 @@ func (u *MapUDFProcessor) Start(ctx context.Context) error {
 	// wait for all the forwarders to exit
 	finalWg.Wait()

-	// Close the watermark fetcher and publisher
+	// close the watermark fetcher and publisher since we created them
 	err = fetchWatermark.Close()
 	if err != nil {
 		log.Error("Failed to close the watermark fetcher", zap.Error(err))


pkg/watermark/fetch/edge_fetcher.go

Signed-off-by: Yashash H L <yashashhl25@gmail.com>

KeranYang

Thank you very much for making this change. The ownership among fetcher, publisher, processor manger and kv stores are much clearer now. +1

pkg/isbsvc/jetstream_service.go

Signed-off-by: Vigith Maurice <vigith@gmail.com>

whynowy · 2023-08-18T15:57:23Z

pkg/sinks/sink.go

+	for _, publisher := range publishWatermark {
+		err = publisher.Close()
+		if err != nil {
+			log.Error("Failed to close the watermark publisher", zap.Error(err))


Suggested change

log.Error("Failed to close the watermark publisher", zap.Error(err))

log.Errorw("Failed to close the watermark publisher", zap.Error(err))

whynowy · 2023-08-18T16:03:11Z

pkg/udf/map_udf.go

-			return err
+		if u.VertexInstance.Vertex.Spec.Watermark.Disabled {
+			names := u.VertexInstance.Vertex.GetToBuffers()
+			if u.VertexInstance.Vertex.IsASink() {


whynowy · 2023-08-18T16:03:40Z

pkg/udf/reduce_udf.go

-			return err
+		if u.VertexInstance.Vertex.Spec.Watermark.Disabled {
+			names := u.VertexInstance.Vertex.GetToBuffers()
+			if u.VertexInstance.Vertex.IsASink() {


whynowy · 2023-08-18T16:09:05Z

pkg/sinks/sink.go

+		if u.VertexInstance.Vertex.Spec.Watermark.Disabled {
+			names := u.VertexInstance.Vertex.GetToBuffers()
+			// sink has no to buffers, so we use the vertex name to publish the watermark
+			names = append(names, u.VertexInstance.Vertex.Spec.Name)


names := []string{u.VertexInstance.Vertex.Spec.Name}

jy4096 · 2023-08-18T17:19:04Z

pkg/sources/source.go

 		if err != nil {
 			return err
 		}
+
+		// build publisher stores for source (we publish twice for source)


could you add a comment here to explain why we need two sets of publisher stores?..

Signed-off-by: Derek Wang <whynowy@gmail.com>

vigith

We need to rename these, using BuildProcessorManagers to build fetchers, and BuildToVertexWatermarkStores to build publishers does not make sense.

The code should be self-documenting, as of now, it is quite difficult for a casual reader to understand the code.

Fetch

processorManagers, err = jetstream.BuildProcessorManagers(ctx, u.VertexInstance, natsClientPool.NextAvailableClient())

// create watermark fetcher using processor managers
fetchWatermark = fetch.NewEdgeFetcherSet(ctx, u.VertexInstance, processorManagers)

Publish

// create watermark stores
wmStores, err = jetstream.BuildToVertexWatermarkStores(ctx, u.VertexInstance, natsClientPool.NextAvailableClient())


// create watermark publisher using watermark stores
publishWatermark = jetstream.BuildPublishersFromStores(ctx, u.VertexInstance, wmStores)

readers, writers, err = buildJetStreamBufferIO(ctx, u.VertexInstance, natsClientPool)

fix wm components ownership

9a1bab0

Signed-off-by: Yashash H L <yashashhl25@gmail.com>

yhl25 requested a review from jy4096 August 18, 2023 02:57

yhl25 added 2 commits August 18, 2023 08:46

fix lint and source

ef2bf10

Signed-off-by: Yashash H L <yashashhl25@gmail.com>

fix panic

9a06c43

Signed-off-by: Yashash H L <yashashhl25@gmail.com>

whynowy reviewed Aug 18, 2023

View reviewed changes

yhl25 added 2 commits August 18, 2023 12:50

remove closer from fetcher interface

71e0431

Signed-off-by: Yashash H L <yashashhl25@gmail.com>

review comments

a482c8c

Signed-off-by: Yashash H L <yashashhl25@gmail.com>

yhl25 requested a review from whynowy August 18, 2023 07:43

yhl25 marked this pull request as ready for review August 18, 2023 07:43

yhl25 requested a review from vigith as a code owner August 18, 2023 07:43

KeranYang reviewed Aug 18, 2023

View reviewed changes

pkg/isbsvc/jetstream_service.go Show resolved Hide resolved

vigith added 2 commits August 18, 2023 08:42

chore: minor refactor

6937209

Signed-off-by: Vigith Maurice <vigith@gmail.com>

fix: #962

7310372

Signed-off-by: Vigith Maurice <vigith@gmail.com>

whynowy reviewed Aug 18, 2023

View reviewed changes

jy4096 reviewed Aug 18, 2023

View reviewed changes

whynowy added 2 commits August 18, 2023 14:13

Merge remote-tracking branch 'upstream/main' into wm-store

0360f88

Signed-off-by: Derek Wang <whynowy@gmail.com>

.

478bc8a

Signed-off-by: Derek Wang <whynowy@gmail.com>

whynowy approved these changes Aug 18, 2023

View reviewed changes

vigith approved these changes Aug 19, 2023

View reviewed changes

vigith merged commit ed53342 into main Aug 19, 2023
15 checks passed

vigith deleted the wm-store branch August 19, 2023 00:35

vigith mentioned this pull request Aug 19, 2023

use consistent name for creating fetcher and publisher #969

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

chore: fix watermark components ownership #963

chore: fix watermark components ownership #963

yhl25 commented Aug 18, 2023 •

edited by vigith

Loading

whynowy Aug 18, 2023

yhl25 Aug 18, 2023

whynowy Aug 18, 2023

yhl25 Aug 18, 2023

whynowy Aug 18, 2023

whynowy Aug 18, 2023

KeranYang left a comment

whynowy Aug 18, 2023

whynowy Aug 18, 2023

whynowy Aug 18, 2023

whynowy Aug 18, 2023

jy4096 Aug 18, 2023

vigith left a comment

	log.Error("Failed to close the watermark publisher", zap.Error(err))
	log.Errorw("Failed to close the watermark publisher", zap.Error(err))

chore: fix watermark components ownership #963

chore: fix watermark components ownership #963

Conversation

yhl25 commented Aug 18, 2023 • edited by vigith 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

KeranYang left a 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

vigith left a comment

Choose a reason for hiding this comment

Fetch

Publish

yhl25 commented Aug 18, 2023 •

edited by vigith

Loading