kafka(ticdc): fix sarama admin client reset by error. (#9481) (#9507)

close #9480
pingcap · Aug 14, 2023 · db5fb4d · db5fb4d
1 parent fe96b21
commit db5fb4d
Showing 1 changed file with 3 additions and 10 deletions.
diff --git a/pkg/sink/kafka/admin.go b/pkg/sink/kafka/admin.go
@@ -102,17 +102,10 @@ func (a *saramaAdminClient) queryClusterWithRetry(ctx context.Context, query fun
 			zap.String("changefeed", a.changefeed.ID),
 			zap.Error(err))
 
-		if !errors.Is(err, syscall.EPIPE) {
-			return err
-		}
-		if !errors.Is(err, net.ErrClosed) {
-			return err
+		if errors.Is(err, syscall.EPIPE) || errors.Is(err, net.ErrClosed) || errors.Is(err, io.EOF) {
+			return a.reset()
 		}
-		if !errors.Is(err, io.EOF) {
-			return err
-		}
-
-		return a.reset()
+		return err
 	}, retry.WithBackoffBaseDelay(defaultRetryBackoff), retry.WithMaxTries(defaultRetryMaxTries))
 	return err
 }