Skip to content

Commit

Permalink
introducing eot
Browse files Browse the repository at this point in the history 
Signed-off-by: Yashash H L <[email protected]>
  • Loading branch information
@yhl25
yhl25 committed Sep 8, 2024
1 parent 44a0a17 commit 2506e0d
Show file tree
Hide file tree
Showing 28 changed files with 458 additions and 252 deletions.
2 changes: 0 additions & 2 deletions go.mod
View file
Original file line number Diff line number Diff line change
Expand Up @@ -2,8 +2,6 @@ module github.com/numaproj/numaflow-go

go 1.21

toolchain go1.22.4

require (
github.com/golang/mock v1.6.0
github.com/stretchr/testify v1.8.1
Expand Down
493 changes: 354 additions & 139 deletions pkg/apis/proto/source/v1/source.pb.go
View file

Large diffs are not rendered by default.

22 changes: 22 additions & 0 deletions pkg/apis/proto/source/v1/source.proto
View file
Original file line number Diff line number Diff line change
Expand Up @@ -68,8 +68,30 @@ message ReadResponse {
// e.g. Kafka and Redis Stream message usually include information about the headers.
map<string, string> headers = 5;
}
message Status {
// Code to indicate the status of the response.
enum Code {
SUCCESS = 0;
FAILURE = 1;
}

// Error to indicate the error type. If the code is FAILURE, then the error field will be populated.
enum Error {
UNACKED = 0;
OTHER = 1;
}

// End of transmission flag.
bool eot = 1;
Code code = 2;
Error error = 3;
optional string msg = 4;
}
// Required field holding the result.
Result result = 1;
// Status of the response. Holds the end of transmission flag and the status code.
//
Status status = 2;
}

/*
Expand Down
2 changes: 0 additions & 2 deletions pkg/batchmapper/examples/batchmap_flatmap/go.mod
View file
Original file line number Diff line number Diff line change
Expand Up @@ -2,8 +2,6 @@ module batchmap-flatmap

go 1.21

toolchain go1.22.4

replace github.com/numaproj/numaflow-go => ../../../..

require github.com/numaproj/numaflow-go v0.8.0
Expand Down
2 changes: 0 additions & 2 deletions pkg/mapper/examples/even_odd/go.mod
View file
Original file line number Diff line number Diff line change
Expand Up @@ -2,8 +2,6 @@ module even_odd

go 1.21

toolchain go1.22.4

replace github.com/numaproj/numaflow-go => ../../../..

require github.com/numaproj/numaflow-go v0.8.0
Expand Down
2 changes: 0 additions & 2 deletions pkg/mapper/examples/flatmap/go.mod
View file
Original file line number Diff line number Diff line change
Expand Up @@ -2,8 +2,6 @@ module flatmap

go 1.21

toolchain go1.22.4

replace github.com/numaproj/numaflow-go => ../../../..

require github.com/numaproj/numaflow-go v0.8.0
Expand Down
2 changes: 0 additions & 2 deletions pkg/mapper/examples/forward_message/go.mod
View file
Original file line number Diff line number Diff line change
Expand Up @@ -2,8 +2,6 @@ module forward_message

go 1.21

toolchain go1.22.4

replace github.com/numaproj/numaflow-go => ../../../..

require github.com/numaproj/numaflow-go v0.8.0
Expand Down
2 changes: 0 additions & 2 deletions pkg/mapper/examples/retry/go.mod
View file
Original file line number Diff line number Diff line change
Expand Up @@ -2,8 +2,6 @@ module retry

go 1.21

toolchain go1.22.4

replace github.com/numaproj/numaflow-go => ../../../..

require github.com/numaproj/numaflow-go v0.8.0
Expand Down
2 changes: 0 additions & 2 deletions pkg/mapper/examples/tickgen/go.mod
View file
Original file line number Diff line number Diff line change
Expand Up @@ -2,8 +2,6 @@ module tickgen

go 1.21

toolchain go1.22.4

replace github.com/numaproj/numaflow-go => ../../../..

require github.com/numaproj/numaflow-go v0.8.0
Expand Down
2 changes: 0 additions & 2 deletions pkg/mapstreamer/examples/flatmap_stream/go.mod
View file
Original file line number Diff line number Diff line change
Expand Up @@ -2,8 +2,6 @@ module flatmap_stream

go 1.21

toolchain go1.22.4

replace github.com/numaproj/numaflow-go => ../../../..

require github.com/numaproj/numaflow-go v0.8.0
Expand Down
2 changes: 0 additions & 2 deletions pkg/reducer/examples/counter/go.mod
View file
Original file line number Diff line number Diff line change
Expand Up @@ -2,8 +2,6 @@ module counter

go 1.21

toolchain go1.22.4

replace github.com/numaproj/numaflow-go => ../../../..

require github.com/numaproj/numaflow-go v0.8.0
Expand Down
2 changes: 0 additions & 2 deletions pkg/reducer/examples/sum/go.mod
View file
Original file line number Diff line number Diff line change
Expand Up @@ -2,8 +2,6 @@ module sum

go 1.21

toolchain go1.22.4

replace github.com/numaproj/numaflow-go => ../../../..

require github.com/numaproj/numaflow-go v0.8.0
Expand Down
2 changes: 0 additions & 2 deletions pkg/reducestreamer/examples/counter/go.mod
View file
Original file line number Diff line number Diff line change
Expand Up @@ -2,8 +2,6 @@ module counter

go 1.21

toolchain go1.22.4

replace github.com/numaproj/numaflow-go => ../../../..

require github.com/numaproj/numaflow-go v0.8.0
Expand Down
2 changes: 0 additions & 2 deletions pkg/reducestreamer/examples/sum/go.mod
View file
Original file line number Diff line number Diff line change
Expand Up @@ -2,8 +2,6 @@ module sum

go 1.21

toolchain go1.22.4

replace github.com/numaproj/numaflow-go => ../../../..

require github.com/numaproj/numaflow-go v0.8.0
Expand Down
2 changes: 0 additions & 2 deletions pkg/sessionreducer/examples/counter/go.mod
View file
Original file line number Diff line number Diff line change
Expand Up @@ -2,8 +2,6 @@ module counter

go 1.21

toolchain go1.22.4

replace github.com/numaproj/numaflow-go => ../../../..

require (
Expand Down
2 changes: 0 additions & 2 deletions pkg/sessionreducer/examples/sum/go.mod
View file
Original file line number Diff line number Diff line change
Expand Up @@ -2,8 +2,6 @@ module sum

go 1.21

toolchain go1.22.4

replace github.com/numaproj/numaflow-go => ../../../..

require (
Expand Down
2 changes: 0 additions & 2 deletions pkg/sideinput/examples/simple_sideinput/go.mod
View file
Original file line number Diff line number Diff line change
Expand Up @@ -2,8 +2,6 @@ module simple_sideinput

go 1.21

toolchain go1.22.4

replace github.com/numaproj/numaflow-go => ../../../..

require github.com/numaproj/numaflow-go v0.8.0
Expand Down
2 changes: 0 additions & 2 deletions pkg/sideinput/examples/simple_sideinput/udf/go.mod
View file
Original file line number Diff line number Diff line change
Expand Up @@ -2,8 +2,6 @@ module udf

go 1.21

toolchain go1.22.4

replace github.com/numaproj/numaflow-go => ../../../../..

require (
Expand Down
2 changes: 0 additions & 2 deletions pkg/sinker/examples/fallback/go.mod
View file
Original file line number Diff line number Diff line change
Expand Up @@ -2,8 +2,6 @@ module fallback

go 1.21

toolchain go1.22.4

replace github.com/numaproj/numaflow-go => ../../../..

require github.com/numaproj/numaflow-go v0.8.0
Expand Down
2 changes: 0 additions & 2 deletions pkg/sinker/examples/log/go.mod
View file
Original file line number Diff line number Diff line change
Expand Up @@ -2,8 +2,6 @@ module log_sink

go 1.21

toolchain go1.22.4

replace github.com/numaproj/numaflow-go => ../../../..

require github.com/numaproj/numaflow-go v0.8.0
Expand Down
2 changes: 0 additions & 2 deletions pkg/sinker/examples/redis_sink/go.mod
View file
Original file line number Diff line number Diff line change
Expand Up @@ -2,8 +2,6 @@ module redis-e2e-test-sink

go 1.21

toolchain go1.22.4

replace github.com/numaproj/numaflow-go => ../../../..

require (
Expand Down
5 changes: 1 addition & 4 deletions pkg/sourcer/examples/simple_source/go.mod
View file
Original file line number Diff line number Diff line change
Expand Up @@ -2,8 +2,6 @@ module simple_source

go 1.21

toolchain go1.22.4

replace github.com/numaproj/numaflow-go => ../../../..

require (
Expand All @@ -14,11 +12,10 @@ require (

require (
github.com/davecgh/go-spew v1.1.1 // indirect
github.com/golang/protobuf v1.5.3 // indirect
github.com/kr/text v0.2.0 // indirect
github.com/pmezard/go-difflib v1.0.0 // indirect
github.com/rogpeppe/go-internal v1.12.0 // indirect
golang.org/x/net v0.28.0 // indirect
golang.org/x/net v0.29.0 // indirect
golang.org/x/sys v0.25.0 // indirect
golang.org/x/text v0.18.0 // indirect
google.golang.org/genproto/googleapis/rpc v0.0.0-20240903143218-8af14fe29dc1 // indirect
Expand Down
35 changes: 13 additions & 22 deletions pkg/sourcer/examples/simple_source/go.sum
View file
Original file line number Diff line number Diff line change
Expand Up @@ -2,15 +2,12 @@ github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ3
github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
github.com/golang/protobuf v1.5.0/go.mod h1:FsONVRAS9T7sI+LIUmWTfcYkHO4aIWwzhcaSAoJOfIk=
github.com/golang/protobuf v1.5.3 h1:KhyjKVUg7Usr/dYsdSqoFveMYd5ko72D+zANwlG1mmg=
github.com/golang/protobuf v1.5.3/go.mod h1:XVQd3VNwM+JqD3oG2Ue2ip4fOMUkwXdXDdiuN0vRsmY=
github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
github.com/google/go-cmp v0.5.9 h1:O2Tfq5qg4qc4AmwVlvv0oLiVAGB7enBSJ2x2DqQFi38=
github.com/google/uuid v1.3.0 h1:t6JiXgmwXMjEs8VusXIJk2BXHsn+wx8BZdTaoZ5fu7I=
github.com/google/uuid v1.3.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
github.com/google/go-cmp v0.6.0 h1:ofyhxvXcZhMsU5ulbFiLKl/XBFqE1GSq7atu8tAmTRI=
github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE=
github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk=
github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
Expand All @@ -24,29 +21,23 @@ github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/
github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU=
github.com/stretchr/testify v1.8.1 h1:w7B6lhMri9wdJUVmEZPGGhZzrYTPvgJArz7wNPgYKsk=
github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4=
golang.org/x/net v0.9.0 h1:aWJ/m6xSmxWBx+V0XRHTlrYrPG56jKsLdTFmsSsCzOM=
golang.org/x/net v0.9.0/go.mod h1:d48xBJpPfHeWQsugry2m+kC02ZBRGRgulfHnEXEuWns=
golang.org/x/net v0.28.0 h1:a9JDOJc5GMUJ0+UDqmLT86WiEy7iWyIhz8gz8E4e5hE=
golang.org/x/net v0.28.0/go.mod h1:yqtgsTWOOnlGLG9GFRrK3++bGOUEkNBoHZc8MEDWPNg=
golang.org/x/sys v0.7.0 h1:3jlCCIQZPdOYu1h8BkNvLz8Kgwtae2cagcG/VamtZRU=
golang.org/x/sys v0.7.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/net v0.29.0 h1:5ORfpBpCs4HzDYoodCDBbwHzdR5UrLBZ3sOnUJmFoHo=
golang.org/x/net v0.29.0/go.mod h1:gLkgy8jTGERgjzMic6DS9+SP0ajcu6Xu3Orq/SpETg0=
golang.org/x/sys v0.25.0 h1:r+8e+loiHxRqhXVl6ML1nO3l1+oFoWbnlu2Ehimmi34=
golang.org/x/sys v0.25.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
golang.org/x/text v0.9.0 h1:2sjJmO8cDvYveuX97RDLsxlyUxLl+GHoLxBiRdHllBE=
golang.org/x/text v0.9.0/go.mod h1:e1OnstbJyHTd6l/uOt8jFFHp6TRDWZR/bV3emEE/zU8=
golang.org/x/text v0.18.0 h1:XvMDiNzPAl0jr17s6W9lcaIhGUfUORdGCNsuLmPG224=
golang.org/x/text v0.18.0/go.mod h1:BuEKDfySbSR4drPmRPG/7iBdf8hvFMuRexcpahXilzY=
golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
google.golang.org/genproto/googleapis/rpc v0.0.0-20230525234030-28d5490b6b19 h1:0nDDozoAU19Qb2HwhXadU8OcsiO/09cnTqhUtq2MEOM=
google.golang.org/genproto/googleapis/rpc v0.0.0-20230525234030-28d5490b6b19/go.mod h1:66JfowdXAEgad5O9NnYcsNPLCPZJD++2L9X0PCMODrA=
google.golang.org/genproto/googleapis/rpc v0.0.0-20240903143218-8af14fe29dc1 h1:pPJltXNxVzT4pK9yD8vR9X75DaWYYmLGMsEvBfFQZzQ=
google.golang.org/genproto/googleapis/rpc v0.0.0-20240903143218-8af14fe29dc1/go.mod h1:UqMtugtsSgubUsoxbuAoiCXvqvErP7Gf0so0mK9tHxU=
google.golang.org/grpc v1.57.0 h1:kfzNeI/klCGD2YPMUlaGNT3pxvYfga7smW3Vth8Zsiw=
google.golang.org/grpc v1.57.0/go.mod h1:Sd+9RMTACXwmub0zcNY2c4arhtrbBYD1AUHI/dt16Mo=
google.golang.org/grpc v1.66.0 h1:DibZuoBznOxbDQxRINckZcUvnCEvrW9pcWIE2yF9r1c=
google.golang.org/grpc v1.66.0/go.mod h1:s3/l6xSSCURdVfAnL+TqCNMyTDAGN6+lZeVxnZR128Y=
google.golang.org/protobuf v1.26.0-rc.1/go.mod h1:jlhhOSvTdKEhbULTjvd4ARK9grFBp09yW+WbY/TyQbw=
google.golang.org/protobuf v1.26.0/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQnmE0givc=
google.golang.org/protobuf v1.31.0 h1:g0LDEJHgrBl9N9r17Ru3sqWhkIx2NB67okBHPwC7hs8=
google.golang.org/protobuf v1.31.0/go.mod h1:HV8QOd/L58Z+nl8r43ehVNZIU/HEI6OcFqwMG9pJV4I=
google.golang.org/protobuf v1.34.2 h1:6xV6lTsCfpGD21XK49h7MhtcApnLqkfYgPcdHftf6hg=
google.golang.org/protobuf v1.34.2/go.mod h1:qYOHts0dSfpeUzUFpOMr/WGzszTmLH+DiWniOlNbLDw=
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127 h1:qIbj1fsPNlZgppZ+VLlY7N33q108Sa+fhmuc+sWQYwY=
gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
17 changes: 12 additions & 5 deletions pkg/sourcer/examples/simple_source/impl/simple_source.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@ package impl

import (
"context"
"log"
"strconv"
"sync"
"time"
Expand All @@ -27,8 +28,11 @@ func NewSimpleSource() *SimpleSource {
}

func (s *SimpleSource) Pending(_ context.Context) int64 {
s.lock.Lock()
defer s.lock.Unlock()
log.Println("Number of pending records: ", len(s.toAckSet))
// The simple source always returns zero to indicate there is no pending record.
return 0
return int64(len(s.toAckSet))
}

func (s *SimpleSource) Read(_ context.Context, readRequest sourcesdk.ReadRequest, messageCh chan<- sourcesdk.Message) {
Expand All @@ -42,9 +46,9 @@ func (s *SimpleSource) Read(_ context.Context, readRequest sourcesdk.ReadRequest
// leaving the toAckSet not empty on the UDSource container side.
// In this case, for the next batch read, we should read the data from the last acked offset instead of returning.
// Our built-in Kafka source follows this logic.
if len(s.toAckSet) > 0 {
return
}
//if len(s.toAckSet) > 0 {
// return
//}

// Read the data from the source and send the data to the message channel.
for i := 0; uint64(i) < readRequest.Count(); i++ {
Expand Down Expand Up @@ -72,7 +76,10 @@ func (s *SimpleSource) Read(_ context.Context, readRequest sourcesdk.ReadRequest
}

func (s *SimpleSource) Ack(_ context.Context, request sourcesdk.AckRequest) {
delete(s.toAckSet, deserializeOffset(request.Offset().Value()))
s.lock.Lock()
defer s.lock.Unlock()
offset := deserializeOffset(request.Offset().Value())
delete(s.toAckSet, offset)
}

func (s *SimpleSource) Partitions(_ context.Context) []int32 {
Expand Down
Loading

0 comments on commit 2506e0d

Please sign in to comment.