fix: issue-238 (#315)

* fix: issue-238

Signed-off-by: delu <[email protected]>

* fix: issue-238

Signed-off-by: delu <[email protected]>

Signed-off-by: delu <[email protected]>

Author: xdlbdy

internal/controller/trigger/controller_test.go

@@ -296,22 +296,6 @@ func TestController_UpdateSubscription(t *testing.T) {
 				So(resp.SinkCredential.GetAws().SecretAccessKey, ShouldEqual, primitive.SecretsMask)
 			})
 		})
-		Convey("update rate limit", func() {
-			subManager.EXPECT().UpdateSubscription(gomock.Any(), gomock.Any()).Return(nil)
-			request := &ctrlpb.UpdateSubscriptionRequest{
-				Id: subID.Uint64(),
-				Subscription: &ctrlpb.SubscriptionRequest{
-					EventBus: "test-eb",
-					Sink:     "test-sink",
-					Config: &metapb.SubscriptionConfig{
-						RateLimit: -1,
-					},
-				},
-			}
-			resp, err := ctrl.UpdateSubscription(ctx, request)
-			So(err, ShouldBeNil)
-			So(resp.Config.RateLimit, ShouldEqual, request.Subscription.Config.RateLimit)
-		})
 		Convey("update subscription sink", func() {
 			subManager.EXPECT().UpdateSubscription(gomock.Any(), gomock.Any()).Return(nil)
 			request := &ctrlpb.UpdateSubscriptionRequest{

internal/controller/trigger/validation/subscripton.go

@@ -138,9 +138,6 @@ func validateSubscriptionConfig(ctx context.Context, cfg *metapb.SubscriptionCon
 	if cfg == nil {
 		return nil
 	}
-	if cfg.RateLimit < -1 {
-		return errors.ErrInvalidRequest.WithMessage("rate limit is -1 or gt than 0")
-	}
 	switch cfg.OffsetType {
 	case metapb.SubscriptionConfig_LATEST, metapb.SubscriptionConfig_EARLIEST:
 	case metapb.SubscriptionConfig_TIMESTAMP:
@@ -153,7 +150,7 @@ func validateSubscriptionConfig(ctx context.Context, cfg *metapb.SubscriptionCon
 	if cfg.DeadLetterEventbus != "" {
 		return errors.ErrInvalidRequest.WithMessage("no support to set dead letter eventbus")
 	}
-	if cfg.MaxRetryAttempts > primitive.MaxRetryAttempts {
+	if cfg.GetMaxRetryAttempts() > primitive.MaxRetryAttempts {
 		return errors.ErrInvalidRequest.WithMessage(
 			fmt.Sprintf("could not set max retry attempts greater than %d", primitive.MaxRetryAttempts))
 	}

internal/controller/trigger/validation/suscription_test.go

@@ -53,21 +53,16 @@ func TestSubscriptionRequestValidator(t *testing.T) {
 func TestValidateSubscriptionConfig(t *testing.T) {
 	ctx := context.Background()
 	Convey("test validate subscription config", t, func() {
-		Convey("test rate limit", func() {
-			config := &metapb.SubscriptionConfig{
-				RateLimit: -2,
-			}
-			So(validateSubscriptionConfig(ctx, config), ShouldNotBeNil)
-		})
 		Convey("test offset timestamp", func() {
 			config := &metapb.SubscriptionConfig{
 				OffsetType: metapb.SubscriptionConfig_TIMESTAMP,
 			}
 			So(validateSubscriptionConfig(ctx, config), ShouldNotBeNil)
 		})
 		Convey("test max retry attempts", func() {
+			attempt := uint32(10000)
 			config := &metapb.SubscriptionConfig{
-				MaxRetryAttempts: 10000,
+				MaxRetryAttempts: &attempt,
 			}
 			So(validateSubscriptionConfig(ctx, config), ShouldNotBeNil)
 		})

internal/primitive/subscription.go

@@ -64,15 +64,23 @@ const (
 )
 
 type SubscriptionConfig struct {
-	RateLimit int32 `json:"rate_limit,omitempty"`
+	RateLimit uint32 `json:"rate_limit,omitempty"`
 	// consumer from
 	OffsetType         OffsetType `json:"offset_type,omitempty"`
 	OffsetTimestamp    *uint64    `json:"offset_timestamp,omitempty"`
-	DeliveryTimeout    int32      `json:"delivery_timeout,omitempty"`
-	MaxRetryAttempts   int32      `json:"max_retry_attempts,omitempty"`
+	DeliveryTimeout    uint32     `json:"delivery_timeout,omitempty"`
+	MaxRetryAttempts   *uint32    `json:"max_retry_attempts,omitempty"`
 	DeadLetterEventbus string     `json:"dead_letter_eventbus,omitempty"`
 }
 
+// GetMaxRetryAttempts return MaxRetryAttempts if nil return -1.
+func (c *SubscriptionConfig) GetMaxRetryAttempts() int32 {
+	if c != nil && c.MaxRetryAttempts != nil {
+		return int32(*c.MaxRetryAttempts)
+	}
+	return -1
+}
+
 func (c *SubscriptionConfig) String() string {
 	if c == nil {
 		return ""

internal/trigger/trigger/config.go

@@ -33,7 +33,7 @@ type Config struct {
 	BufferSize         int
 	MaxRetryAttempts   int32
 	DeliveryTimeout    time.Duration
-	RateLimit          int32
+	RateLimit          uint32
 	Controllers        []string
 	DeadLetterEventbus string
 }
@@ -71,27 +71,27 @@ func WithBufferSize(size int) Option {
 
 func WithMaxRetryAttempts(attempts int32) Option {
 	return func(t *trigger) {
-		if attempts <= 0 {
+		if attempts < 0 {
 			attempts = primitive.MaxRetryAttempts
 		}
 		t.config.MaxRetryAttempts = attempts
 	}
 }
 
-func WithDeliveryTimeout(timeout int32) Option {
+func WithDeliveryTimeout(timeout uint32) Option {
 	return func(t *trigger) {
-		if timeout <= 0 {
+		if timeout == 0 {
 			t.config.DeliveryTimeout = defaultDeliveryTimeout
 			return
 		}
 		t.config.DeliveryTimeout = time.Duration(timeout) * time.Millisecond
 	}
 }
 
-func WithRateLimit(rateLimit int32) Option {
+func WithRateLimit(rateLimit uint32) Option {
 	return func(t *trigger) {
 		t.config.RateLimit = rateLimit
-		if rateLimit <= 0 {
+		if rateLimit == 0 {
 			t.rateLimiter = ratelimit.NewUnlimited()
 			return
 		}

internal/trigger/trigger/trigger.go

@@ -172,8 +172,8 @@ func (t *trigger) changeConfig(config primitive.SubscriptionConfig) {
 	if config.DeliveryTimeout != t.subscription.Config.DeliveryTimeout {
 		t.applyOptions(WithDeliveryTimeout(config.DeliveryTimeout))
 	}
-	if config.MaxRetryAttempts != t.subscription.Config.MaxRetryAttempts {
-		t.applyOptions(WithMaxRetryAttempts(config.MaxRetryAttempts))
+	if config.GetMaxRetryAttempts() != t.subscription.Config.GetMaxRetryAttempts() {
+		t.applyOptions(WithMaxRetryAttempts(config.GetMaxRetryAttempts()))
 	}
 	t.subscription.Config = config
 }

internal/trigger/trigger/trigger_test.go

@@ -49,9 +49,9 @@ func TestTrigger_Options(t *testing.T) {
 		WithDeliveryTimeout(0)(tg)
 		So(tg.config.DeliveryTimeout, ShouldEqual, defaultDeliveryTimeout)
 		size = rand.Intn(1000) + size
-		WithDeliveryTimeout(int32(size))(tg)
+		WithDeliveryTimeout(uint32(size))(tg)
 		So(tg.config.DeliveryTimeout, ShouldEqual, time.Duration(size)*time.Millisecond)
-		WithMaxRetryAttempts(0)(tg)
+		WithMaxRetryAttempts(-1)(tg)
 		So(tg.config.MaxRetryAttempts, ShouldEqual, primitive.MaxRetryAttempts)
 		size = rand.Intn(1000) + size
 		WithMaxRetryAttempts(int32(size))(tg)
@@ -64,7 +64,7 @@ func TestTrigger_Options(t *testing.T) {
 		WithRateLimit(0)(tg)
 		So(tg.config.RateLimit, ShouldEqual, 0)
 		size = rand.Intn(1000) + size
-		WithRateLimit(int32(size))(tg)
+		WithRateLimit(uint32(size))(tg)
 		So(tg.config.RateLimit, ShouldEqual, size)
 		WithDeadLetterEventbus("")(tg)
 		So(tg.config.DeadLetterEventbus, ShouldEqual, primitive.DeadLetterEventbusName)
@@ -220,7 +220,7 @@ func TestTriggerRateLimit(t *testing.T) {
 		tg := NewTrigger(makeSubscription(id), WithControllers([]string{"test"})).(*trigger)
 		tg.eventCli = cli
 		cli.EXPECT().Send(gomock.Any(), gomock.Any()).AnyTimes().Return(client.Success)
-		rateLimit := int32(10000)
+		rateLimit := uint32(10000)
 		Convey("test no rate limit", func() {
 			c := testSendEvent(tg)
 			So(c, ShouldBeGreaterThan, rateLimit)

internal/trigger/worker.go

@@ -290,7 +290,7 @@ func (w *worker) getTriggerOptions(subscription *primitive.Subscription) []trigg
 	config := subscription.Config
 	opts = append(opts, trigger.WithRateLimit(config.RateLimit),
 		trigger.WithDeliveryTimeout(config.DeliveryTimeout),
-		trigger.WithMaxRetryAttempts(config.MaxRetryAttempts),
+		trigger.WithMaxRetryAttempts(config.GetMaxRetryAttempts()),
 		trigger.WithDeadLetterEventbus(config.DeadLetterEventbus))
 	return opts
 }