@Override
    public void invoke(T value, Context context) {
        checkErrorAndRethrow();
        RowData rowData = (RowData) value;
        if (RowKind.UPDATE_BEFORE != rowData.getRowKind()) {
            Mutation mutation = null;
            try {
                mutation = Preconditions.checkNotNull(mutationConverter.convertToMutation(value));
                rowSize++;
                dataSize = dataSize + CalculateObjectSizeUtils.getDataSize(value);
            } catch (Exception e) {
                LOGGER.error("Convert to mutation error", e);
                if (!dirtyOptions.ignoreDirty()) {
                    throw new RuntimeException(e);
                }
                sinkMetricData.invokeDirtyWithEstimate(value);
                if (dirtySink != null) {
                    DirtyData.Builder<Object> builder = DirtyData.builder();
                    try {
                        builder.setData(rowData)
                                .setDirtyType(DirtyType.UNDEFINED)
                                .setLabels(dirtyOptions.getLabels())
                                .setLogTag(dirtyOptions.getLogTag())
                                .setDirtyMessage(e.getMessage())
                                .setIdentifier(dirtyOptions.getIdentifier());
                        dirtySink.invoke(builder.build());
                    } catch (Exception ex) {
                        if (!dirtyOptions.ignoreSideOutputErrors()) {
                            throw new RuntimeException(ex);
                        }
                        LOGGER.warn("Dirty sink failed", ex);
                    }
                }
                return;
            }
            try {
                mutator.mutate(mutation);
            } catch (Exception e) {
                failureThrowable.compareAndSet(null, e);
            }
        } else {
            rowSize++;
            dataSize = dataSize + value.toString().getBytes(StandardCharsets.UTF_8).length;
        }
        // flush when the buffer number of mutations greater than the configured max size.
        if (bufferFlushMaxMutations > 0
                && numPendingRequests.incrementAndGet() >= bufferFlushMaxMutations) {
            reportMetricAfterFlush();
        }
    }
This is set from inside the {@link BufferedMutator.ExceptionListener} if a {@link Throwable}
was thrown.

<p>
Errors will be checked and rethrown before processing each input element, and when the
sink is closed.
</p>
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/hbase/src/main/java/org/apache/inlong/sort/hbase/sink/HBaseSinkFunction.java