[hotfix][pulsar] PulsarSource consumer ack exception. (#4237)

* fix pulsar source ack bug.

* update doc.

---------

Co-authored-by: lightzhao <zhaolianyong777@gmail.com>

Author: lightzhao

docs/en/connector-v2/Error-Quick-Reference-Manual.md

@@ -208,6 +208,7 @@ problems encountered by users.
 | PULSAR-04 | Subscribe topic from pulsar failed               | When users encounter this error code, it means that Subscribe topic from pulsar failed, please check it               |
 | PULSAR-05 | Get last cursor of pulsar topic failed           | When users encounter this error code, it means that get last cursor of pulsar topic failed, please check it           |
 | PULSAR-06 | Get partition information of pulsar topic failed | When users encounter this error code, it means that Get partition information of pulsar topic failed, please check it |
+| PULSAR-07 | Pulsar consumer acknowledgeCumulative failed     | When users encounter this error code, it means that Pulsar consumer acknowledgeCumulative failed                      |
 
 ## StarRocks Connector Error Codes
 

seatunnel-connectors-v2/connector-pulsar/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/exception/PulsarConnectorErrorCode.java

@@ -25,7 +25,8 @@ public enum PulsarConnectorErrorCode implements SeaTunnelErrorCode {
     PULSAR_AUTHENTICATION_FAILED("PULSAR-03", "Pulsar authentication failed"),
     SUBSCRIBE_TOPIC_FAILED("PULSAR-04", "Subscribe topic from pulsar failed"),
     GET_LAST_CURSOR_FAILED("PULSAR-05", "Get last cursor of pulsar topic failed"),
-    GET_TOPIC_PARTITION_FAILED("PULSAR-06", "Get partition information of pulsar topic failed");
+    GET_TOPIC_PARTITION_FAILED("PULSAR-06", "Get partition information of pulsar topic failed"),
+    ACK_CUMULATE_FAILED("PULSAR-07", "Pulsar consumer acknowledgeCumulative failed");
 
     private final String code;
     private final String description;

seatunnel-connectors-v2/connector-pulsar/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/source/reader/PulsarSourceReader.java

@@ -25,6 +25,7 @@
 import org.apache.seatunnel.connectors.seatunnel.pulsar.config.PulsarClientConfig;
 import org.apache.seatunnel.connectors.seatunnel.pulsar.config.PulsarConfigUtil;
 import org.apache.seatunnel.connectors.seatunnel.pulsar.config.PulsarConsumerConfig;
+import org.apache.seatunnel.connectors.seatunnel.pulsar.exception.PulsarConnectorErrorCode;
 import org.apache.seatunnel.connectors.seatunnel.pulsar.exception.PulsarConnectorException;
 import org.apache.seatunnel.connectors.seatunnel.pulsar.source.enumerator.cursor.start.StartCursor;
 import org.apache.seatunnel.connectors.seatunnel.pulsar.source.split.PulsarPartitionSplit;
@@ -225,20 +226,26 @@ public void notifyCheckpointComplete(long checkpointId) throws Exception {
                     if (finishedSplits.contains(splitId)) {
                         return;
                     }
+                    try {
+                        splitReaders.get(splitId).committingCursor(messageId);
 
-                    splitReaders.get(splitId).committingCursor(messageId);
-
-                    if (pendingCursorsToFinish.containsKey(splitId)
-                            && pendingCursorsToFinish.get(splitId).compareTo(messageId) == 0) {
-                        finishedSplits.add(splitId);
-                        try {
-                            splitReaders.get(splitId).close();
-                        } catch (IOException e) {
-                            throw new PulsarConnectorException(
-                                    CommonErrorCode.READER_OPERATION_FAILED,
-                                    "Failed to close the split reader thread.",
-                                    e);
+                        if (pendingCursorsToFinish.containsKey(splitId)
+                                && pendingCursorsToFinish.get(splitId).compareTo(messageId) == 0) {
+                            finishedSplits.add(splitId);
+                            try {
+                                splitReaders.get(splitId).close();
+                            } catch (IOException e) {
+                                throw new PulsarConnectorException(
+                                        CommonErrorCode.READER_OPERATION_FAILED,
+                                        "Failed to close the split reader thread.",
+                                        e);
+                            }
                         }
+                    } catch (PulsarClientException e) {
+                        throw new PulsarConnectorException(
+                                PulsarConnectorErrorCode.ACK_CUMULATE_FAILED,
+                                "pulsar consumer acknowledgeCumulative failed.",
+                                e);
                     }
                 });
     }

seatunnel-connectors-v2/connector-pulsar/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/source/reader/PulsarSplitReaderThread.java

@@ -118,11 +118,11 @@ public void close() throws IOException {
         }
     }
 
-    public void committingCursor(MessageId offsetsToCommit) {
+    public void committingCursor(MessageId offsetsToCommit) throws PulsarClientException {
         if (consumer == null) {
             consumer = createPulsarConsumer(split);
         }
-        consumer.acknowledgeAsync(offsetsToCommit);
+        consumer.acknowledgeCumulative(offsetsToCommit);
     }
 
     /** Create a specified {@link Consumer} by the given split information. */