wehub-resource-sync
498b235461
Build and test / Build and test AMD64 Ubuntu 22.04 (push) Failing after 0s
Publish Builder / amazonlinux2023 (push) Failing after 1s
Build and test / UT for Go (push) Has been skipped
Publish KRTE Images / KRTE (push) Failing after 1s
Build and test / Integration Test (push) Has been skipped
Build and test / Upload Code Coverage (push) Has been skipped
Publish Builder / rockylinux9 (push) Failing after 1s
Publish Builder / ubuntu22.04 (push) Failing after 0s
Publish Builder / ubuntu24.04 (push) Failing after 0s
Publish Gpu Builder / publish-gpu-builder (push) Failing after 1s
Publish Test Images / PyTest (push) Failing after 0s
Build and test / UT for Cpp (push) Has been cancelled
333 lines
12 KiB
Go
333 lines
12 KiB
Go
package handler
|
|
|
|
import (
|
|
"context"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/cockroachdb/errors"
|
|
"github.com/stretchr/testify/assert"
|
|
"github.com/stretchr/testify/mock"
|
|
"google.golang.org/grpc"
|
|
|
|
"github.com/milvus-io/milvus-proto/go-api/v3/commonpb"
|
|
"github.com/milvus-io/milvus/internal/mocks/streamingnode/client/handler/mock_assignment"
|
|
"github.com/milvus-io/milvus/internal/mocks/streamingnode/client/handler/mock_consumer"
|
|
"github.com/milvus-io/milvus/internal/mocks/streamingnode/client/handler/mock_producer"
|
|
"github.com/milvus-io/milvus/internal/mocks/util/streamingutil/service/mock_lazygrpc"
|
|
"github.com/milvus-io/milvus/internal/mocks/util/streamingutil/service/mock_resolver"
|
|
"github.com/milvus-io/milvus/internal/streamingnode/client/handler/consumer"
|
|
"github.com/milvus-io/milvus/internal/streamingnode/client/handler/producer"
|
|
"github.com/milvus-io/milvus/internal/util/streamingutil/service/contextutil"
|
|
"github.com/milvus-io/milvus/internal/util/streamingutil/status"
|
|
"github.com/milvus-io/milvus/pkg/v3/mocks/proto/mock_streamingpb"
|
|
"github.com/milvus-io/milvus/pkg/v3/mocks/streaming/util/mock_types"
|
|
"github.com/milvus-io/milvus/pkg/v3/proto/streamingpb"
|
|
"github.com/milvus-io/milvus/pkg/v3/streaming/util/message/adaptor"
|
|
"github.com/milvus-io/milvus/pkg/v3/streaming/util/options"
|
|
"github.com/milvus-io/milvus/pkg/v3/streaming/util/types"
|
|
"github.com/milvus-io/milvus/pkg/v3/util/paramtable"
|
|
"github.com/milvus-io/milvus/pkg/v3/util/typeutil"
|
|
)
|
|
|
|
func TestHandlerClient(t *testing.T) {
|
|
assignment := &types.PChannelInfoAssigned{
|
|
Channel: types.PChannelInfo{Name: "pchannel", Term: 1},
|
|
Node: types.StreamingNodeInfo{ServerID: 1, Address: "localhost"},
|
|
}
|
|
|
|
service := mock_lazygrpc.NewMockService[streamingpb.StreamingNodeHandlerServiceClient](t)
|
|
handlerServiceClient := mock_streamingpb.NewMockStreamingNodeHandlerServiceClient(t)
|
|
handlerServiceClient.EXPECT().GetReplicateCheckpoint(mock.Anything, mock.Anything).RunAndReturn(func(ctx context.Context, grcr *streamingpb.GetReplicateCheckpointRequest, co ...grpc.CallOption) (*streamingpb.GetReplicateCheckpointResponse, error) {
|
|
serverID, ok := contextutil.GetPickServerID(ctx)
|
|
assert.True(t, ok)
|
|
assert.Equal(t, serverID, assignment.Node.ServerID)
|
|
return &streamingpb.GetReplicateCheckpointResponse{
|
|
Checkpoint: &commonpb.ReplicateCheckpoint{
|
|
ClusterId: "pchannel",
|
|
Pchannel: "pchannel",
|
|
MessageId: nil,
|
|
TimeTick: 0,
|
|
},
|
|
}, nil
|
|
})
|
|
service.EXPECT().GetService(mock.Anything).Return(handlerServiceClient, nil)
|
|
rb := mock_resolver.NewMockBuilder(t)
|
|
rb.EXPECT().Close().Run(func() {})
|
|
w := mock_assignment.NewMockWatcher(t)
|
|
w.EXPECT().Close().Run(func() {})
|
|
|
|
p := mock_producer.NewMockProducer(t)
|
|
p.EXPECT().Register(mock.Anything).Return()
|
|
p.EXPECT().Close().RunAndReturn(func() {})
|
|
c := mock_consumer.NewMockConsumer(t)
|
|
c.EXPECT().Close().RunAndReturn(func() error { return nil })
|
|
|
|
rebalanceTrigger := mock_types.NewMockAssignmentRebalanceTrigger(t)
|
|
rebalanceTrigger.EXPECT().ReportAssignmentError(mock.Anything, mock.Anything, mock.Anything).Return(nil)
|
|
|
|
pK := 0
|
|
handler := &handlerClientImpl{
|
|
lifetime: typeutil.NewLifetime(),
|
|
service: service,
|
|
rb: rb,
|
|
watcher: w,
|
|
rebalanceTrigger: rebalanceTrigger,
|
|
newProducer: func(ctx context.Context, opts *producer.ProducerOptions, handler streamingpb.StreamingNodeHandlerServiceClient) (Producer, error) {
|
|
serverID, ok := contextutil.GetPickServerID(ctx)
|
|
assert.True(t, ok)
|
|
assert.Equal(t, serverID, assignment.Node.ServerID)
|
|
if pK == 0 {
|
|
pK++
|
|
return nil, status.NewUnmatchedChannelTerm("pchannel", 1, 2)
|
|
}
|
|
return p, nil
|
|
},
|
|
newConsumer: func(ctx context.Context, opts *consumer.ConsumerOptions, handlerClient streamingpb.StreamingNodeHandlerServiceClient) (Consumer, error) {
|
|
serverID, ok := contextutil.GetPickServerID(ctx)
|
|
assert.True(t, ok)
|
|
assert.Equal(t, serverID, assignment.Node.ServerID)
|
|
return c, nil
|
|
},
|
|
}
|
|
ctx := context.Background()
|
|
|
|
k := 0
|
|
w.EXPECT().Get(mock.Anything, mock.Anything).RunAndReturn(func(ctx context.Context, s string) *types.PChannelInfoAssigned {
|
|
if k == 0 {
|
|
k++
|
|
return nil
|
|
}
|
|
return assignment
|
|
})
|
|
w.EXPECT().Watch(mock.Anything, mock.Anything, mock.Anything).Return(nil)
|
|
|
|
producer, err := handler.CreateProducer(ctx, &ProducerOptions{PChannel: "pchannel"})
|
|
assert.NoError(t, err)
|
|
assert.NotNil(t, producer)
|
|
producer2, err := handler.CreateProducer(ctx, &ProducerOptions{PChannel: "pchannel"})
|
|
assert.NoError(t, err)
|
|
assert.NotNil(t, producer)
|
|
producer3, err := handler.CreateProducer(ctx, &ProducerOptions{PChannel: "pchannel"})
|
|
assert.NoError(t, err)
|
|
assert.NotNil(t, producer3)
|
|
producer.Close()
|
|
producer2.Close()
|
|
producer3.Close()
|
|
|
|
rcp, err := handler.GetReplicateCheckpoint(ctx, "pchannel")
|
|
assert.NoError(t, err)
|
|
assert.NotNil(t, rcp)
|
|
|
|
handler.GetLatestMVCCTimestampIfLocal(ctx, "pchannel")
|
|
producer4, err := handler.CreateProducer(ctx, &ProducerOptions{PChannel: "pchannel"})
|
|
assert.NoError(t, err)
|
|
assert.NotNil(t, producer4)
|
|
producer4.Close()
|
|
|
|
consumer, err := handler.CreateConsumer(ctx, &ConsumerOptions{
|
|
PChannel: "pchannel",
|
|
VChannel: "vchannel",
|
|
DeliverPolicy: options.DeliverPolicyAll(),
|
|
DeliverFilters: []options.DeliverFilter{
|
|
options.DeliverFilterTimeTickGT(10),
|
|
options.DeliverFilterTimeTickGTE(10),
|
|
},
|
|
MessageHandler: make(adaptor.ChanMessageHandler),
|
|
})
|
|
assert.NoError(t, err)
|
|
assert.NotNil(t, consumer)
|
|
consumer.Close()
|
|
|
|
service.EXPECT().Close().Return()
|
|
handler.Close()
|
|
producer, err = handler.CreateProducer(ctx, nil)
|
|
assert.Error(t, err)
|
|
assert.ErrorIs(t, err, ErrClientClosed)
|
|
assert.Nil(t, producer)
|
|
|
|
consumer, err = handler.CreateConsumer(ctx, nil)
|
|
assert.Error(t, err)
|
|
assert.ErrorIs(t, err, ErrClientClosed)
|
|
assert.Nil(t, consumer)
|
|
|
|
handler.GetLatestMVCCTimestampIfLocal(ctx, "pchannel")
|
|
}
|
|
|
|
func TestHandlerClient_GetSalvageCheckpoint(t *testing.T) {
|
|
assignment := &types.PChannelInfoAssigned{
|
|
Channel: types.PChannelInfo{Name: "pchannel", Term: 1},
|
|
Node: types.StreamingNodeInfo{ServerID: 1, Address: "localhost"},
|
|
}
|
|
|
|
service := mock_lazygrpc.NewMockService[streamingpb.StreamingNodeHandlerServiceClient](t)
|
|
rb := mock_resolver.NewMockBuilder(t)
|
|
rb.EXPECT().Close().Run(func() {})
|
|
w := mock_assignment.NewMockWatcher(t)
|
|
w.EXPECT().Close().Run(func() {})
|
|
// Always return the assignment so the create func is invoked.
|
|
w.EXPECT().Get(mock.Anything, mock.Anything).Return(assignment)
|
|
// Watch returns context.Canceled to break the backoff retry loop.
|
|
w.EXPECT().Watch(mock.Anything, mock.Anything, mock.Anything).Return(context.Canceled)
|
|
rebalanceTrigger := mock_types.NewMockAssignmentRebalanceTrigger(t)
|
|
|
|
handler := &handlerClientImpl{
|
|
lifetime: typeutil.NewLifetime(),
|
|
service: service,
|
|
rb: rb,
|
|
watcher: w,
|
|
rebalanceTrigger: rebalanceTrigger,
|
|
}
|
|
ctx := context.Background()
|
|
|
|
// Remote WAL returns "not implemented"; Watch returns Canceled to exit the loop.
|
|
service.EXPECT().GetService(mock.Anything).Return(nil, errors.New("not implemented"))
|
|
cps, err := handler.GetSalvageCheckpoint(ctx, "pchannel")
|
|
assert.Error(t, err)
|
|
assert.Nil(t, cps)
|
|
|
|
// After close, GetSalvageCheckpoint returns ErrClientClosed immediately.
|
|
service.EXPECT().Close().Return()
|
|
handler.Close()
|
|
cps, err = handler.GetSalvageCheckpoint(ctx, "pchannel")
|
|
assert.ErrorIs(t, err, ErrClientClosed)
|
|
assert.Nil(t, cps)
|
|
}
|
|
|
|
func TestHandlerClientReadOnlyAssignmentWaitsForNextAssignment(t *testing.T) {
|
|
assignment := &types.PChannelInfoAssigned{
|
|
Channel: types.PChannelInfo{Name: "pchannel", Term: 1, AccessMode: types.AccessModeRO},
|
|
Node: types.StreamingNodeInfo{ServerID: 1, Address: "localhost"},
|
|
}
|
|
|
|
testcases := []struct {
|
|
name string
|
|
run func(*handlerClientImpl) error
|
|
}{
|
|
{
|
|
name: "replicate checkpoint",
|
|
run: func(handler *handlerClientImpl) error {
|
|
cp, err := handler.GetReplicateCheckpoint(context.Background(), "pchannel")
|
|
assert.Nil(t, cp)
|
|
return err
|
|
},
|
|
},
|
|
{
|
|
name: "salvage checkpoint",
|
|
run: func(handler *handlerClientImpl) error {
|
|
cps, err := handler.GetSalvageCheckpoint(context.Background(), "pchannel")
|
|
assert.Nil(t, cps)
|
|
return err
|
|
},
|
|
},
|
|
{
|
|
name: "producer",
|
|
run: func(handler *handlerClientImpl) error {
|
|
producer, err := handler.CreateProducer(context.Background(), &ProducerOptions{PChannel: "pchannel"})
|
|
assert.Nil(t, producer)
|
|
return err
|
|
},
|
|
},
|
|
}
|
|
|
|
for _, tc := range testcases {
|
|
t.Run(tc.name, func(t *testing.T) {
|
|
w := mock_assignment.NewMockWatcher(t)
|
|
w.EXPECT().Get(mock.Anything, "pchannel").Return(assignment)
|
|
w.EXPECT().Watch(mock.Anything, "pchannel", assignment).Return(context.Canceled)
|
|
|
|
handler := &handlerClientImpl{
|
|
lifetime: typeutil.NewLifetime(),
|
|
watcher: w,
|
|
}
|
|
|
|
err := tc.run(handler)
|
|
assert.ErrorIs(t, err, context.Canceled)
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestHandlerClient_PrepareReleaseManualFlush(t *testing.T) {
|
|
assignment := &types.PChannelInfoAssigned{
|
|
Channel: types.PChannelInfo{Name: "pchannel", Term: 1, AccessMode: types.AccessModeRO},
|
|
Node: types.StreamingNodeInfo{ServerID: 1, Address: "localhost"},
|
|
}
|
|
vchannel := "pchannel_100v0"
|
|
releaseSegmentIDs := []int64{1001}
|
|
|
|
service := mock_lazygrpc.NewMockService[streamingpb.StreamingNodeHandlerServiceClient](t)
|
|
service.EXPECT().Close().Return()
|
|
|
|
rb := mock_resolver.NewMockBuilder(t)
|
|
rb.EXPECT().Close().Run(func() {})
|
|
w := mock_assignment.NewMockWatcher(t)
|
|
w.EXPECT().Get(mock.Anything, "pchannel").Return(assignment)
|
|
w.EXPECT().Close().Run(func() {})
|
|
|
|
handler := &handlerClientImpl{
|
|
lifetime: typeutil.NewLifetime(),
|
|
service: service,
|
|
rb: rb,
|
|
watcher: w,
|
|
}
|
|
|
|
prepared, err := handler.PrepareReleaseManualFlush(context.Background(), 100, vchannel, releaseSegmentIDs)
|
|
assert.NoError(t, err)
|
|
assert.False(t, prepared)
|
|
|
|
handler.Close()
|
|
prepared, err = handler.PrepareReleaseManualFlush(context.Background(), 100, vchannel, releaseSegmentIDs)
|
|
assert.ErrorIs(t, err, ErrClientClosed)
|
|
assert.False(t, prepared)
|
|
}
|
|
|
|
func TestHandlerClient_GetReplicateCheckpointReplicateViolation(t *testing.T) {
|
|
assignment := &types.PChannelInfoAssigned{
|
|
Channel: types.PChannelInfo{Name: "pchannel", Term: 1},
|
|
Node: types.StreamingNodeInfo{ServerID: 1, Address: "localhost"},
|
|
}
|
|
|
|
service := mock_lazygrpc.NewMockService[streamingpb.StreamingNodeHandlerServiceClient](t)
|
|
handlerServiceClient := mock_streamingpb.NewMockStreamingNodeHandlerServiceClient(t)
|
|
// Remote WAL reports a replicate violation: the target is no longer a secondary
|
|
// cluster (e.g. after force_promote). This is unrecoverable for the current WAL
|
|
// role, so it must be returned immediately rather than retried to the deadline.
|
|
handlerServiceClient.EXPECT().GetReplicateCheckpoint(mock.Anything, mock.Anything).Return(
|
|
nil, status.NewReplicateViolation("wal is not a secondary cluster in replicating topology"))
|
|
service.EXPECT().GetService(mock.Anything).Return(handlerServiceClient, nil)
|
|
|
|
w := mock_assignment.NewMockWatcher(t)
|
|
// Always return the assignment so the create func is invoked.
|
|
w.EXPECT().Get(mock.Anything, mock.Anything).Return(assignment)
|
|
// Watch is intentionally NOT expected: an immediate return must not enter the
|
|
// backoff retry loop. If it did, the mock would fail on an unexpected Watch call.
|
|
rebalanceTrigger := mock_types.NewMockAssignmentRebalanceTrigger(t)
|
|
|
|
handler := &handlerClientImpl{
|
|
lifetime: typeutil.NewLifetime(),
|
|
service: service,
|
|
watcher: w,
|
|
rebalanceTrigger: rebalanceTrigger,
|
|
}
|
|
|
|
cp, err := handler.GetReplicateCheckpoint(context.Background(), "pchannel")
|
|
assert.Error(t, err)
|
|
assert.Nil(t, cp)
|
|
assert.True(t, status.AsStreamingError(err).IsReplicateViolation())
|
|
}
|
|
|
|
func TestDial(t *testing.T) {
|
|
paramtable.Init()
|
|
|
|
w := mock_types.NewMockAssignmentDiscoverWatcher(t)
|
|
w.EXPECT().AssignmentDiscover(mock.Anything, mock.Anything).RunAndReturn(
|
|
func(ctx context.Context, f func(*types.VersionedStreamingNodeAssignments) error) error {
|
|
return context.Canceled
|
|
},
|
|
)
|
|
handler := NewHandlerClient(w)
|
|
assert.NotNil(t, handler)
|
|
time.Sleep(100 * time.Millisecond)
|
|
handler.Close()
|
|
}
|