Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Replace panic with error message when not declaring a source node in .yml config #1695

Merged
merged 6 commits into from
Jul 30, 2024
Merged

Replace panic with error message when not declaring a source node in .yml config #1695

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
136f6a6
move append for DLQhandler node to only occur if there are source nod…
AdamHaffar Jul 4, 2024
1b54d6e
unit test
AdamHaffar Jul 4, 2024
069519d
fix test cases for unit test and add no destination test
AdamHaffar Jul 10, 2024
a84506b
Merge branch 'main' into adam/better-error-message
AdamHaffar Jul 10, 2024
cba99ce
Merge branch 'main' into adam/better-error-message
lovromazgon Jul 30, 2024
4f33ce5
fix tests
lovromazgon Jul 30, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 3 additions & 1 deletion pkg/pipeline/lifecycle.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -352,7 +352,6 @@ func (s *Service) buildSourceNodes(
if err != nil {
return nil, err
}
nodes = append(nodes, dlqHandlerNode)

for _, connID := range pl.ConnectorIDs {
instance, err := connFetcher.Get(ctx, connID)
Expand Down Expand Up @@ -391,6 +390,9 @@ func (s *Service) buildSourceNodes(
nodes = append(nodes, procNodes...)
}

if len(nodes) != 0 {
nodes = append(nodes, dlqHandlerNode)
}
return nodes, nil
}

Expand Down
97 changes: 96 additions & 1 deletion pkg/pipeline/lifecycle_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -87,10 +87,10 @@ func TestServiceLifecycle_buildNodes(t *testing.T) {
is.NoErr(err)

want := []stream.Node{
&stream.DLQHandlerNode{},
&stream.SourceNode{},
&stream.SourceAckerNode{},
&stream.MetricsNode{},
&stream.DLQHandlerNode{},
&stream.FaninNode{},
&stream.FanoutNode{},
&stream.MetricsNode{},
Expand Down Expand Up @@ -125,6 +125,101 @@ func TestServiceLifecycle_buildNodes(t *testing.T) {
}
}

func TestService_buildNodes_NoSourceNode(t *testing.T) {
is := is.New(t)
ctx, killAll := context.WithCancel(context.Background())
defer killAll()
ctrl := gomock.NewController(t)
logger := log.New(zerolog.Nop())
db := &inmemory.DB{}
persister := connector.NewPersister(logger, db, time.Second, 3)

ps := NewService(logger, db)

wantErr := "can't build pipeline without any source connectors"

destination := dummyDestination(persister)
dlq := dummyDestination(persister)
pl := &Instance{
ID: uuid.NewString(),
Config: Config{Name: "test-pipeline"},
Status: StatusUserStopped,
DLQ: DLQ{
Plugin: dlq.Plugin,
Settings: map[string]string{},
WindowSize: 3,
WindowNackThreshold: 2,
},
ConnectorIDs: []string{destination.ID},
}

got, err := ps.buildNodes(
ctx,
testConnectorFetcher{
destination.ID: destination,
testDLQID: dlq,
},
testProcessorFetcher{},
testPluginFetcher{
destination.Plugin: pmock.NewDispenser(ctrl),
dlq.Plugin: pmock.NewDispenser(ctrl),
},
pl,
)

is.True(err != nil)
is.Equal(err.Error(), wantErr)
hariso marked this conversation as resolved.
Show resolved Hide resolved
is.Equal(got, nil)
}

func TestService_buildNodes_NoDestinationNode(t *testing.T) {
is := is.New(t)
ctx, killAll := context.WithCancel(context.Background())
defer killAll()
ctrl := gomock.NewController(t)
logger := log.New(zerolog.Nop())
db := &inmemory.DB{}
persister := connector.NewPersister(logger, db, time.Second, 3)

ps := NewService(logger, db)

wantErr := "can't build pipeline without any destination connectors"

source := dummySource(persister)
dlq := dummyDestination(persister)

pl := &Instance{
ID: uuid.NewString(),
Config: Config{Name: "test-pipeline"},
Status: StatusUserStopped,
DLQ: DLQ{
Plugin: dlq.Plugin,
Settings: map[string]string{},
WindowSize: 3,
WindowNackThreshold: 2,
},
ConnectorIDs: []string{source.ID},
}

got, err := ps.buildNodes(
ctx,
testConnectorFetcher{
source.ID: source,
testDLQID: dlq,
},
testProcessorFetcher{},
testPluginFetcher{
source.Plugin: pmock.NewDispenser(ctrl),
dlq.Plugin: pmock.NewDispenser(ctrl),
},
pl,
)

is.True(err != nil)
is.Equal(err.Error(), wantErr)
is.Equal(got, nil)
}

func TestServiceLifecycle_PipelineSuccess(t *testing.T) {
is := is.New(t)
ctx, killAll := context.WithCancel(context.Background())
Expand Down