revert tryagain error

hwjiangkai · hwjiangkai · commit 16661cb34306 · 2022-12-23T13:40:50.000+08:00
Signed-off-by: jyjiangkai &lt;jyjiangkai@163.com&gt;
diff --git a/client/pkg/eventlog/eventlog_impl.go b/client/pkg/eventlog/eventlog_impl.go
@@ -505,7 +505,9 @@ func (r *logReader) ReadStream(ctx context.Context, size int16) ([]*ce.Event, er
 	if err != nil {
 		if errors.Is(err, errors.ErrOffsetOverflow) {
 			r.elog.refreshReadableSegments(ctx)
-			r.switchSegment(ctx)
+			if r.switchSegment(ctx) {
+				return nil, errors.ErrTryAgain
+			}
 		}
 		return nil, err
 	}
diff --git a/internal/timer/timingwheel/bucket.go b/internal/timer/timingwheel/bucket.go
@@ -199,7 +199,7 @@ func (b *bucket) run(ctx context.Context) {
 				// batch read
 				events, err := b.getEvent(ctx, defaultNumberOfEventsRead)
 				if err != nil {
-					if !errors.Is(err, errors.ErrOffsetOnEnd) {
+					if !errors.Is(err, errors.ErrOffsetOnEnd) && !errors.Is(err, errors.ErrTryAgain) {
 						log.Error(ctx, "get event failed when bucket running", map[string]interface{}{
 							"eventbus":   b.getEventbus(),
 							log.KeyError: err,
diff --git a/internal/trigger/reader/reader.go b/internal/trigger/reader/reader.go
@@ -286,7 +286,7 @@ func (elReader *eventLogReader) run(ctx context.Context) {
 		for {
 			err = elReader.readEvent(ctx, lr)
 			switch {
-			case err == nil, errors.Is(err, errors.ErrOffsetOnEnd), errors.Is(err, errors.ErrOffsetOverflow):
+			case err == nil, errors.Is(err, errors.ErrOffsetOnEnd), errors.Is(err, errors.ErrTryAgain):
 			case stderr.Is(err, context.Canceled):
 				return
 			case errors.Is(err, errors.ErrOffsetUnderflow):
diff --git a/pkg/errors/errors.go b/pkg/errors/errors.go
@@ -37,6 +37,7 @@ var (
 	ErrOffsetOnEnd     = New("the offset on end").WithGRPCCode(errpb.ErrorCode_OFFSET_ON_END)
 	ErrOffsetOverflow  = New("the offset overflow").WithGRPCCode(errpb.ErrorCode_OFFSET_OVERFLOW)
 	ErrOffsetUnderflow = New("the offset underflow").WithGRPCCode(errpb.ErrorCode_OFFSET_UNDERFLOW)
+	ErrTryAgain        = New("try again").WithGRPCCode(errpb.ErrorCode_TRY_AGAIN)
 
 	// RESOURCE_NOT_FOUND
 	ErrResourceNotFound = New("resource not found").WithGRPCCode(errpb.ErrorCode_RESOURCE_NOT_FOUND)
diff --git a/proto/pkg/errors/errors.pb.go b/proto/pkg/errors/errors.pb.go
diff --git a/proto/proto/errors.proto b/proto/proto/errors.proto
@@ -34,6 +34,7 @@ enum ErrorCode {
   OFFSET_ON_END = 9303;
   OFFSET_OVERFLOW = 9304;
   OFFSET_UNDERFLOW = 9305;
+  TRY_AGAIN = 9306;
 
   RESOURCE_NOT_FOUND = 9400;
 

Original file line number	Diff line number	Diff line change
`@@ -505,7 +505,9 @@ func (r logReader) ReadStream(ctx context.Context, size int16) ([]ce.Event, er`
`505`	`505`	`if err != nil {`
`506`	`506`	`if errors.Is(err, errors.ErrOffsetOverflow) {`
`507`	`507`	`r.elog.refreshReadableSegments(ctx)`
`508`		`- r.switchSegment(ctx)`
	`508`	`+ if r.switchSegment(ctx) {`
	`509`	`+ return nil, errors.ErrTryAgain`
	`510`	`+ }`
`509`	`511`	`}`
`510`	`512`	`return nil, err`
`511`	`513`	`}`