You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "mateczagany (via GitHub)" <gi...@apache.org> on 2025/09/10 19:15:38 UTC
[PR] [FLINK-35280] Migrate HBase Sink to use Async Sink API [flink-connector-hbase]
mateczagany opened a new pull request, #57:
URL: https://github.com/apache/flink-connector-hbase/pull/57
- I have kept the previous configurations, deprecated them and mapped them to the async configurations in `HBase2DynamicTableFactory`
- I had to add `SerializableMutation` because `Mutation` is not serializable
- I have added `commons-io` to the shaded JAR because the async version of HBase client needed it
- I had to disable test `testTableSinkDisabledBufferFlush` because it was not working without Flink being able to chain all the operators into 1 operator.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-35280] Migrate HBase Sink to use Async Sink API [flink-connector-hbase]
Posted by "mateczagany (via GitHub)" <gi...@apache.org>.
mateczagany commented on code in PR #57:
URL: https://github.com/apache/flink-connector-hbase/pull/57#discussion_r2555154630
##########
flink-connector-hbase-base/src/main/java/org/apache/flink/connector/hbase/sink/HBaseWriterAsyncHandler.java:
##########
@@ -0,0 +1,156 @@
+package org.apache.flink.connector.hbase.sink;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.connector.base.sink.writer.ResultHandler;
+import org.apache.flink.metrics.Counter;
+import org.apache.flink.util.Preconditions;
+
+import org.apache.hadoop.hbase.DoNotRetryIOException;
+import org.apache.hadoop.hbase.client.Mutation;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.stream.Collectors;
+
+/**
+ * This class is responsible for managing the async calls to HBase and managing the {@link
+ * ResultHandler} to decide which request can be retried.
+ */
+@Internal
+public class HBaseWriterAsyncHandler {
+ private static final Logger LOG = LoggerFactory.getLogger(HBaseWriterAsyncHandler.class);
+
+ private final Counter numRecordsOutErrorsCounter;
+
+ public HBaseWriterAsyncHandler(Counter numRecordsOutErrorsCounter) {
+ this.numRecordsOutErrorsCounter = numRecordsOutErrorsCounter;
+ }
+
+ /**
+ * For a given list of HBase write futures, this method will asynchronously analyze their
+ * result, and using the provided {@link ResultHandler}, it will instruct {@link
+ * org.apache.flink.connector.base.sink.writer.AsyncSinkWriter} to retry some mutations. In case
+ * of errors which should not be retried, the Flink job will stop with an error.
+ *
+ * @param futures list of HBase write futures
+ * @param processedMutationsInOrder list of mutations with their indices matching that of the
+ * futures
+ * @param resultHandler result handler to manage retries and exceptions
+ */
+ public void handleWriteFutures(
+ List<CompletableFuture<Mutation>> futures,
+ List<SerializableMutation> processedMutationsInOrder,
+ ResultHandler<SerializableMutation> resultHandler) {
+ Preconditions.checkArgument(
+ futures.size() == processedMutationsInOrder.size(),
+ "Different number of HBase futures was supplied than mutations.");
+
+ ConcurrentLinkedQueue<FailedMutation> failedMutations = new ConcurrentLinkedQueue<>();
+
+ // Handle each future separately and store failures.
+ CompletableFuture<?>[] handledFutures = new CompletableFuture[futures.size()];
+ for (int i = 0; i < futures.size(); i++) {
+ final int index = i;
+ handledFutures[index] =
+ futures.get(index)
+ .exceptionally(
+ throwable -> {
+ failedMutations.add(
+ new FailedMutation(
+ processedMutationsInOrder.get(index),
+ throwable));
+ return null;
+ });
+ }
Review Comment:
I remember that I've spent some trying to make this cleaner, but this did not cross my mind, thank you
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-35280] Migrate HBase Sink to use Async Sink API [flink-connector-hbase]
Posted by "mateczagany (via GitHub)" <gi...@apache.org>.
mateczagany commented on code in PR #57:
URL: https://github.com/apache/flink-connector-hbase/pull/57#discussion_r2567299976
##########
flink-connector-hbase-base/src/main/java/org/apache/flink/connector/hbase/sink/HBaseWriterAsyncHandler.java:
##########
@@ -0,0 +1,156 @@
+package org.apache.flink.connector.hbase.sink;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.connector.base.sink.writer.ResultHandler;
+import org.apache.flink.metrics.Counter;
+import org.apache.flink.util.Preconditions;
+
+import org.apache.hadoop.hbase.DoNotRetryIOException;
+import org.apache.hadoop.hbase.client.Mutation;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.stream.Collectors;
+
+/**
+ * This class is responsible for managing the async calls to HBase and managing the {@link
+ * ResultHandler} to decide which request can be retried.
+ */
+@Internal
+public class HBaseWriterAsyncHandler {
+ private static final Logger LOG = LoggerFactory.getLogger(HBaseWriterAsyncHandler.class);
+
+ private final Counter numRecordsOutErrorsCounter;
+
+ public HBaseWriterAsyncHandler(Counter numRecordsOutErrorsCounter) {
+ this.numRecordsOutErrorsCounter = numRecordsOutErrorsCounter;
+ }
+
+ /**
+ * For a given list of HBase write futures, this method will asynchronously analyze their
+ * result, and using the provided {@link ResultHandler}, it will instruct {@link
+ * org.apache.flink.connector.base.sink.writer.AsyncSinkWriter} to retry some mutations. In case
+ * of errors which should not be retried, the Flink job will stop with an error.
+ *
+ * @param futures list of HBase write futures
+ * @param processedMutationsInOrder list of mutations with their indices matching that of the
+ * futures
+ * @param resultHandler result handler to manage retries and exceptions
+ */
+ public void handleWriteFutures(
+ List<CompletableFuture<Mutation>> futures,
+ List<SerializableMutation> processedMutationsInOrder,
+ ResultHandler<SerializableMutation> resultHandler) {
+ Preconditions.checkArgument(
+ futures.size() == processedMutationsInOrder.size(),
+ "Different number of HBase futures was supplied than mutations.");
+
+ ConcurrentLinkedQueue<FailedMutation> failedMutations = new ConcurrentLinkedQueue<>();
+
+ // Handle each future separately and store failures.
+ CompletableFuture<?>[] handledFutures = new CompletableFuture[futures.size()];
+ for (int i = 0; i < futures.size(); i++) {
+ final int index = i;
+ handledFutures[index] =
+ futures.get(index)
+ .exceptionally(
+ throwable -> {
+ failedMutations.add(
+ new FailedMutation(
+ processedMutationsInOrder.get(index),
+ throwable));
+ return null;
+ });
+ }
+
+ // Exceptions are already handled here, so it's safe to use `thenRun()`.
+ CompletableFuture.allOf(handledFutures)
+ .thenRun(
+ () -> {
+ handleFailedRequests(failedMutations, resultHandler);
+ });
+ }
+
+ /**
+ * Handles mutations that failed to write to HBase.
+ *
+ * <p>This method increments the error counter and schedules the failed mutations for retry
+ * through the result handler. If the exception should not be retried, the job will fail instead
+ * with an exception.
+ *
+ * @param failedMutations the list of mutations that failed to write
+ * @param resultHandler the handler responsible for retry logic
+ */
+ private void handleFailedRequests(
Review Comment:
Good idea! I have added that field to `SerializableMutation`, as that's going to be persisted in the state when the `AsyncSinkWriter` saves it to its internal buffer, that way we can follow the number of writes already attempted.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-35280] Migrate HBase Sink to use Async Sink API [flink-connector-hbase]
Posted by "ferenc-csaky (via GitHub)" <gi...@apache.org>.
ferenc-csaky merged PR #57:
URL: https://github.com/apache/flink-connector-hbase/pull/57
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-35280] Migrate HBase Sink to use Async Sink API [flink-connector-hbase]
Posted by "ferenc-csaky (via GitHub)" <gi...@apache.org>.
ferenc-csaky commented on code in PR #57:
URL: https://github.com/apache/flink-connector-hbase/pull/57#discussion_r2537016822
##########
flink-connector-hbase-base/src/main/java/org/apache/flink/connector/hbase/util/HBaseMutationSerialization.java:
##########
@@ -0,0 +1,45 @@
+package org.apache.flink.connector.hbase.util;
+
+import org.apache.flink.annotation.Internal;
+
+import org.apache.hadoop.hbase.client.Delete;
+import org.apache.hadoop.hbase.client.Mutation;
+import org.apache.hadoop.hbase.client.Put;
+import org.apache.hadoop.hbase.protobuf.ProtobufUtil;
+import org.apache.hadoop.hbase.protobuf.generated.ClientProtos;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+
+/**
+ * Internal utility class for serializing and deserializing HBase mutations.
+ *
+ * <p>This class provides methods to convert HBase {@link Mutation} objects to and from their
+ * Protocol Buffer representations for transmission over the wire or storage. It supports the
+ * following HBase mutation types: {@link Put} and {@link Delete}.
+ */
+@Internal
+public class HBaseMutationSerialization {
Review Comment:
Any reason to not put this logic to `HBaseStateSerializer`? That's the only class using this util, so I do not see why we need this separate util.
Also, would it make sense to add some unit tests for this ser/deser logic, or its functionality already tested properly by IT cases?
##########
flink-connector-hbase-base/src/main/java/org/apache/flink/connector/hbase/sink/SerializableMutation.java:
##########
@@ -0,0 +1,27 @@
+package org.apache.flink.connector.hbase.sink;
+
+import org.apache.flink.annotation.Internal;
+
+import org.apache.hadoop.hbase.client.Mutation;
+
+import java.io.Serializable;
+
+/**
+ * This class is used by {@link HBaseSink} and {@link HBaseWriter} to wrap HBase {@link Mutation}
+ * objects to be able to serialize them.
+ */
+@Internal
+public class SerializableMutation implements Serializable {
Review Comment:
Maybe put this to `util` package instead?
##########
flink-connector-hbase-base/src/main/java/org/apache/flink/connector/hbase/sink/HBaseWriterAsyncHandler.java:
##########
@@ -0,0 +1,156 @@
+package org.apache.flink.connector.hbase.sink;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.connector.base.sink.writer.ResultHandler;
+import org.apache.flink.metrics.Counter;
+import org.apache.flink.util.Preconditions;
+
+import org.apache.hadoop.hbase.DoNotRetryIOException;
+import org.apache.hadoop.hbase.client.Mutation;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.stream.Collectors;
+
+/**
+ * This class is responsible for managing the async calls to HBase and managing the {@link
+ * ResultHandler} to decide which request can be retried.
+ */
+@Internal
+public class HBaseWriterAsyncHandler {
+ private static final Logger LOG = LoggerFactory.getLogger(HBaseWriterAsyncHandler.class);
+
+ private final Counter numRecordsOutErrorsCounter;
+
+ public HBaseWriterAsyncHandler(Counter numRecordsOutErrorsCounter) {
+ this.numRecordsOutErrorsCounter = numRecordsOutErrorsCounter;
+ }
+
+ /**
+ * For a given list of HBase write futures, this method will asynchronously analyze their
+ * result, and using the provided {@link ResultHandler}, it will instruct {@link
+ * org.apache.flink.connector.base.sink.writer.AsyncSinkWriter} to retry some mutations. In case
+ * of errors which should not be retried, the Flink job will stop with an error.
+ *
+ * @param futures list of HBase write futures
+ * @param processedMutationsInOrder list of mutations with their indices matching that of the
+ * futures
+ * @param resultHandler result handler to manage retries and exceptions
+ */
+ public void handleWriteFutures(
+ List<CompletableFuture<Mutation>> futures,
+ List<SerializableMutation> processedMutationsInOrder,
+ ResultHandler<SerializableMutation> resultHandler) {
+ Preconditions.checkArgument(
+ futures.size() == processedMutationsInOrder.size(),
+ "Different number of HBase futures was supplied than mutations.");
+
+ ConcurrentLinkedQueue<FailedMutation> failedMutations = new ConcurrentLinkedQueue<>();
Review Comment:
nit: `Queue<FailedMutation>`
##########
flink-connector-hbase-base/src/main/java/org/apache/flink/connector/hbase/sink/HBaseWriterAsyncHandler.java:
##########
@@ -0,0 +1,156 @@
+package org.apache.flink.connector.hbase.sink;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.connector.base.sink.writer.ResultHandler;
+import org.apache.flink.metrics.Counter;
+import org.apache.flink.util.Preconditions;
+
+import org.apache.hadoop.hbase.DoNotRetryIOException;
+import org.apache.hadoop.hbase.client.Mutation;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.stream.Collectors;
+
+/**
+ * This class is responsible for managing the async calls to HBase and managing the {@link
+ * ResultHandler} to decide which request can be retried.
+ */
+@Internal
+public class HBaseWriterAsyncHandler {
+ private static final Logger LOG = LoggerFactory.getLogger(HBaseWriterAsyncHandler.class);
+
+ private final Counter numRecordsOutErrorsCounter;
+
+ public HBaseWriterAsyncHandler(Counter numRecordsOutErrorsCounter) {
+ this.numRecordsOutErrorsCounter = numRecordsOutErrorsCounter;
+ }
+
+ /**
+ * For a given list of HBase write futures, this method will asynchronously analyze their
+ * result, and using the provided {@link ResultHandler}, it will instruct {@link
+ * org.apache.flink.connector.base.sink.writer.AsyncSinkWriter} to retry some mutations. In case
+ * of errors which should not be retried, the Flink job will stop with an error.
+ *
+ * @param futures list of HBase write futures
+ * @param processedMutationsInOrder list of mutations with their indices matching that of the
+ * futures
+ * @param resultHandler result handler to manage retries and exceptions
+ */
+ public void handleWriteFutures(
+ List<CompletableFuture<Mutation>> futures,
+ List<SerializableMutation> processedMutationsInOrder,
+ ResultHandler<SerializableMutation> resultHandler) {
+ Preconditions.checkArgument(
+ futures.size() == processedMutationsInOrder.size(),
+ "Different number of HBase futures was supplied than mutations.");
+
+ ConcurrentLinkedQueue<FailedMutation> failedMutations = new ConcurrentLinkedQueue<>();
+
+ // Handle each future separately and store failures.
+ CompletableFuture<?>[] handledFutures = new CompletableFuture[futures.size()];
+ for (int i = 0; i < futures.size(); i++) {
+ final int index = i;
+ handledFutures[index] =
+ futures.get(index)
+ .exceptionally(
+ throwable -> {
+ failedMutations.add(
+ new FailedMutation(
+ processedMutationsInOrder.get(index),
+ throwable));
+ return null;
+ });
+ }
+
+ // Exceptions are already handled here, so it's safe to use `thenRun()`.
+ CompletableFuture.allOf(handledFutures)
+ .thenRun(
+ () -> {
+ handleFailedRequests(failedMutations, resultHandler);
+ });
+ }
+
+ /**
+ * Handles mutations that failed to write to HBase.
+ *
+ * <p>This method increments the error counter and schedules the failed mutations for retry
+ * through the result handler. If the exception should not be retried, the job will fail instead
+ * with an exception.
+ *
+ * @param failedMutations the list of mutations that failed to write
+ * @param resultHandler the handler responsible for retry logic
+ */
+ private void handleFailedRequests(
Review Comment:
Can this get into an endless loop when some mutations always fail and we do not detect the given ex as fatal? :)
##########
flink-connector-hbase-base/src/main/java/org/apache/flink/connector/hbase/sink/HBaseWriterAsyncHandler.java:
##########
@@ -0,0 +1,156 @@
+package org.apache.flink.connector.hbase.sink;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.connector.base.sink.writer.ResultHandler;
+import org.apache.flink.metrics.Counter;
+import org.apache.flink.util.Preconditions;
+
+import org.apache.hadoop.hbase.DoNotRetryIOException;
+import org.apache.hadoop.hbase.client.Mutation;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.stream.Collectors;
+
+/**
+ * This class is responsible for managing the async calls to HBase and managing the {@link
+ * ResultHandler} to decide which request can be retried.
+ */
+@Internal
+public class HBaseWriterAsyncHandler {
+ private static final Logger LOG = LoggerFactory.getLogger(HBaseWriterAsyncHandler.class);
+
+ private final Counter numRecordsOutErrorsCounter;
+
+ public HBaseWriterAsyncHandler(Counter numRecordsOutErrorsCounter) {
+ this.numRecordsOutErrorsCounter = numRecordsOutErrorsCounter;
+ }
+
+ /**
+ * For a given list of HBase write futures, this method will asynchronously analyze their
+ * result, and using the provided {@link ResultHandler}, it will instruct {@link
+ * org.apache.flink.connector.base.sink.writer.AsyncSinkWriter} to retry some mutations. In case
+ * of errors which should not be retried, the Flink job will stop with an error.
+ *
+ * @param futures list of HBase write futures
+ * @param processedMutationsInOrder list of mutations with their indices matching that of the
+ * futures
+ * @param resultHandler result handler to manage retries and exceptions
+ */
+ public void handleWriteFutures(
+ List<CompletableFuture<Mutation>> futures,
+ List<SerializableMutation> processedMutationsInOrder,
+ ResultHandler<SerializableMutation> resultHandler) {
+ Preconditions.checkArgument(
+ futures.size() == processedMutationsInOrder.size(),
+ "Different number of HBase futures was supplied than mutations.");
+
+ ConcurrentLinkedQueue<FailedMutation> failedMutations = new ConcurrentLinkedQueue<>();
+
+ // Handle each future separately and store failures.
+ CompletableFuture<?>[] handledFutures = new CompletableFuture[futures.size()];
+ for (int i = 0; i < futures.size(); i++) {
+ final int index = i;
+ handledFutures[index] =
+ futures.get(index)
+ .exceptionally(
+ throwable -> {
+ failedMutations.add(
+ new FailedMutation(
+ processedMutationsInOrder.get(index),
+ throwable));
+ return null;
+ });
+ }
+
+ // Exceptions are already handled here, so it's safe to use `thenRun()`.
+ CompletableFuture.allOf(handledFutures)
+ .thenRun(
+ () -> {
Review Comment:
nit: remove curly braces
##########
flink-connector-hbase-base/src/main/java/org/apache/flink/connector/hbase/sink/WrappedElementConverter.java:
##########
@@ -0,0 +1,36 @@
+package org.apache.flink.connector.hbase.sink;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.connector.sink2.Sink;
+import org.apache.flink.api.connector.sink2.SinkWriter;
+import org.apache.flink.connector.base.sink.writer.ElementConverter;
+
+import org.apache.hadoop.hbase.client.Mutation;
+
+/**
+ * This is a helper class used to wrap an {@link ElementConverter} supplied by the user that
+ * converts the input data to {@link Mutation}. With this class, the elements will be seamlessly
+ * converted to internal {@link SerializableMutation} objects that can be serialized by the sink.
+ */
+@Internal
+public class WrappedElementConverter<InputT>
Review Comment:
Maybe put this to the `util` package?
##########
flink-connector-hbase-base/src/main/java/org/apache/flink/connector/hbase/sink/HBaseWriterAsyncHandler.java:
##########
@@ -0,0 +1,156 @@
+package org.apache.flink.connector.hbase.sink;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.connector.base.sink.writer.ResultHandler;
+import org.apache.flink.metrics.Counter;
+import org.apache.flink.util.Preconditions;
+
+import org.apache.hadoop.hbase.DoNotRetryIOException;
+import org.apache.hadoop.hbase.client.Mutation;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.stream.Collectors;
+
+/**
+ * This class is responsible for managing the async calls to HBase and managing the {@link
+ * ResultHandler} to decide which request can be retried.
+ */
+@Internal
+public class HBaseWriterAsyncHandler {
+ private static final Logger LOG = LoggerFactory.getLogger(HBaseWriterAsyncHandler.class);
+
+ private final Counter numRecordsOutErrorsCounter;
+
+ public HBaseWriterAsyncHandler(Counter numRecordsOutErrorsCounter) {
+ this.numRecordsOutErrorsCounter = numRecordsOutErrorsCounter;
+ }
+
+ /**
+ * For a given list of HBase write futures, this method will asynchronously analyze their
+ * result, and using the provided {@link ResultHandler}, it will instruct {@link
+ * org.apache.flink.connector.base.sink.writer.AsyncSinkWriter} to retry some mutations. In case
+ * of errors which should not be retried, the Flink job will stop with an error.
+ *
+ * @param futures list of HBase write futures
+ * @param processedMutationsInOrder list of mutations with their indices matching that of the
+ * futures
+ * @param resultHandler result handler to manage retries and exceptions
+ */
+ public void handleWriteFutures(
+ List<CompletableFuture<Mutation>> futures,
+ List<SerializableMutation> processedMutationsInOrder,
+ ResultHandler<SerializableMutation> resultHandler) {
+ Preconditions.checkArgument(
Review Comment:
nit: static import `Preconditions`
##########
flink-connector-hbase-base/src/main/java/org/apache/flink/connector/hbase/sink/SerializableMutation.java:
##########
@@ -0,0 +1,27 @@
+package org.apache.flink.connector.hbase.sink;
+
+import org.apache.flink.annotation.Internal;
+
+import org.apache.hadoop.hbase.client.Mutation;
+
+import java.io.Serializable;
+
+/**
+ * This class is used by {@link HBaseSink} and {@link HBaseWriter} to wrap HBase {@link Mutation}
+ * objects to be able to serialize them.
+ */
+@Internal
+public class SerializableMutation implements Serializable {
+ private static final long serialVersionUID = 1L;
+
+ private transient Mutation mutation;
Review Comment:
This can be `final`.
##########
flink-connector-hbase-base/src/main/java/org/apache/flink/connector/hbase/sink/HBaseWriterAsyncHandler.java:
##########
@@ -0,0 +1,156 @@
+package org.apache.flink.connector.hbase.sink;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.connector.base.sink.writer.ResultHandler;
+import org.apache.flink.metrics.Counter;
+import org.apache.flink.util.Preconditions;
+
+import org.apache.hadoop.hbase.DoNotRetryIOException;
+import org.apache.hadoop.hbase.client.Mutation;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.stream.Collectors;
+
+/**
+ * This class is responsible for managing the async calls to HBase and managing the {@link
+ * ResultHandler} to decide which request can be retried.
+ */
+@Internal
+public class HBaseWriterAsyncHandler {
+ private static final Logger LOG = LoggerFactory.getLogger(HBaseWriterAsyncHandler.class);
+
+ private final Counter numRecordsOutErrorsCounter;
+
+ public HBaseWriterAsyncHandler(Counter numRecordsOutErrorsCounter) {
+ this.numRecordsOutErrorsCounter = numRecordsOutErrorsCounter;
+ }
+
+ /**
+ * For a given list of HBase write futures, this method will asynchronously analyze their
+ * result, and using the provided {@link ResultHandler}, it will instruct {@link
+ * org.apache.flink.connector.base.sink.writer.AsyncSinkWriter} to retry some mutations. In case
+ * of errors which should not be retried, the Flink job will stop with an error.
+ *
+ * @param futures list of HBase write futures
+ * @param processedMutationsInOrder list of mutations with their indices matching that of the
+ * futures
+ * @param resultHandler result handler to manage retries and exceptions
+ */
+ public void handleWriteFutures(
+ List<CompletableFuture<Mutation>> futures,
+ List<SerializableMutation> processedMutationsInOrder,
+ ResultHandler<SerializableMutation> resultHandler) {
+ Preconditions.checkArgument(
+ futures.size() == processedMutationsInOrder.size(),
+ "Different number of HBase futures was supplied than mutations.");
+
+ ConcurrentLinkedQueue<FailedMutation> failedMutations = new ConcurrentLinkedQueue<>();
+
+ // Handle each future separately and store failures.
+ CompletableFuture<?>[] handledFutures = new CompletableFuture[futures.size()];
+ for (int i = 0; i < futures.size(); i++) {
+ final int index = i;
+ handledFutures[index] =
+ futures.get(index)
+ .exceptionally(
+ throwable -> {
+ failedMutations.add(
+ new FailedMutation(
+ processedMutationsInOrder.get(index),
+ throwable));
+ return null;
+ });
+ }
Review Comment:
Maybe doing it via the Java stream API and adding a separate method as the body of exceptionally makes it a bit cleaner:
```java
// Handle each future separately and store failures.
CompletableFuture<?>[] handledFutures =
IntStream.range(0, futures.size())
.mapToObj(
i ->
addFailedCallback(
futures.get(i),
failedMutations,
processedMutationsInOrder.get(i)))
.toArray(CompletableFuture[]::new);
...
private CompletableFuture<Mutation> addFailedCallback(
CompletableFuture<Mutation> cf,
Queue<FailedMutation> failedMutations,
SerializableMutation failedMutation) {
return cf.exceptionally(
throwable -> {
failedMutations.add(new FailedMutation(failedMutation, throwable));
return null;
});
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-35280] Migrate HBase Sink to use Async Sink API [flink-connector-hbase]
Posted by "ferenc-csaky (via GitHub)" <gi...@apache.org>.
ferenc-csaky commented on code in PR #57:
URL: https://github.com/apache/flink-connector-hbase/pull/57#discussion_r2556020867
##########
flink-connector-hbase-base/src/main/java/org/apache/flink/connector/hbase/sink/HBaseWriterAsyncHandler.java:
##########
@@ -0,0 +1,156 @@
+package org.apache.flink.connector.hbase.sink;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.connector.base.sink.writer.ResultHandler;
+import org.apache.flink.metrics.Counter;
+import org.apache.flink.util.Preconditions;
+
+import org.apache.hadoop.hbase.DoNotRetryIOException;
+import org.apache.hadoop.hbase.client.Mutation;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.stream.Collectors;
+
+/**
+ * This class is responsible for managing the async calls to HBase and managing the {@link
+ * ResultHandler} to decide which request can be retried.
+ */
+@Internal
+public class HBaseWriterAsyncHandler {
+ private static final Logger LOG = LoggerFactory.getLogger(HBaseWriterAsyncHandler.class);
+
+ private final Counter numRecordsOutErrorsCounter;
+
+ public HBaseWriterAsyncHandler(Counter numRecordsOutErrorsCounter) {
+ this.numRecordsOutErrorsCounter = numRecordsOutErrorsCounter;
+ }
+
+ /**
+ * For a given list of HBase write futures, this method will asynchronously analyze their
+ * result, and using the provided {@link ResultHandler}, it will instruct {@link
+ * org.apache.flink.connector.base.sink.writer.AsyncSinkWriter} to retry some mutations. In case
+ * of errors which should not be retried, the Flink job will stop with an error.
+ *
+ * @param futures list of HBase write futures
+ * @param processedMutationsInOrder list of mutations with their indices matching that of the
+ * futures
+ * @param resultHandler result handler to manage retries and exceptions
+ */
+ public void handleWriteFutures(
+ List<CompletableFuture<Mutation>> futures,
+ List<SerializableMutation> processedMutationsInOrder,
+ ResultHandler<SerializableMutation> resultHandler) {
+ Preconditions.checkArgument(
+ futures.size() == processedMutationsInOrder.size(),
+ "Different number of HBase futures was supplied than mutations.");
+
+ ConcurrentLinkedQueue<FailedMutation> failedMutations = new ConcurrentLinkedQueue<>();
+
+ // Handle each future separately and store failures.
+ CompletableFuture<?>[] handledFutures = new CompletableFuture[futures.size()];
+ for (int i = 0; i < futures.size(); i++) {
+ final int index = i;
+ handledFutures[index] =
+ futures.get(index)
+ .exceptionally(
+ throwable -> {
+ failedMutations.add(
+ new FailedMutation(
+ processedMutationsInOrder.get(index),
+ throwable));
+ return null;
+ });
+ }
+
+ // Exceptions are already handled here, so it's safe to use `thenRun()`.
+ CompletableFuture.allOf(handledFutures)
+ .thenRun(
+ () -> {
+ handleFailedRequests(failedMutations, resultHandler);
+ });
+ }
+
+ /**
+ * Handles mutations that failed to write to HBase.
+ *
+ * <p>This method increments the error counter and schedules the failed mutations for retry
+ * through the result handler. If the exception should not be retried, the job will fail instead
+ * with an exception.
+ *
+ * @param failedMutations the list of mutations that failed to write
+ * @param resultHandler the handler responsible for retry logic
+ */
+ private void handleFailedRequests(
Review Comment:
Maybe a less destructive solution would be better here. I agree that dropping those requests are not acceptable, but I think we could introduce a some kind of "max request retry" option, so we store a counter for every failed request, and if that counter reaches the configured max for one request, only then we blow up the job.
I also feel like this would be an edge-case, so I think this config option should be optional, and by default we can keep the current behavior. And e.g. we can make that counter nullable and an optional and if the "max request retry" is not configured (null or 0, whatever makes more sense from an impl perspective), then we don't even need to manage that counter.
WDYT?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-35280] Migrate HBase Sink to use Async Sink API [flink-connector-hbase]
Posted by "mateczagany (via GitHub)" <gi...@apache.org>.
mateczagany commented on code in PR #57:
URL: https://github.com/apache/flink-connector-hbase/pull/57#discussion_r2555261639
##########
flink-connector-hbase-base/src/main/java/org/apache/flink/connector/hbase/sink/HBaseWriterAsyncHandler.java:
##########
@@ -0,0 +1,156 @@
+package org.apache.flink.connector.hbase.sink;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.connector.base.sink.writer.ResultHandler;
+import org.apache.flink.metrics.Counter;
+import org.apache.flink.util.Preconditions;
+
+import org.apache.hadoop.hbase.DoNotRetryIOException;
+import org.apache.hadoop.hbase.client.Mutation;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.stream.Collectors;
+
+/**
+ * This class is responsible for managing the async calls to HBase and managing the {@link
+ * ResultHandler} to decide which request can be retried.
+ */
+@Internal
+public class HBaseWriterAsyncHandler {
+ private static final Logger LOG = LoggerFactory.getLogger(HBaseWriterAsyncHandler.class);
+
+ private final Counter numRecordsOutErrorsCounter;
+
+ public HBaseWriterAsyncHandler(Counter numRecordsOutErrorsCounter) {
+ this.numRecordsOutErrorsCounter = numRecordsOutErrorsCounter;
+ }
+
+ /**
+ * For a given list of HBase write futures, this method will asynchronously analyze their
+ * result, and using the provided {@link ResultHandler}, it will instruct {@link
+ * org.apache.flink.connector.base.sink.writer.AsyncSinkWriter} to retry some mutations. In case
+ * of errors which should not be retried, the Flink job will stop with an error.
+ *
+ * @param futures list of HBase write futures
+ * @param processedMutationsInOrder list of mutations with their indices matching that of the
+ * futures
+ * @param resultHandler result handler to manage retries and exceptions
+ */
+ public void handleWriteFutures(
+ List<CompletableFuture<Mutation>> futures,
+ List<SerializableMutation> processedMutationsInOrder,
+ ResultHandler<SerializableMutation> resultHandler) {
+ Preconditions.checkArgument(
+ futures.size() == processedMutationsInOrder.size(),
+ "Different number of HBase futures was supplied than mutations.");
+
+ ConcurrentLinkedQueue<FailedMutation> failedMutations = new ConcurrentLinkedQueue<>();
+
+ // Handle each future separately and store failures.
+ CompletableFuture<?>[] handledFutures = new CompletableFuture[futures.size()];
+ for (int i = 0; i < futures.size(); i++) {
+ final int index = i;
+ handledFutures[index] =
+ futures.get(index)
+ .exceptionally(
+ throwable -> {
+ failedMutations.add(
+ new FailedMutation(
+ processedMutationsInOrder.get(index),
+ throwable));
+ return null;
+ });
+ }
+
+ // Exceptions are already handled here, so it's safe to use `thenRun()`.
+ CompletableFuture.allOf(handledFutures)
+ .thenRun(
+ () -> {
+ handleFailedRequests(failedMutations, resultHandler);
+ });
+ }
+
+ /**
+ * Handles mutations that failed to write to HBase.
+ *
+ * <p>This method increments the error counter and schedules the failed mutations for retry
+ * through the result handler. If the exception should not be retried, the job will fail instead
+ * with an exception.
+ *
+ * @param failedMutations the list of mutations that failed to write
+ * @param resultHandler the handler responsible for retry logic
+ */
+ private void handleFailedRequests(
Review Comment:
That can definitely happen. I don't know if that's a bad thing, or how often can a job actually run into a non-fatal failure, e.g. `MasterNotRunningException`.
IIRC, the Async Sink hangs checkpoints until all in-flight requests are finished, but won't block a checkpoint from finishing if there are any failed, retryable requests. It will add those requests to its Deque at the head, and retry them after `maxTimeInBufferMS` in processing time.
I don't think it's appropriate to drop these requests, maybe it could be configurable to fail the job if any exception is thrown? (even if theoretically retryable according to the HBase client)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-35280] Migrate HBase Sink to use Async Sink API [flink-connector-hbase]
Posted by "mateczagany (via GitHub)" <gi...@apache.org>.
mateczagany commented on code in PR #57:
URL: https://github.com/apache/flink-connector-hbase/pull/57#discussion_r2555095620
##########
flink-connector-hbase-base/src/main/java/org/apache/flink/connector/hbase/util/HBaseMutationSerialization.java:
##########
@@ -0,0 +1,45 @@
+package org.apache.flink.connector.hbase.util;
+
+import org.apache.flink.annotation.Internal;
+
+import org.apache.hadoop.hbase.client.Delete;
+import org.apache.hadoop.hbase.client.Mutation;
+import org.apache.hadoop.hbase.client.Put;
+import org.apache.hadoop.hbase.protobuf.ProtobufUtil;
+import org.apache.hadoop.hbase.protobuf.generated.ClientProtos;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+
+/**
+ * Internal utility class for serializing and deserializing HBase mutations.
+ *
+ * <p>This class provides methods to convert HBase {@link Mutation} objects to and from their
+ * Protocol Buffer representations for transmission over the wire or storage. It supports the
+ * following HBase mutation types: {@link Put} and {@link Delete}.
+ */
+@Internal
+public class HBaseMutationSerialization {
Review Comment:
You are right, it's only used by that class. This is tested by the IT cases, but I've also added some unit tests.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-35280] Migrate HBase Sink to use Async Sink API [flink-connector-hbase]
Posted by "mateczagany (via GitHub)" <gi...@apache.org>.
mateczagany commented on PR #57:
URL: https://github.com/apache/flink-connector-hbase/pull/57#issuecomment-3570068696
I have upgraded HBase to 2.6.4, as the IT test was failing because 2.6.3 has been removed from the public repository: https://dlcdn.apache.org/hbase/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-35280] Migrate HBase Sink to use Async Sink API [flink-connector-hbase]
Posted by "mateczagany (via GitHub)" <gi...@apache.org>.
mateczagany commented on PR #57:
URL: https://github.com/apache/flink-connector-hbase/pull/57#issuecomment-3584495705
Thank you, I forgot about that.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org