bump 0.50.0-public

Author: dingxin-tech

CHANGELOG.md

@@ -1,5 +1,15 @@
 # Changelog
 
+## [0.50.0-public] - 2024-10-09
+
+### Features
+- Added `SchemaMismatchException`: This exception will be thrown when using `StreamUploadSession` if the Record structure uploaded by the user does not match the table structure. This exception will additionally carry the latest schema version to assist users in rebuilding the Session and performing retry operations.
+- Added `allowSchemaMismatch` method in `StreamUploadSession.Builder`: This method specifies whether to tolerate mismatches between the user's uploaded Record structure and the table structure without throwing an exception. The default value is `true`.
+
+### Fixes
+- Fixed an issue where specifying `tunnelEndpoint` in Odps was ineffective when using `StreamUploadSession`.
+- Fixed a potential NPE issue in `TunnelRetryHandler`.
+
 ## [0.50.0-rc1] - 2024-09-19
 
 ### Features

CHANGELOG_CN.md

@@ -1,4 +1,15 @@
 # 更新日志
+## [0.50.0-public] - 2024-10-09
+
+### 功能
+- 新增 `SchemaMismatchException` 异常：当使用 `StreamUploadSession` 时，如果用户上传的 Record 结构与表结构不匹配，将抛出该异常。此异常将额外携带最新的 schema version，方便用户重建 Session 并进行重试操作。
+- 在 `StreamUploadSession.Builder` 中新增 `allowSchemaMismatch` 方法，用于指定是否容忍用户上传的 Record 结构与表结构不匹配时是否抛出异常。默认值为 `true`。
+
+### 修复
+- 修复了在 Odps 中指定 `tunnelEndpoint` 时，使用 `StreamUploadSession` 无法生效的问题。
+- 修复了 `TunnelRetryHandler` 潜在的 NPE 问题。
+
+
 ## [0.50.0-rc1] - 2024-09-19
 ### 功能
 - **SQLExecutor** 新增 `isUseInstanceTunnel` 方法：

odps-examples/tunnel-examples/pom.xml

@@ -11,12 +11,12 @@
         <dependency>
             <groupId>com.aliyun.odps</groupId>
             <artifactId>odps-sdk-commons</artifactId>
-            <version>0.48.8-public</version>
+            <version>0.50.0-public</version>
         </dependency>
         <dependency>
             <groupId>com.aliyun.odps</groupId>
             <artifactId>odps-sdk-core</artifactId>
-            <version>0.48.8-public</version>
+            <version>0.50.0-public</version>
         </dependency>
     </dependencies>
 </project>

odps-sdk/odps-sdk-core/src/main/java/com/aliyun/odps/exceptions/SchemaMismatchException.java

@@ -0,0 +1,20 @@
+package com.aliyun.odps.exceptions;
+
+import java.io.IOException;
+
+/**
+ * @author dingxin ([email protected])
+ */
+public class SchemaMismatchException extends IOException {
+
+  private final String latestSchemaVersion;
+
+  public SchemaMismatchException(String msg, String latestSchemaVersion) {
+    super(msg);
+    this.latestSchemaVersion = latestSchemaVersion;
+  }
+
+  public String getLatestSchemaVersion() {
+    return latestSchemaVersion;
+  }
+}

odps-sdk/odps-sdk-core/src/main/java/com/aliyun/odps/tunnel/Configuration.java

@@ -50,6 +50,9 @@ public class Configuration extends GeneralConfiguration {
 
   public Configuration(Odps odps) {
     super(odps);
+    if (!StringUtils.isNullOrEmpty(odps.getTunnelEndpoint())) {
+      endpoint = URI.create(odps.getTunnelEndpoint());
+    }
   }
 
   public Configuration(Builder builder) {

odps-sdk/odps-sdk-core/src/main/java/com/aliyun/odps/tunnel/HttpHeaders.java

@@ -34,5 +34,6 @@ public class HttpHeaders extends Headers {
   public static final String HEADER_ODPS_SLOT_NUM = "odps-tunnel-slot-num";
   public static final String HEADER_ODPS_TUNNEL_TAGS= "odps-tunnel-tags";
   public static final String HEADER_ODPS_TUNNEL_SDK_SUPPORT_SCHEMA_EVOLUTION = "odps-tunnel-sdk-support-schema-evolution";
+  public static final String HEADER_ODPS_TUNNEL_LATEST_SCHEMA_VERSION = "odps-tunnel-latest-schema-version";
 
 }

odps-sdk/odps-sdk-core/src/main/java/com/aliyun/odps/tunnel/TableTunnel.java

@@ -1156,6 +1156,7 @@ abstract class Builder {
       private long slotNum = 0;
       private boolean createPartition = false;
       private String schemaVersion;
+      protected boolean allowSchemaMismatch = true;
       public String getSchemaName() {
         return schemaName;
       }
@@ -1206,6 +1207,11 @@ public Builder setCreatePartition(boolean createPartition) {
         return this;
       }
 
+      public Builder allowSchemaMismatch(boolean allowSchemaMismatch) {
+        this.allowSchemaMismatch = allowSchemaMismatch;
+        return this;
+      }
+
       abstract public StreamUploadSession build() throws TunnelException;
     }
   }

odps-sdk/odps-sdk-core/src/main/java/com/aliyun/odps/tunnel/TunnelConstants.java

@@ -85,6 +85,7 @@ public interface TunnelConstants {
   public static String PARAM_QUOTA_NAME= "quotaName";
   public static String PARAM_SHARING_QUOTA_TOKEN= "sharingQuotaToken";
   public static String PARAM_BLOCK_VERSION = "block_version";
+  public static String PARAM_CHECK_LATEST_SCHEMA = "check_latest_schema";
   public static String ENABLE_PARTIAL_UPDATE = "enable_partial_update";
   public static String SCHEMA_VERSION = "schema_version";
   public static String PARAM_DISABLE_MODIFIED_CHECK = "disable_modified_check";

odps-sdk/odps-sdk-core/src/main/java/com/aliyun/odps/tunnel/impl/StreamUploadSessionImpl.java

@@ -20,6 +20,7 @@
 import com.aliyun.odps.commons.transport.Response;
 import com.aliyun.odps.data.ArrayRecord;
 import com.aliyun.odps.data.Record;
+import com.aliyun.odps.exceptions.SchemaMismatchException;
 import com.aliyun.odps.tunnel.Configuration;
 import com.aliyun.odps.tunnel.HttpHeaders;
 import com.aliyun.odps.tunnel.TableTunnel;
@@ -98,13 +99,15 @@ public TableTunnel.StreamUploadSession build() throws TunnelException {
                     isCreatePartition(),
                     getSlotNum(),
                     zorderColumns,
-                    getSchemaVersion());
+                    getSchemaVersion(),
+                    allowSchemaMismatch);
         }
     }
 
     protected StreamUploadSessionImpl.Slots slots;
     private boolean p2pMode = false;
     private List<Column> columns;
+    private boolean checkLatestSchema;
 
     public StreamUploadSessionImpl(Configuration conf,
                                    String projectName,
@@ -114,7 +117,8 @@ public StreamUploadSessionImpl(Configuration conf,
                                    boolean cretaPartition,
                                    long slotNum,
                                    List<Column> zorderColumns,
-                                   String schemaVersion) throws TunnelException {
+                                   String schemaVersion,
+                                   boolean allowSchemaMismatch) throws TunnelException {
         this.config = conf;
         this.projectName = projectName;
         this.schemaName = schemaName;
@@ -123,6 +127,7 @@ public StreamUploadSessionImpl(Configuration conf,
         this.columns = zorderColumns;
         this.httpClient = Util.newRestClient(conf, projectName);
         this.schemaVersion = schemaVersion;
+        this.checkLatestSchema = !allowSchemaMismatch;
         initiate(slotNum, cretaPartition);
     }
 
@@ -275,6 +280,7 @@ private Connection getConnection(CompressOption compress, Slot slot, long size,
         if (!StringUtils.isNullOrEmpty(config.getQuotaName())) {
             params.put(TunnelConstants.PARAM_QUOTA_NAME, config.getQuotaName());
         }
+        params.put(TunnelConstants.PARAM_CHECK_LATEST_SCHEMA, String.valueOf(checkLatestSchema));
 
         switch (compress.algorithm) {
             case ODPS_RAW: {
@@ -352,7 +358,7 @@ public String writeBlock(ProtobufRecordPack pack, long timeout)
                     }
                 }
             });
-        } catch (RuntimeException re) {
+        } catch (RuntimeException | IOException re) {
             throw re;
         } catch (Exception e) {
             throw new IOException(e.getMessage(), e);
@@ -385,9 +391,15 @@ private String sendBlock(ProtobufRecordPack pack, Connection conn, Slot slot, lo
             }
         }
         if (!response.isOK()) {
-            TunnelException exception =
-                    new TunnelException(response.getHeader(HEADER_ODPS_REQUEST_ID), conn.getInputStream(),
-                            response.getStatus());
+            TunnelException exception = new TunnelException(response.getHeader(HEADER_ODPS_REQUEST_ID),
+                                                      conn.getInputStream(),
+                                                      response.getStatus());
+            if (exception.getErrorCode().equals("SchemaModified") &&
+                exception.getStatus().equals(412)) {
+                throw new SchemaMismatchException("SchemaModified",
+                                                  response.getHeader(
+                                                      HttpHeaders.HEADER_ODPS_TUNNEL_LATEST_SCHEMA_VERSION));
+            }
             throw new IOException(exception.getMessage(), exception);
         }
 

odps-sdk/odps-sdk-core/src/main/java/com/aliyun/odps/tunnel/io/TunnelRetryHandler.java

@@ -113,7 +113,9 @@ public <T> T executeWithRetry(Callable<T> action, IntConsumer errorCodeHandler)
                 if (!policy.shouldRetry(e, attempt)) {
                     throw e;
                 }
-                retryLogger.onRetryLog(e, attempt, policy.getRetryWaitTime(attempt));
+                if (retryLogger != null) {
+                    retryLogger.onRetryLog(e, attempt, policy.getRetryWaitTime(attempt));
+                }
                 try {
                     policy.waitForNextRetry(attempt);
                     if (errorCodeHandler != null && e instanceof TunnelException) {