dannycranmer commented on a change in pull request #18449:
URL: https://github.com/apache/flink/pull/18449#discussion_r790538487
##########
File path:
flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/sink/AsyncSinkBaseBuilder.java
##########
@@ -37,6 +38,7 @@
ConcreteBuilderT extends AsyncSinkBaseBuilder<?, ?, ?>> {
private ElementConverter<InputT, RequestEntryT> elementConverter;
+ private FatalExceptionHandler<RequestEntryT> fatalExceptionHandler;
Review comment:
I have been discussing with @CrynetLogistics about fully encapsulating
the `RequestEntryT` to the user code. At the moment it is unnecessarily bubbled
up, breaking encapsulation and preventing us to change the backing
implementation with breaking user code. This now poses an interesting challenge
in terms of implementation with `FatalExceptionHandler`. I would like
@CrynetLogistics opinion here on how we can add this without exposing
`RequestEntryT`.
##########
File path:
flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/sink/writer/FatalExceptionHandler.java
##########
@@ -0,0 +1,37 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.connector.base.sink.writer;
+
+import org.apache.flink.annotation.PublicEvolving;
+
+import java.io.Serializable;
+import java.util.List;
+
+/**
+ * This interface specifies how the sink should handle an exception raised by
{@code
+ * AsyncSinkWriter} that should not be retried. The mapping is provided by the
end-user of a sink,
+ * not the sink creator.
+ *
+ * @param <RequestEntryT> Corresponds to the type parameter of the same name
in {@code *
+ * AsyncSinkWriter}
+ */
+@PublicEvolving
+public interface FatalExceptionHandler<RequestEntryT> extends Serializable {
+ void handle(List<RequestEntryT> failedRequestEntries, Exception
fatalException)
+ throws Exception;
Review comment:
Why does this interface require `List<RequestEntryT>`? As I commented
above, this would require user code to understand the underlying
implementation. Arguably, so does the exception class. Is it possible to expose
a single param here, like so?
```
void handle(Exception fatalException)
```
More generally I am wondering if it would be best to not throw here, but
return an Exception classification instead? The base could then act
appropriately. This would allow us to implement other strategies in the future.
For example:
```
ExceptionClassification handle(Exception exception) {
if (exception instanceof VeryBadException) {
return ExceptionClassification.fatal();
} else if (exception instanceof NotSoBad) {
return ExceptionClassification.retryable(); // retry with retry
strategy
} else if (exception instanceof CommonError) {
return ExceptionClassification.transient(); // retry immediately
}
}
```
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
