[hotfix][connector] Fix sink not stop when exception is occurs.

[loserwang1024]

Purpose

in #283, there is also an abnormal log for multiple times:

WARN  com.alibaba.fluss.client.write.Sender                        [] - Get error write response on table bucket TableBucket{tableId=2, bucket=152}, fail. Error: UNKNOWN_SERVER_ERROR. Error Message: java.lang.IllegalArgumentException: The log sequence number must be non-decreasing. The current log sequence number is 600582, but the new log sequence number is 600581
	at com.alibaba.fluss.server.kv.prewrite.KvPreWriteBuffer.update(KvPreWriteBuffer.java:126)

the coresponing code is com.alibaba.fluss.client.write.Sender#handleWriteBatchException.

    private Set<PhysicalTablePath> handleWriteBatchException(
            WriteBatch writeBatch, ApiError error) {
        Set<PhysicalTablePath> invalidMetadataTables = new HashSet<>();
        if (canRetry(writeBatch, error.error())) {
            // ....ignore
        } else if (error.error() == Errors.DUPLICATE_SEQUENCE_EXCEPTION) {
          // ...
        } else {
            LOG.warn(
                    "Get error write response on table bucket {}, fail. Error: {}",
                    writeBatch.tableBucket(),
                    error.formatErrMsg());
            // tell the user the result of their request. We only adjust batch sequence if the
            // batch didn't exhaust its retries -- if it did, we don't know whether the batch
            // sequence was accepted or not, and thus it is not safe to reassign the sequence.
            failBatch(writeBatch, error.exception(), writeBatch.attempts() < this.retries);
        }
        return invalidMetadataTables;
    }

It means that IllegalArgumentException is not a triable exception, which will be thrown out of the client. However, the flink job didn't stop but stuck to allocate memory.

Then, I find some thing wrong in sinkfunction: asyncWriterException can never be set.

 @Override
    public void invoke(RowData value, SinkFunction.Context context) throws IOException {
        checkAsyncException();
        InternalRow internalRow = dataConverter.toInternalRow(value);
        CompletableFuture<Void> writeFuture = writeRow(value.getRowKind(), internalRow);
        writeFuture.exceptionally(
                exception -> {
                    if (this.asyncWriterException != null) {
                        this.asyncWriterException = exception;
                    }
                    return null;
                });

        numRecordsOutCounter.inc();
    }

[luoyuxia]

@loserwang1024 is it checked that it will cause job failover when exception is thrown?

[swuferhong]

@loserwang1024 is it checked that it will cause job failover when exception is thrown?

+1, I think the root cause of this error still need to find?