fix(reader): skip dataloss gap error

client/src/main/java/io/hstream/HServerException.java

@@ -7,10 +7,10 @@
 public class HServerException extends RuntimeException {
   ErrBody errBody;
 
-  static class ErrBody {
-    int error;
-    String message;
-    JsonElement extra;
+  public static class ErrBody {
+    public int error;
+    public String message;
+    public JsonElement extra;
   }
 
   HServerException(ErrBody errBody) {
@@ -26,6 +26,10 @@ public String getErrorMessage() {
     return errBody.message;
   }
 
+  public ErrBody getErrBody() {
+    return errBody;
+  }
+
   public String getRawErrorBody() {
     return new Gson().toJson(errBody);
   }

client/src/main/kotlin/io/hstream/impl/ReaderKtImpl.kt

@@ -1,12 +1,17 @@
 package io.hstream.impl
 
+import io.grpc.Status
+import io.grpc.StatusException
+import io.grpc.StatusRuntimeException
+import io.hstream.HServerException
 import io.hstream.HStreamDBClientException
 import io.hstream.Reader
 import io.hstream.ReceivedRecord
 import io.hstream.Record
 import io.hstream.StreamShardOffset
 import io.hstream.internal.CreateShardReaderRequest
 import io.hstream.internal.DeleteShardReaderRequest
+import io.hstream.internal.ErrorCode
 import io.hstream.internal.LookupShardReaderRequest
 import io.hstream.internal.ReadShardRequest
 import io.hstream.util.GrpcUtils
@@ -70,13 +75,27 @@ class ReaderKtImpl(
                     }
                 }
                 readFuture.complete(res as MutableList<ReceivedRecord>?)
+            } catch (e: StatusException) {
+                handleGrpcError(e.status, readFuture)
+            } catch (e: StatusRuntimeException) {
+                handleGrpcError(e.status, readFuture)
             } catch (e: Throwable) {
                 readFuture.completeExceptionally(HStreamDBClientException(e))
             }
         }
         return readFuture
     }
 
+    private fun handleGrpcError(status: Status, future: CompletableFuture<MutableList<ReceivedRecord>>) {
+        val e = HServerException.tryToHServerException(status.description)
+        if (e != null && e.errBody.error == ErrorCode.ShardReaderDataLossGap.number) {
+            logger.warn("skip a data loss gap error, {}", status)
+            future.complete(mutableListOf())
+        } else {
+            future.completeExceptionally(HStreamDBClientException(status.asException()))
+        }
+    }
+
     override fun close() {
         val deleteShardReaderRequest = DeleteShardReaderRequest.newBuilder()
             .setReaderId(readerId)