feat(planner): plan sink batch as an op

Signed-off-by: Jiyong Huang <[email protected]>
lf-edge · Mar 29, 2024 · 06b618e · 06b618e
1 parent d2b8eb4
commit 06b618e
Show file tree

Hide file tree

Showing 8 changed files with 273 additions and 424 deletions.
diff --git a/internal/io/sink/send_manager.go b/internal/io/sink/send_manager.go
diff --git a/internal/io/sink/send_manager_test.go b/internal/io/sink/send_manager_test.go
diff --git a/internal/topo/node/batch_op_test.go b/internal/topo/node/batch_op_test.go
@@ -68,7 +68,6 @@ func TestRun(t *testing.T) {
 	}
 	mc := conf.Clock.(*clock.Mock)
 	for i, tc := range testcases {
-		mc.Set(mc.Now())
 		t.Run(fmt.Sprintf("testcase %d", i), func(t *testing.T) {
 			op, err := NewBatchOp("test", &api.RuleOption{BufferLength: 10, SendError: true}, tc.batchSize, tc.lingerInterval)
 			if len(tc.err) > 0 {