Merge branch 'master' into feature/sync_write

Author: zhuogoon

README.md

@@ -7,7 +7,7 @@ Java sdk for PA-REC config server. Aliyun product [link](https://pairec.console.
 <dependency>
   <groupId>com.aliyun.openservices.aiservice</groupId>
   <artifactId>pairec-sdk</artifactId>
-  <version>1.0.7</version>
+  <version>1.0.8</version>
 </dependency>
 ```
 
@@ -66,7 +66,11 @@ public class ExperimentTest {
 
 ```
 ## 版本说明
-1.0.7 (2026-02-03)
+1.0.8 (2026-03-23)
+* 支持InsertMode更新数据，upsert 部分更新字段功能支持KV类型的表
+* write 写入接口支持异步写入
+
+  1.0.7 (2026-02-03)
 * 召回管理client发布 
 
 1.0.6 (2025-11-17)

pom.xml

@@ -5,7 +5,7 @@
   <artifactId>pairec-sdk</artifactId>
   <packaging>jar</packaging>
   <name>pairec-sdk</name>
-  <version>1.0.7</version>
+  <version>1.0.8</version>
   <url>>http://maven.apache.org</url>
   <description>SDK for PAI-REC config service</description>
   <scm>

src/main/java/com/aliyun/openservices/pairec/recallengine/InsertMode.java

@@ -0,0 +1,49 @@
+package com.aliyun.openservices.pairec.recallengine;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonValue;
+
+/**
+ * InsertMode defines the write mode for recall engine.
+ */
+public enum InsertMode {
+    /**
+     * Insert mode: insert new records only, fail if record exists (default)
+     */
+    INSERT("insert"),
+
+    /**
+     * Upsert mode: insert new records or update existing ones
+     */
+    UPSERT("upsert");
+
+    private final String value;
+
+    InsertMode(String value) {
+        this.value = value;
+    }
+
+    @JsonValue
+    public String getValue() {
+        return value;
+    }
+
+    @JsonCreator
+    public static InsertMode fromValue(String value) {
+        if (value == null) {
+            return INSERT;
+        }
+        for (InsertMode mode : InsertMode.values()) {
+            if (mode.value.equalsIgnoreCase(value)) {
+                return mode;
+            }
+        }
+        // Default to INSERT for unknown values
+        return INSERT;
+    }
+
+    @Override
+    public String toString() {
+        return value;
+    }
+}

src/main/java/com/aliyun/openservices/pairec/recallengine/RecallEngineClient.java

@@ -60,9 +60,9 @@ public class RecallEngineClient {
     private final List<WriteItem> writeData = new ArrayList<>();
     private final ReentrantLock writeLock = new ReentrantLock();
     private final Condition writeCondition = writeLock.newCondition();
-    private ExecutorService writeExecutor;
+    private volatile ExecutorService writeExecutor;
     private volatile boolean running = true;
-    private Thread asyncWriteThread;
+    private volatile Thread asyncWriteThread;
     private int batchSize = 20;
     private long flushIntervalMs = 50;
     private WriteCallback globalCallback;
@@ -278,7 +278,7 @@ public WriteResponse write(String instanceId, String table, WriteRequest request
         }
 
         int itemCount = request.getContent().size();
-        String insertMode = request.getInsertMode();
+        InsertMode insertMode = request.getInsertMode();
 
         writeLock.lock();
         try {
@@ -324,7 +324,7 @@ public WriteResponse write(String instanceId, String table, WriteRequest request
         }
 
         int itemCount = request.getContent().size();
-        String insertMode = request.getInsertMode();
+        InsertMode insertMode = request.getInsertMode();
 
         writeLock.lock();
         try {
@@ -471,52 +471,85 @@ public RecallEngineClient withWriteThreadPoolSize(int poolSize) {
     }
 
     /**
-     * Get or create the write executor
+     * Get or create the write executor using Double-Checked Locking for thread safety and performance.
+     * This pattern ensures:
+     * - Thread safety: only one thread creates the executor
+     * - Performance: subsequent calls only need a volatile read (no synchronization)
      */
     private ExecutorService getWriteExecutor() {
-        if (writeExecutor == null || writeExecutor.isShutdown()) {
-            writeExecutor = Executors.newFixedThreadPool(writeThreadPoolSize);
+        ExecutorService executor = writeExecutor;
+        if (executor == null || executor.isShutdown()) {
+            synchronized (this) {
+                executor = writeExecutor;
+                if (executor == null || executor.isShutdown()) {
+                    executor = Executors.newFixedThreadPool(writeThreadPoolSize);
+                    writeExecutor = executor;
+                }
+            }
         }
-        return writeExecutor;
+        return executor;
     }
 
     /**
-     * Start the async write background thread
+     * Start the async write background thread.
+     * Uses Double-Checked Locking for thread safety and performance:
+     * - First check without lock (fast path for most calls)
+     * - Only acquire lock when thread needs to be created
      */
     private void startAsyncWriteThread() {
-        if (asyncWriteThread != null && asyncWriteThread.isAlive()) {
+        // Fast path: if thread is already running, return immediately (no lock needed)
+        Thread thread = asyncWriteThread;
+        if (thread != null && thread.isAlive()) {
             return;
         }
 
-        String threadName = "RecallEngineAsyncWriter";
-        asyncWriteThread = new Thread(() -> {
-            while (running) {
+        // Slow path: need to create thread, use synchronized
+        synchronized (this) {
+            // Double-check after acquiring lock
+            thread = asyncWriteThread;
+            if (thread != null && thread.isAlive()) {
+                return;
+            }
+
+            String threadName = "RecallEngineAsyncWriter";
+            asyncWriteThread = new Thread(() -> {
+                while (running) {
+                    writeLock.lock();
+                    try {
+                        writeCondition.await(flushIntervalMs, TimeUnit.MILLISECONDS);
+                        if (!writeData.isEmpty()) {
+                            doAsyncWrite();
+                        }
+                    } catch (InterruptedException e) {
+                        logger.warn("{} interrupted, flushing remaining data before exit", threadName);
+                        Thread.currentThread().interrupt();
+                        // Flush remaining data before exit
+                        try {
+                            if (!writeData.isEmpty()) {
+                                doAsyncWrite();
+                            }
+                        } catch (Exception ex) {
+                            logger.error("Failed to flush data on interrupt", ex);
+                        }
+                        break;
+                    } finally {
+                        writeLock.unlock();
+                    }
+                }
+                // Handle remaining data after thread stops
                 writeLock.lock();
                 try {
-                    writeCondition.await(flushIntervalMs, TimeUnit.MILLISECONDS);
                     if (!writeData.isEmpty()) {
                         doAsyncWrite();
                     }
-                } catch (InterruptedException e) {
-                    Thread.currentThread().interrupt();
-                    break;
                 } finally {
                     writeLock.unlock();
                 }
-            }
-            // Handle remaining data after thread stops
-            writeLock.lock();
-            try {
-                if (!writeData.isEmpty()) {
-                    doAsyncWrite();
-                }
-            } finally {
-                writeLock.unlock();
-            }
-            logger.info(threadName + " has stopped.");
-        }, threadName);
-        asyncWriteThread.setDaemon(true);
-        asyncWriteThread.start();
+                logger.info("{} has stopped.", threadName);
+            }, threadName);
+            asyncWriteThread.setDaemon(true);
+            asyncWriteThread.start();
+        }
     }
 
     /**
@@ -555,19 +588,19 @@ private Future<?> doAsyncWrite() {
         writeData.clear();
 
         return getWriteExecutor().submit(() -> {
-            // Group by instanceId, table, callback, and insertMode
-            Map<String, List<WriteItem>> groupedItems = new HashMap<>();
+            // Group by instanceId, table, callback, and insertMode using a composite key object
+            Map<GroupKey, List<WriteItem>> groupedItems = new HashMap<>();
             for (WriteItem item : tempList) {
-                // Include callback in the key to group items with same callback together
-                String key = item.instanceId + ":" + item.table + ":" + System.identityHashCode(item.callback)+":" + item.insertMode;
+                GroupKey key = new GroupKey(item.instanceId, item.table, 
+                                            System.identityHashCode(item.callback), item.insertMode);
                 groupedItems.computeIfAbsent(key, k -> new ArrayList<>()).add(item);
             }
 
             // Write each group
-            for (Map.Entry<String, List<WriteItem>> entry : groupedItems.entrySet()) {
-                String[] parts = entry.getKey().split(":", 3);
-                String instanceId = parts[0];
-                String table = parts[1];
+            for (Map.Entry<GroupKey, List<WriteItem>> entry : groupedItems.entrySet()) {
+                GroupKey key = entry.getKey();
+                String instanceId = key.instanceId;
+                String table = key.table;
                 List<WriteItem> items = entry.getValue();
 
                 // Get callback from first item (all items in group have same callback)
@@ -594,18 +627,18 @@ private Future<?> doAsyncWrite() {
                         try {
                             callback.onSuccess(instanceId, table, response);
                         } catch (Exception e) {
-                            logger.error("Callback onError raised exception: {}", e.getMessage());
+                            logger.error("Callback onSuccess raised exception", e);
                         }
                     }
                 } catch (Exception e) {
-                    logger.error("Async write failed for {}/{}: {}", instanceId, table, e.getMessage());
+                    logger.error("Async write failed for {}/{}: {}", instanceId, table, e.getMessage(), e);
 
                     // Invoke error callback
                     if (callback != null) {
                         try {
                             callback.onError(instanceId, table, e);
                         } catch (Exception ex) {
-                            logger.error("Callback onError raised exception: {}", ex.getMessage());
+                            logger.error("Callback onError raised exception", ex);
                         }
                     }
                 }
@@ -659,25 +692,49 @@ private static class WriteItem {
         final String instanceId;
         final String table;
         final Map<String, Object> data;
-        final long timestamp;
         final WriteCallback callback;
-        final String insertMode;
+        final InsertMode insertMode;
 
-        WriteItem(String instanceId, String table, Map<String, Object> data) {
-            this(instanceId, table, data, null, null);
+        WriteItem(String instanceId, String table, Map<String, Object> data, WriteCallback callback, InsertMode insertMode) {
+            this.instanceId = instanceId;
+            this.table = table;
+            this.data = data;
+            this.callback = callback;
+            this.insertMode = insertMode;
         }
+    }
 
-        WriteItem(String instanceId, String table, Map<String, Object> data, WriteCallback callback) {
-            this(instanceId, table, data, callback, null);
-        }
+    /**
+     * Internal class for grouping write items by instanceId, table, callback, and insertMode.
+     * Using a proper key object avoids string parsing issues when instanceId or table contains ":".
+     */
+    private static class GroupKey {
+        final String instanceId;
+        final String table;
+        final int callbackHash;
+        final InsertMode insertMode;
 
-        WriteItem(String instanceId, String table, Map<String, Object> data, WriteCallback callback, String insertMode) {
+        GroupKey(String instanceId, String table, int callbackHash, InsertMode insertMode) {
             this.instanceId = instanceId;
             this.table = table;
-            this.data = data;
-            this.timestamp = System.currentTimeMillis();
-            this.callback = callback;
+            this.callbackHash = callbackHash;
             this.insertMode = insertMode;
         }
+
+        @Override
+        public boolean equals(Object o) {
+            if (this == o) return true;
+            if (o == null || getClass() != o.getClass()) return false;
+            GroupKey groupKey = (GroupKey) o;
+            return callbackHash == groupKey.callbackHash &&
+                   java.util.Objects.equals(instanceId, groupKey.instanceId) &&
+                   java.util.Objects.equals(table, groupKey.table) &&
+                   insertMode == groupKey.insertMode;
+        }
+
+        @Override
+        public int hashCode() {
+            return java.util.Objects.hash(instanceId, table, callbackHash, insertMode);
+        }
     }
 }

src/main/java/com/aliyun/openservices/pairec/recallengine/WriteRequest.java

@@ -17,7 +17,7 @@ public class WriteRequest {
     private String versionId;
 
     @JsonProperty("insert_mode")
-    private String insertMode;
+    private InsertMode insertMode = InsertMode.INSERT;
 
     public WriteRequest() {
     }
@@ -46,11 +46,11 @@ public void setVersionId(String versionId) {
         this.versionId = versionId;
     }
 
-    public String getInsertMode() {
+    public InsertMode getInsertMode() {
         return insertMode;
     }
 
-    public void setInsertMode(String insertMode) {
+    public void setInsertMode(InsertMode insertMode) {
         this.insertMode = insertMode;
     }
 }