cherry pick pingcap#1196 to release-4.0

Signed-off-by: ti-srebot <ti-srebot@pingcap.com>

cdc/kv/client.go

@@ -351,14 +351,20 @@ func (c *CDCClient) newStream(ctx context.Context, addr string, storeID uint64)
 		}
 		err = version.CheckStoreVersion(ctx, c.pd, storeID)
 		if err != nil {
-			conn.Close()
+			// TODO: we don't close gPRC conn here, let it goes into TransientFailure
+			// state. If the store recovers, the gPRC conn can be reused. But if
+			// store goes away forever, the conn will be leaked, we need a better
+			// connection pool.
 			log.Error("check tikv version failed", zap.Error(err), zap.Uint64("storeID", storeID))
 			return errors.Trace(err)
 		}
 		client := cdcpb.NewChangeDataClient(conn)
 		stream, err = client.EventFeed(ctx)
 		if err != nil {
-			conn.Close()
+			// TODO: we don't close gPRC conn here, let it goes into TransientFailure
+			// state. If the store recovers, the gPRC conn can be reused. But if
+			// store goes away forever, the conn will be leaked, we need a better
+			// connection pool.
 			err = cerror.WrapError(cerror.ErrTiKVEventFeed, err)
 			log.Info("establish stream to store failed, retry later", zap.String("addr", addr), zap.Error(err))
 			return err