Repository: vanus-labs/vanus
Pull request #432 (2023-02-13 12:07:07)

fix: blocking consumption due to unsafe concurrent issue
<!--

Thank you for contributing to Vanus!

PR Title Format: 

feat/fix/refactor/docs/...: what's changed

Note: see https://github.com/linkall-labs/vanus/blob/main/CONTRIBUTING.md to know details about title format
-->

### What problem does this PR solve?
<!--

Please create an issue first to describe the problem

There MUST be one line starting with "Issue Number:  ".

-->

Issue Number: close #xxx

### Problem Summary
fix blocking consumption due to unsafe concurrent issue
### What is changed and how does it work?

### Check List

<!-- At least one of them must be included. -->

#### Tests

- [x] Unit test
- [ ] Integration test
- [x] Manual test (add detailed scripts or steps below)
- [ ] No code



Patch:
diff --git a/internal/controller/eventbus/eventlog/eventlog.go b/internal/controller/eventbus/eventlog/eventlog.go
index 301a9e5..38045ed 100644
--- a/internal/controller/eventbus/eventlog/eventlog.go
+++ b/internal/controller/eventbus/eventlog/eventlog.go
@@ -52,14 +52,12 @@ type Manager interface {
 	AcquireEventLog(ctx context.Context, eventbusID vanus.ID, eventbusName string) (*metadata.Eventlog, error)
 	GetEventLog(ctx context.Context, id vanus.ID) *metadata.Eventlog
 	DeleteEventlog(ctx context.Context, id vanus.ID)
-	GetEventLogSegmentList(elID vanus.ID) []*Segment
-	GetAppendableSegment(ctx context.Context, eli *metadata.Eventlog,
-		num int) ([]*Segment, error)
-	UpdateSegment(ctx context.Context, m map[string][]Segment)
-	GetSegmentByBlockID(block *metadata.Block) (*Segment, error)
 	GetBlock(id vanus.ID) *metadata.Block
-	GetSegment(id vanus.ID) *Segment
 	UpdateSegmentReplicas(ctx context.Context, segID vanus.ID, term uint64) error
+	GetEventLogSegmentList(elID vanus.ID) []Segment
+	GetAppendableSegment(ctx context.Context, eli *metadata.Eventlog, num int) ([]Segment, error)
+	UpdateSegment(ctx context.Context, m map[string][]Segment)
+	GetSegmentByBlockID(block *metadata.Block) (Segment, error)
 }
 
 var mgr = &eventlogManager{
@@ -135,15 +133,11 @@ func (mgr *eventlogManager) Run(ctx context.Context, kvClient kv.Client, startTa
 			return err2
 		}
 		mgr.eventLogMap.Store(elMD.ID.Key(), el)
-		ptr := el.segmentList.Front()
-		if ptr != nil {
-			seg, _ := ptr.Value.(*Segment)
-			for seg != nil {
-				mgr.globalSegmentMap.Store(seg.ID.Key(), seg)
-				for _, v := range seg.Replicas.Peers {
-					mgr.globalBlockMap.Store(v.ID.Key(), v)
-				}
-				seg = el.nextOf(seg)
+
+		for _, ptr := range el.getAllSegments() {
+			mgr.globalSegmentMap.Store(ptr.ID.Key(), ptr)
+			for _, v := range ptr.Replicas.Peers {
+				mgr.globalBlockMap.Store(v.ID.Key(), v)
 			}
 		}
 	}
@@ -250,9 +244,8 @@ func (mgr *eventlogManager) DeleteEventlog(ctx context.Context, id vanus.ID) {
 }
 
 func (mgr *eventlogManager) GetAppendableSegment(ctx context.Context,
-	eli *metadata.Eventlog, num int) ([]*Segment, error) {
-	result := make([]*Segment, 0)
-
+	eli *metadata.Eventlog, num int) ([]Segment, error) {
+	result := make([]Segment, 0, num)
 	if eli == nil || num == 0 {
 		return result, nil
 	}
@@ -286,14 +279,13 @@ func (mgr *eventlogManager) GetAppendableSegment(ctx context.Context,
 		s = el.currentAppendableSegment()
 	}
 
-	for len(result) < num && s != nil {
-		result = append(result, s)
-		s = el.nextOf(s)
-	}
-	return result, nil
+	return el.listOfRight(s, true), nil
 }
 
 func (mgr *eventlogManager) UpdateSegment(ctx context.Context, m map[string][]Segment) {
+	mgr.mutex.Lock()
+	defer mgr.mutex.Unlock()
+
 	// iterate eventlog
 	for eventlogID, segments := range m {
 		v, exist := mgr.eventLogMap.Load(eventlogID)
@@ -336,17 +328,20 @@ func (mgr *eventlogManager) UpdateSegment(ctx context.Context, m map[string][]Se
 	}
 }
 
-func (mgr *eventlogManager) GetEventLogSegmentList(elID vanus.ID) []*Segment {
-	result := make([]*Segment, 0)
+func (mgr *eventlogManager) GetEventLogSegmentList(elID vanus.ID) []Segment {
+	result := make([]Segment, 0)
 	v, exist := mgr.eventLogMap.Load(elID.Key())
 	if !exist {
 		return result
 	}
+
 	el, _ := v.(*eventlog)
-	s := el.head()
-	for s != nil {
-		result = append(result, s)
-		s = el.nextOf(s)
+	list := el.getAllSegments()
+	el.lock()
+	defer el.unlock()
+
+	for _, v := range list {
+		result = append(result, v.Copy())
 	}
 	return result
 }
@@ -359,7 +354,7 @@ func (mgr *eventlogManager) GetBlock(id vanus.ID) *metadata.Block {
 	return v.(*metadata.Block)
 }
 
-func (mgr *eventlogManager) GetSegment(id vanus.ID) *Segment {
+func (mgr *eventlogManager) getSegment(id vanus.ID) *Segment {
 	v, exist := mgr.globalSegmentMap.Load(id.Key())
 	if !exist {
 		return nil
@@ -373,7 +368,7 @@ func (mgr *eventlogManager) UpdateSegmentReplicas(ctx context.Context, leaderID
 		return errors.ErrBlockNotFound
 	}
 
-	seg := mgr.GetSegment(blk.SegmentID)
+	seg := mgr.getSegment(blk.SegmentID)
 	if seg == nil {
 		return errors.ErrSegmentNotFound
 	}
@@ -382,6 +377,13 @@ func (mgr *eventlogManager) UpdateSegmentReplicas(ctx context.Context, leaderID
 		return nil
 	}
 
+	el := mgr.getEventLog(seg.EventLogID)
+	if el == nil {
+		return errors.ErrEventLogNotFound
+	}
+	el.lock()
+	defer el.unlock()
+
 	seg.Replicas.Leader = leaderID.Uint64()
 	seg.Replicas.Term = term
 	data, _ := json.Marshal(seg)
@@ -396,20 +398,22 @@ func (mgr *eventlogManager) UpdateSegmentReplicas(ctx context.Context, leaderID
 	return nil
 }
 
-func (mgr *eventlogManager) GetSegmentByBlockID(block *metadata.Block) (*Segment, error) {
+func (mgr *eventlogManager) GetSegmentByBlockID(block *metadata.Block) (Segment, error) {
 	v, exist := mgr.eventLogMap.Load(block.EventlogID.Key())
 	if !exist {
-		return nil, errors.ErrEventLogNotFound
+		return Segment{}, errors.ErrEventLogNotFound
 	}
 	el, _ := v.(*eventlog)
-	return el.get(block.SegmentID), nil
+	el.rLock()
+	defer el.rUnlock()
+	return el.get(block.SegmentID).Copy(), nil
 }
 
 func (mgr *eventlogManager) stop() {
 	mgr.cancel()
 }
 
-func (mgr *eventlogManager) getSegmentTopology(_ context.Context, seg *Segment) map[uint64]string {
+func (mgr *eventlogManager) getSegmentTopology(_ context.Context, seg Segment) map[uint64]string {
 	var addrs = map[uint64]string{}
 	for _, v := range seg.Replicas.Peers {
 		ins := mgr.volMgr.GetVolumeInstanceByID(v.VolumeID)
@@ -742,7 +746,7 @@ func (mgr *eventlogManager) createSegment(ctx context.Context, el *eventlog) (*S
 	_, err = srv.GetClient().ActivateSegment(ctx, &segment.ActivateSegmentRequest{
 		EventLogId:     seg.EventLogID.Uint64(),
 		ReplicaGroupId: seg.Replicas.ID.Uint64(),
-		Replicas:       mgr.getSegmentTopology(ctx, seg),
+		Replicas:       mgr.getSegmentTopology(ctx, *seg),
 	})
 
 	if err != nil {
@@ -945,12 +949,8 @@ func (el *eventlog) appendableSegmentNumber() int {
 	if cur == nil {
 		return 0
 	}
-	count := 0
-	for cur != nil {
-		count++
-		cur = el.nextOf(cur)
-	}
-	return count
+
+	return len(el.listOfRight(cur, true))
 }
 
 func (el *eventlog) currentAppendableSegment() *Segment {
@@ -959,6 +959,7 @@ func (el *eventlog) currentAppendableSegment() *Segment {
 	if el.size() == 0 {
 		return nil
 	}
+
 	if el.writePtr == nil {
 		head := el.segmentList.Front()
 		for head != nil {
@@ -989,6 +990,7 @@ func (el *eventlog) currentAppendableSegment() *Segment {
 func (el *eventlog) add(ctx context.Context, seg *Segment) error {
 	el.mutex.Lock()
 	defer el.mutex.Unlock()
+
 	if !seg.isReady() {
 		return errors.ErrInvalidSegment
 	}
@@ -1091,37 +1093,69 @@ func (el *eventlog) size() int {
 	return el.segmentList.Len()
 }
 
-func (el *eventlog) nextOf(seg *Segment) *Segment {
-	if seg == nil {
-		return nil
-	}
+func (el *eventlog) getAllSegments() []*Segment {
 	el.mutex.RLock()
 	defer el.mutex.RUnlock()
 
-	node := el.segmentList.Get(seg.ID.Uint64())
-	if node == nil {
-		return nil
+	segs := make([]*Segment, 0)
+
+	if el.size() == 0 {
+		return segs
 	}
-	next := node.Next()
-	if next == nil {
-		return nil
+	ptr := el.segmentList.Front()
+	for ptr != nil {
+		next, _ := ptr.Value.(*Segment)
+		segs = append(segs, next)
+		ptr = ptr.Next()
 	}
-	return next.Value.(*Segment)
+	return segs
 }
 
-func (el *eventlog) previousOf(seg *Segment) *Segment {
+func (el *eventlog) listOfRight(seg *Segment, includeSelf bool) []Segment {
+	el.mutex.RLock()
+	defer el.mutex.RUnlock()
+	segs := make([]Segment, 0)
 	if seg == nil {
-		return nil
+		return segs
+	}
+	if includeSelf {
+		segs = append(segs, *seg)
 	}
+	node := el.segmentList.Get(seg.ID.Uint64())
+	if node == nil {
+		return segs
+	}
+	v := node.Next()
+	for v != nil {
+		next, _ := v.Value.(*Segment)
+		segs = append(segs, *next)
+		v = v.Next()
+	}
+
+	return segs
+}
+
+func (el *eventlog) listOfPrevious(seg *Segment) []*Segment { //nolint:unused // ok
 	el.mutex.RLock()
 	defer el.mutex.RUnlock()
-
+	segs := make([]*Segment, 0)
 	node := el.segmentList.Get(seg.ID.Uint64())
-	prev := node.Prev()
-	if prev == nil {
-		return nil
+	if node == nil {
+		return segs
+	}
+	v := node.Prev()
+	for v != nil {
+		next, _ := v.Value.(*Segment)
+		segs = append(segs, next)
+		v = v.Prev()
+	}
+
+	// reverse slice
+	for i, j := 0, len(segs)-1; i < j; i, j = i+1, j-1 {
+		segs[i], segs[j] = segs[j], segs[i]
 	}
-	return prev.Value.(*Segment)
+
+	return segs
 }
 
 func (el *eventlog) deleteHead(ctx context.Context) error {
@@ -1192,3 +1226,11 @@ func (el *eventlog) lock() {
 func (el *eventlog) unlock() {
 	el.mutex.Unlock()
 }
+
+func (el *eventlog) rLock() {
+	el.mutex.RLock()
+}
+
+func (el *eventlog) rUnlock() {
+	el.mutex.RUnlock()
+}
diff --git a/internal/controller/eventbus/eventlog/mock_eventlog.go b/internal/controller/eventbus/eventlog/mock_eventlog.go
index bcac927..aeae26f 100644
--- a/internal/controller/eventbus/eventlog/mock_eventlog.go
+++ b/internal/controller/eventbus/eventlog/mock_eventlog.go
@@ -65,10 +65,10 @@ func (mr *MockManagerMockRecorder) DeleteEventlog(ctx, id interface{}) *gomock.C
 }
 
 // GetAppendableSegment mocks base method.
-func (m *MockManager) GetAppendableSegment(ctx context.Context, eli *metadata.Eventlog, num int) ([]*Segment, error) {
+func (m *MockManager) GetAppendableSegment(ctx context.Context, eli *metadata.Eventlog, num int) ([]Segment, error) {
 	m.ctrl.T.Helper()
 	ret := m.ctrl.Call(m, "GetAppendableSegment", ctx, eli, num)
-	ret0, _ := ret[0].([]*Segment)
+	ret0, _ := ret[0].([]Segment)
 	ret1, _ := ret[1].(error)
 	return ret0, ret1
 }
@@ -108,10 +108,10 @@ func (mr *MockManagerMockRecorder) GetEventLog(ctx, id interface{}) *gomock.Call
 }
 
 // GetEventLogSegmentList mocks base method.
-func (m *MockManager) GetEventLogSegmentList(elID vanus.ID) []*Segment {
+func (m *MockManager) GetEventLogSegmentList(elID vanus.ID) []Segment {
 	m.ctrl.T.Helper()
 	ret := m.ctrl.Call(m, "GetEventLogSegmentList", elID)
-	ret0, _ := ret[0].([]*Segment)
+	ret0, _ := ret[0].([]Segment)
 	return ret0
 }
 
@@ -121,25 +121,11 @@ func (mr *MockManagerMockRecorder) GetEventLogSegmentList(elID interface{}) *gom
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetEventLogSegmentList", reflect.TypeOf((*MockManager)(nil).GetEventLogSegmentList), elID)
 }
 
-// GetSegment mocks base method.
-func (m *MockManager) GetSegment(id vanus.ID) *Segment {
-	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "GetSegment", id)
-	ret0, _ := ret[0].(*Segment)
-	return ret0
-}
-
-// GetSegment indicates an expected call of GetSegment.
-func (mr *MockManagerMockRecorder) GetSegment(id interface{}) *gomock.Call {
-	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetSegment", reflect.TypeOf((*MockManager)(nil).GetSegment), id)
-}
-
 // GetSegmentByBlockID mocks base method.
-func (m *MockManager) GetSegmentByBlockID(block *metadata.Block) (*Segment, error) {
+func (m *MockManager) GetSegmentByBlockID(block *metadata.Block) (Segment, error) {
 	m.ctrl.T.Helper()
 	ret := m.ctrl.Call(m, "GetSegmentByBlockID", block)
-	ret0, _ := ret[0].(*Segment)
+	ret0, _ := ret[0].(Segment)
 	ret1, _ := ret[1].(error)
 	return ret0, ret1
 }
diff --git a/internal/controller/eventbus/eventlog/segment.go b/internal/controller/eventbus/eventlog/segment.go
index c02c4fb..112fa20 100644
--- a/internal/controller/eventbus/eventlog/segment.go
+++ b/internal/controller/eventbus/eventlog/segment.go
@@ -143,7 +143,7 @@ type ReplicaGroup struct {
 	DestroyAt time.Time                  `json:"destroy_at"`
 }
 
-func Convert2ProtoSegment(ctx context.Context, ins ...*Segment) []*metapb.Segment {
+func Convert2ProtoSegment(ctx context.Context, ins ...Segment) []*metapb.Segment {
 	segs := make([]*metapb.Segment, len(ins))
 	for idx := 0; idx < len(ins); idx++ {
 		seg := ins[idx]
diff --git a/internal/primitive/transform/action/strings/split_with_delimiter.go b/internal/primitive/transform/action/strings/split_with_delimiter.go
new file mode 100644
index 0000000..ac05f4d
--- /dev/null
+++ b/internal/primitive/transform/action/strings/split_with_delimiter.go
@@ -0,0 +1,44 @@
+// Copyright 2023 Linkall Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package strings
+
+import (
+	"github.com/linkall-labs/vanus/internal/primitive/transform/action"
+	"github.com/linkall-labs/vanus/internal/primitive/transform/arg"
+	"github.com/linkall-labs/vanus/internal/primitive/transform/common"
+	"github.com/linkall-labs/vanus/internal/primitive/transform/function"
+)
+
+type splitWithDelimiterAction struct {
+	action.FunctionAction
+}
+
+// NewSplitWithDelimiterAction ["split_with_delimiter","sourceJsonPath", "delimiter", "targetJsonPath"].
+func NewSplitWithDelimiterAction() action.Action {
+	a := &splitWithDelimiterAction{}
+	a.CommonAction = action.CommonAction{
+		ActionName: "SPLIT_WITH_DELIMITER",
+		FixedArgs:  []arg.TypeList{arg.EventList, []arg.Type{arg.Constant}, []arg.Type{arg.EventData}},
+		Fn:         function.SplitWithSepFunction,
+	}
+	return a
+}
+
+func (a *splitWithDelimiterAction) Init(args []arg.Arg) error {
+	a.TargetArg = args[2]
+	a.Args = args[:2]
+	a.ArgTypes = []common.Type{common.String, common.String}
+	return nil
+}
diff --git a/internal/primitive/transform/function/strings_functions.go b/internal/primitive/transform/function/strings_functions.go
index e087bf9..9cf4851 100644
--- a/internal/primitive/transform/function/strings_functions.go
+++ b/internal/primitive/transform/function/strings_functions.go
@@ -125,21 +125,18 @@ var CapitalizeWord = function{
 	fixedArgs: []common.Type{common.String},
 	fn: func(args []interface{}) (interface{}, error) {
 		value, _ := args[0].(string)
-		if len(value) == 0 {
-			return value, nil
-		}
-		if len(value) == 1 {
-			return strings.ToUpper(string(value[0])), nil
-		}
-		capWords := make([]rune, 0, len([]rune(value)))
-		prev := ' '
-		for _, v := range value {
-			if v != ' ' && prev == ' ' {
-				v = unicode.ToUpper(v)
+		rs := []rune(value)
+		inWord := false
+		for i, r := range rs {
+			if !unicode.IsSpace(r) {
+				if !inWord {
+					rs[i] = unicode.ToTitle(r)
+				}
+				inWord = true
+			} else {
+				inWord = false
 			}
-			capWords = append(capWords, v)
-			prev = v
 		}
-		return string(capWords), nil
+		return string(rs), nil
 	},
 }
diff --git a/internal/primitive/transform/runtime/init.go b/internal/primitive/transform/runtime/init.go
index ec19d0b..390cd85 100644
--- a/internal/primitive/transform/runtime/init.go
+++ b/internal/primitive/transform/runtime/init.go
@@ -54,6 +54,7 @@ func init() {
 		strings.NewCapitalizeSentenceAction,
 		strings.NewCheckCustomValuesAction,
 		strings.NewCapitalizeWordAction,
+		strings.NewSplitWithDelimiterAction,
 		// condition
 		condition.NewConditionIfAction,
 		// array


Test patch:
diff --git a/internal/controller/eventbus/eventlog/eventlog_test.go b/internal/controller/eventbus/eventlog/eventlog_test.go
index 57fe3b1..3d740d0 100644
--- a/internal/controller/eventbus/eventlog/eventlog_test.go
+++ b/internal/controller/eventbus/eventlog/eventlog_test.go
@@ -511,14 +511,14 @@ func TestEventlogManager_CreateAndGetEventlog(t *testing.T) {
 			So(blockObj.VolumeID, ShouldEqual, vol1.ID)
 			So(blockObj.SegmentID, ShouldEqual, segments[0].ID)
 
-			seg := utMgr.GetSegment(segments2[1].ID)
+			seg := utMgr.getSegment(segments2[1].ID)
 			So(seg, ShouldNotBeNil)
 			So(seg.EventLogID, ShouldEqual, logMD.ID)
 
 			blockObj = utMgr.GetBlock(vanus.NewIDFromUint64(segments[1].Replicas.Leader))
 			segAnother, err := utMgr.GetSegmentByBlockID(blockObj)
 			So(err, ShouldBeNil)
-			So(segAnother, ShouldEqual, seg)
+			So(segAnother, ShouldResemble, *seg)
 		})
 	})
 }
@@ -1064,15 +1064,12 @@ func TestEventlog_All(t *testing.T) {
 			So(el.indexAt(4), ShouldBeNil)
 			So(el.indexAt(999), ShouldBeNil)
 
-			So(el.nextOf(el.indexAt(0)), ShouldEqual, seg2)
-			So(el.nextOf(el.indexAt(1)), ShouldEqual, seg3)
-			So(el.nextOf(el.indexAt(2)), ShouldEqual, seg4)
-			So(el.nextOf(el.indexAt(3)), ShouldBeNil)
+			list := el.getAllSegments()
 
-			So(el.previousOf(el.indexAt(0)), ShouldBeNil)
-			So(el.previousOf(el.indexAt(1)), ShouldEqual, seg1)
-			So(el.previousOf(el.indexAt(2)), ShouldEqual, seg2)
-			So(el.previousOf(el.indexAt(3)), ShouldEqual, seg3)
+			So(list[0], ShouldEqual, seg1)
+			So(list[1], ShouldEqual, seg2)
+			So(list[2], ShouldEqual, seg3)
+			So(list[3], ShouldEqual, seg4)
 		})
 	})
 }
diff --git a/internal/controller/eventbus/eventlog/segment_test.go b/internal/controller/eventbus/eventlog/segment_test.go
index e7a9594..0bada72 100644
--- a/internal/controller/eventbus/eventlog/segment_test.go
+++ b/internal/controller/eventbus/eventlog/segment_test.go
@@ -38,7 +38,7 @@ func TestConvert2ProtoSegment(t *testing.T) {
 		block1 := vanus.NewTestID()
 		block2 := vanus.NewTestID()
 		block3 := vanus.NewTestID()
-		seg := &Segment{
+		seg := Segment{
 			ID:                segID,
 			Capacity:          64 * 1024 * 1024,
 			EventLogID:        eID,
diff --git a/internal/primitive/transform/action/strings/capitalize_word_test.go b/internal/primitive/transform/action/strings/capitalize_word_test.go
index 6c4ae36..8108e4d 100644
--- a/internal/primitive/transform/action/strings/capitalize_word_test.go
+++ b/internal/primitive/transform/action/strings/capitalize_word_test.go
@@ -86,4 +86,16 @@ func TestCapitalizeWordAction(t *testing.T) {
 		So(err, ShouldBeNil)
 		So(e.Extensions()["test"], ShouldEqual, "A")
 	})
+	Convey("test capitalize word: leading symbols", t, func() {
+		a, err := runtime.NewAction([]interface{}{funcName, "$.test"})
+		So(err, ShouldBeNil)
+		e := cetest.MinEvent()
+		e.SetExtension("test", "let 'em go")
+		ceCtx := &context.EventContext{
+			Event: &e,
+		}
+		err = a.Execute(ceCtx)
+		So(err, ShouldBeNil)
+		So(e.Extensions()["test"], ShouldEqual, "Let 'em Go")
+	})
 }
diff --git a/internal/primitive/transform/action/strings/split_with_delimiter_test.go b/internal/primitive/transform/action/strings/split_with_delimiter_test.go
new file mode 100644
index 0000000..19af4f2
--- /dev/null
+++ b/internal/primitive/transform/action/strings/split_with_delimiter_test.go
@@ -0,0 +1,45 @@
+// Copyright 2023 Linkall Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package strings_test
+
+import (
+	"testing"
+
+	cetest "github.com/cloudevents/sdk-go/v2/test"
+	"github.com/linkall-labs/vanus/internal/primitive/transform/action/strings"
+	"github.com/linkall-labs/vanus/internal/primitive/transform/context"
+	"github.com/linkall-labs/vanus/internal/primitive/transform/runtime"
+	. "github.com/smartystreets/goconvey/convey"
+)
+
+func TestSplitWithDelimiterAction(t *testing.T) {
+	funcName := strings.NewSplitWithDelimiterAction().Name()
+	Convey("test split with delimiter", t, func() {
+		a, err := runtime.NewAction([]interface{}{funcName, "$.test", ",", "$.data.target"})
+		So(err, ShouldBeNil)
+		e := cetest.MinEvent()
+		data := map[string]interface{}{}
+		e.SetExtension("test", "one,two,three")
+		ceCtx := &context.EventContext{
+			Event: &e,
+			Data:  data,
+		}
+		err = a.Execute(ceCtx)
+		So(err, ShouldBeNil)
+		res, ok := data["target"]
+		So(ok, ShouldBeTrue)
+		So(res, ShouldResemble, []string{"one", "two", "three"})
+	})
+}
diff --git a/internal/primitive/transform/runtime/action_bench_test.go b/internal/primitive/transform/runtime/action_bench_test.go
index b54f5f4..0ba903b 100644
--- a/internal/primitive/transform/runtime/action_bench_test.go
+++ b/internal/primitive/transform/runtime/action_bench_test.go
@@ -79,4 +79,5 @@ func BenchmarkAction(b *testing.B) {
 	b.Run("replace_with_regex", actionBenchmark([]interface{}{"replace_with_regex", "$.data.str", "a", "Aa"}))
 	b.Run("capitalize_sentence", actionBenchmark([]interface{}{"capitalize_sentence", "$.data.str"}))
 	b.Run("check_custom_values", actionBenchmark([]interface{}{"check_custom_values", "$.data.str", "value", "$.data.target", "true", "false"}))
+	b.Run("split_with_delimiter", actionBenchmark([]interface{}{"split_with_delimiter", "$.data.str", "a", "$.data.target"}))
 }
diff --git a/test/regression/pubsub/main.go b/test/regression/pubsub/main.go
index 3995149..0246e0e 100644
--- a/test/regression/pubsub/main.go
+++ b/test/regression/pubsub/main.go
@@ -32,7 +32,7 @@ var (
 	sentFailed     int64
 	receivedNumber int64
 
-	totalNumber        = int64(1 << 15) // 1024*1024
+	totalNumber        = int64(1 << 20) // 1024*1024
 	parallelism        = 4
 	batchSize          = 8
 	maximumPayloadSize = 4 * 1024