From cfefe78401f38993692e761418e1062ae16799a1 Mon Sep 17 00:00:00 2001
From: Bohan Zhang <tabvision@bupt.icu>
Date: Thu, 28 Mar 2024 18:42:58 +0800
Subject: [PATCH] fix: handle kinesis unhandled error: InternalError (#15979)
 (#16001)

---
 src/connector/src/source/kinesis/source/reader.rs | 12 +++++++++++-
 1 file changed, 11 insertions(+), 1 deletion(-)

diff --git a/src/connector/src/source/kinesis/source/reader.rs b/src/connector/src/source/kinesis/source/reader.rs
index 51b3c77710410..94670a5f6665f 100644
--- a/src/connector/src/source/kinesis/source/reader.rs
+++ b/src/connector/src/source/kinesis/source/reader.rs
@@ -16,7 +16,7 @@ use std::time::Duration;
 
 use anyhow::{anyhow, Result};
 use async_trait::async_trait;
-use aws_sdk_kinesis::error::{DisplayErrorContext, SdkError};
+use aws_sdk_kinesis::error::{DisplayErrorContext, ProvideErrorMetadata, SdkError};
 use aws_sdk_kinesis::operation::get_records::{GetRecordsError, GetRecordsOutput};
 use aws_sdk_kinesis::primitives::DateTime;
 use aws_sdk_kinesis::types::ShardIteratorType;
@@ -188,6 +188,16 @@ impl CommonSplitReader for KinesisSplitReader {
                     self.new_shard_iter().await?;
                     continue;
                 }
+                Err(e) if e.code() == Some("InternalFailure") => {
+                    tracing::warn!(
+                        "stream {:?} shard {:?} met internal failure, retrying",
+                        self.stream_name,
+                        self.shard_id
+                    );
+                    self.new_shard_iter().await?;
+                    tokio::time::sleep(Duration::from_millis(200)).await;
+                    continue;
+                }
                 Err(e) => {
                     let error_msg = format!(
                         "Kinesis got a unhandled error: {:?}, stream {:?}, shard {:?}",