Mapping bigquery errors to http status codes

hermes-consumers/src/main/java/pl/allegro/tech/hermes/consumers/consumer/sender/MessageSendingResult.java

@@ -36,6 +36,10 @@ static SingleMessageSendingResult failedResult(int statusCode) {
     return new SingleMessageSendingResult(statusCode);
   }
 
+  static SingleMessageSendingResult failedResult(int statusCode, Throwable cause) {
+    return new SingleMessageSendingResult(statusCode, cause);
+  }
+
   static SingleMessageSendingResult ofStatusCode(int statusCode) {
     return new SingleMessageSendingResult(statusCode);
   }

hermes-consumers/src/main/java/pl/allegro/tech/hermes/consumers/consumer/sender/SingleMessageSendingResult.java

@@ -75,6 +75,12 @@ public String toString() {
     initializeForStatusCode(statusCode);
   }
 
+  SingleMessageSendingResult(int statusCode, Throwable failure) {
+    this.failure = failure;
+    initializeForStatusCode(statusCode);
+    this.loggable = true;
+  }
+
   SingleMessageSendingResult(int statusCode, URI requestURI) {
     this(statusCode);
     this.requestUri = Optional.of(requestURI);

hermes-consumers/src/main/java/pl/allegro/tech/hermes/consumers/consumer/sender/googlebigquery/GoogleBigQueryAppendCompleteCallback.java

@@ -3,6 +3,7 @@
 import com.google.api.core.ApiFutureCallback;
 import com.google.cloud.bigquery.storage.v1.AppendRowsResponse;
 import com.google.cloud.bigquery.storage.v1.Exceptions;
+import io.grpc.Status;
 import java.util.Objects;
 import java.util.concurrent.CompletableFuture;
 import pl.allegro.tech.hermes.consumers.consumer.sender.MessageSendingResult;
@@ -16,11 +17,26 @@ public GoogleBigQueryAppendCompleteCallback(
     this.resultFuture = resultFuture;
   }
 
+  public static Integer mapToPermanentErrorHttpStatus(Throwable cause) {
+    Status.Code grpcCode = Status.fromThrowable(cause).getCode();
+    return switch (grpcCode) {
+      case NOT_FOUND -> 404; // Table does not exist
+      case PERMISSION_DENIED ->
+          403; // Technical user does not have permissions to write to the table
+      case INVALID_ARGUMENT ->
+          400; // Invalid message format i.e. microsecond timestamp value is sent to millisecond
+      // timestamp field
+      default -> 500;
+    };
+  }
+
   @Override
   public void onFailure(Throwable t) {
     Exceptions.StorageException storageException = Exceptions.toStorageException(t);
-    resultFuture.complete(
-        MessageSendingResult.failedResult(Objects.requireNonNullElse(storageException, t)));
+    Throwable cause = Objects.requireNonNullElse(storageException, t);
+
+    Integer httpStatusCode = mapToPermanentErrorHttpStatus(cause);
+    resultFuture.complete(MessageSendingResult.failedResult(httpStatusCode, cause));
   }
 
   @Override

hermes-consumers/src/main/java/pl/allegro/tech/hermes/consumers/consumer/sender/googlebigquery/GoogleBigQueryDataWriter.java

@@ -50,16 +50,18 @@ public void publish(T message, CompletableFuture<MessageSendingResult> resultFut
               .map(entry -> String.format("\t row %d: %s", entry.getKey(), entry.getValue()))
               .collect(Collectors.joining("\n")),
           e);
-
-      throw e;
+      int statusCode = GoogleBigQueryAppendCompleteCallback.mapToPermanentErrorHttpStatus(e);
+      resultFuture.complete(
+          MessageSendingResult.failedResult(statusCode, new GoogleBigQueryFailedAppendException(e)));
     } catch (Exception e) {
       logger.warn(
           "Writer {} has failed to append rows to stream {} because of {}",
           getWriterId(),
           getStreamName(),
           e.getMessage(),
           e);
-      throw e;
+      Integer statusCode = GoogleBigQueryAppendCompleteCallback.mapToPermanentErrorHttpStatus(e);
+      resultFuture.complete(MessageSendingResult.failedResult(statusCode, e));
     }
   }
 

hermes-consumers/src/main/java/pl/allegro/tech/hermes/consumers/consumer/sender/googlebigquery/GoogleBigQueryFailedAppendException.java

@@ -0,0 +1,30 @@
+package pl.allegro.tech.hermes.consumers.consumer.sender.googlebigquery;
+
+import com.google.cloud.bigquery.storage.v1.Exceptions;
+import java.util.Map;
+
+public class GoogleBigQueryFailedAppendException extends RuntimeException {
+  public GoogleBigQueryFailedAppendException(Exceptions.AppendSerializtionError cause) {
+    super(cause.getMessage(), cause);
+  }
+
+  @Override
+  public String getMessage() {
+    Exceptions.AppendSerializtionError cause =
+        ((Exceptions.AppendSerializtionError) this.getCause());
+    StringBuilder message =
+        new StringBuilder(
+            String.format(
+                "GoogleBigQuery Subscription has failed to append rows to stream %s",
+                cause.getStreamName()));
+    message.append(String.format("\n%s", super.getMessage()));
+    if (cause.getRowIndexToErrorMessage() != null) {
+      for (Map.Entry<Integer, String> entry : cause.getRowIndexToErrorMessage().entrySet()) {
+        message.append(
+            String.format(
+                "\nGoogleBigQuery Subscription has failed because of %s", entry.getValue()));
+      }
+    }
+    return message.toString();
+  }
+}