[incubator-inlong] branch master updated: [INLONG-2385][Feature][Sort] SortStandalone support sort sdk consume batch message (#2968)

[do...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

dockerzhang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-inlong.git


The following commit(s) were added to refs/heads/master by this push:
     new 130ff46  [INLONG-2385][Feature][Sort] SortStandalone support sort sdk consume batch message (#2968)
130ff46 is described below

commit 130ff46af016f5a99ccba759dfb154149c933efa
Author: vernedeng <de...@pku.edu.cn>
AuthorDate: Mon Mar 7 17:20:38 2022 +0800

    [INLONG-2385][Feature][Sort] SortStandalone support sort sdk consume batch message (#2968)
---
 .../sort/standalone/source/sortsdk/FetchCallback.java |  6 +++---
 .../source/sortsdk/SubscribeFetchResult.java          | 19 +++++++++++++++++--
 2 files changed, 20 insertions(+), 5 deletions(-)

diff --git a/inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/source/sortsdk/FetchCallback.java b/inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/source/sortsdk/FetchCallback.java
index 52a8dcb..c603f7c 100644
--- a/inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/source/sortsdk/FetchCallback.java
+++ b/inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/source/sortsdk/FetchCallback.java
@@ -125,11 +125,11 @@ public class FetchCallback implements ReadCallback {
     /**
      * The callback function that SortSDK invoke when fetch messages batch
      *
-     * @param messageRecord {@link List<MessageRecord>}
+     * @param messageRecordList {@link List<MessageRecord>}
      */
     @Override
-    public void onFinishedBatch(List<MessageRecord> messageRecord) {
-        //TODO
+    public void onFinishedBatch(List<MessageRecord> messageRecordList) {
+        messageRecordList.forEach(this::onFinished);
     }
 
     /**
diff --git a/inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/source/sortsdk/SubscribeFetchResult.java b/inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/source/sortsdk/SubscribeFetchResult.java
index 0c6bc21..3572597 100644
--- a/inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/source/sortsdk/SubscribeFetchResult.java
+++ b/inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/source/sortsdk/SubscribeFetchResult.java
@@ -46,7 +46,7 @@ public class SubscribeFetchResult {
      * @param sortId The sortId of fetched message.
      * @param message Message that fetched from upstream data storage.
      */
-
+    @Deprecated
     private SubscribeFetchResult(
             final String sortId,
             final MessageRecord message) {
@@ -122,15 +122,30 @@ public class SubscribeFetchResult {
          * @param messageRecord Message that fetched from upstream data storage.
          * @return One SubscribeFetchResult.
          */
+        @Deprecated
         public static SubscribeFetchResult create(
                 @NotBlank(message = "SortId should not be null or empty.") final String sortId,
                 @NotNull(message = "MessageRecord should not be null.") final MessageRecord messageRecord) {
             return new SubscribeFetchResult(sortId, messageRecord);
         }
 
+        /**
+         * Create one {@link SubscribeFetchResult}.
+         *
+         * @param sortId The sortId of fetched message.
+         * @param msgKey The msgKey to ack.
+         * @param offset The offset of this message.
+         * @param headers Headers of message.
+         * @param recTime Receive time of message.
+         * @param body Data of message.
+         * @return One SubscribeFetchResult.
+         */
         public static SubscribeFetchResult create(
                 final String sortId,
-                final String msgKey, final String offset, final Map<String, String> headers, final long recTime,
+                final String msgKey,
+                final String offset,
+                final Map<String, String> headers,
+                final long recTime,
                 final byte[] body) {
             return new SubscribeFetchResult(sortId, msgKey, offset, headers, recTime, body);
         }