[fix](fe) fix Azure resource persistence

fe/fe-core/src/main/java/org/apache/doris/catalog/Resource.java

@@ -33,6 +33,9 @@
 import com.google.common.base.Strings;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.Maps;
+import com.google.gson.JsonElement;
+import com.google.gson.JsonObject;
+import com.google.gson.JsonParser;
 import com.google.gson.annotations.SerializedName;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
@@ -266,9 +269,60 @@ public void write(DataOutput out) throws IOException {
 
     public static Resource read(DataInput in) throws IOException {
         String json = Text.readString(in);
+        json = addLegacyClazzIfMissing(json);
         return GsonUtils.GSON.fromJson(json, Resource.class);
     }
 
+    private static String addLegacyClazzIfMissing(String json) {
+        JsonElement jsonElement = JsonParser.parseString(json);
+        if (!jsonElement.isJsonObject()) {
+            return json;
+        }
+
+        JsonObject jsonObject = jsonElement.getAsJsonObject();
+        if (jsonObject.has("clazz") || !jsonObject.has("type")) {
+            return json;
+        }
+
+        JsonElement typeElement = jsonObject.get("type");
+        if (!typeElement.isJsonPrimitive()) {
+            return json;
+        }
+
+        ResourceType resourceType = ResourceType.fromString(typeElement.getAsString());
+        String clazz = getLegacyClazz(resourceType);
+        if (clazz == null) {
+            return json;
+        }
+        jsonObject.addProperty("clazz", clazz);
+        return jsonObject.toString();
+    }
+
+    private static String getLegacyClazz(ResourceType resourceType) {
+        switch (resourceType) {
+            case SPARK:
+                return SparkResource.class.getSimpleName();
+            case ODBC_CATALOG:
+                return OdbcCatalogResource.class.getSimpleName();
+            case S3:
+                return S3Resource.class.getSimpleName();
+            case JDBC:
+                return JdbcResource.class.getSimpleName();
+            case HDFS:
+                return HdfsResource.class.getSimpleName();
+            case HMS:
+                return HMSResource.class.getSimpleName();
+            case ES:
+                return EsResource.class.getSimpleName();
+            case AZURE:
+                return AzureResource.class.getSimpleName();
+            case AI:
+                return AIResource.class.getSimpleName();
+            default:
+                return null;
+        }
+    }
+
     @Override
     public void gsonPostProcess() throws IOException {
         // Resource is loaded from meta with older version

fe/fe-core/src/main/java/org/apache/doris/persist/gson/GsonUtils.java

@@ -75,6 +75,7 @@
 import org.apache.doris.catalog.AnyStructType;
 import org.apache.doris.catalog.AnyType;
 import org.apache.doris.catalog.ArrayType;
+import org.apache.doris.catalog.AzureResource;
 import org.apache.doris.catalog.BrokerTable;
 import org.apache.doris.catalog.DatabaseIf;
 import org.apache.doris.catalog.DistributionInfo;
@@ -335,6 +336,7 @@ public class GsonUtils {
             .registerSubtype(SparkResource.class, SparkResource.class.getSimpleName())
             .registerSubtype(OdbcCatalogResource.class, OdbcCatalogResource.class.getSimpleName())
             .registerSubtype(S3Resource.class, S3Resource.class.getSimpleName())
+            .registerSubtype(AzureResource.class, AzureResource.class.getSimpleName())
             .registerSubtype(JdbcResource.class, JdbcResource.class.getSimpleName())
             .registerSubtype(HdfsResource.class, HdfsResource.class.getSimpleName())
             .registerSubtype(HMSResource.class, HMSResource.class.getSimpleName())

fe/fe-core/src/test/java/org/apache/doris/persist/ResourcePersistTest.java

@@ -17,41 +17,79 @@
 
 package org.apache.doris.persist;
 
+import org.apache.doris.catalog.AzureResource;
 import org.apache.doris.catalog.Resource;
 import org.apache.doris.catalog.S3Resource;
+import org.apache.doris.common.io.Text;
 
 import org.junit.Assert;
 import org.junit.Test;
 
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
 import java.io.DataInputStream;
 import java.io.DataOutputStream;
-import java.io.File;
-import java.io.FileInputStream;
-import java.io.FileOutputStream;
 import java.io.IOException;
 
 public class ResourcePersistTest {
     @Test
     public void test() throws IOException {
         Resource resource = new S3Resource("s3_resource");
-        File file = new File("./ResourcePersistTest");
-        try {
-            // 1. Write objects to file
-            file.createNewFile();
-            DataOutputStream dos = new DataOutputStream(new FileOutputStream(file));
-            resource.write(dos);
-            dos.flush();
-            dos.close();
-
-            // 2. Read objects from file
-            DataInputStream dis = new DataInputStream(new FileInputStream(file));
-            S3Resource resource1 = (S3Resource) Resource.read(dis);
-            dis.close();
-            Assert.assertEquals(resource1.toString(), resource.toString());
-            resource1.readLock();
-            resource1.readUnlock();
-        } finally {
-            file.delete();
-        }
+        S3Resource resource1 = (S3Resource) readWrittenResource(resource);
+        Assert.assertEquals(resource1.toString(), resource.toString());
+        resource1.readLock();
+        resource1.readUnlock();
+    }
+
+    @Test
+    public void testAzureResourcePersist() throws IOException {
+        Resource resource = new AzureResource("azure_resource");
+        Assert.assertTrue(resource.toString().contains("\"clazz\":\"AzureResource\""));
+
+        Resource readResource = readWrittenResource(resource);
+        Assert.assertTrue(readResource instanceof AzureResource);
+        Assert.assertEquals("azure_resource", readResource.getName());
+        Assert.assertEquals(Resource.ResourceType.AZURE, readResource.getType());
+        readResource.readLock();
+        readResource.readUnlock();
+    }
+
+    @Test
+    public void testReadLegacyAzureResourceWithoutClazz() throws IOException {
+        String json = "{\"name\":\"legacy_azure_resource\",\"type\":\"AZURE\","
+                + "\"references\":{},\"id\":123,\"version\":0}";
+
+        Resource readResource = readResourceFromJson(json);
+        Assert.assertTrue(readResource instanceof AzureResource);
+        Assert.assertEquals("legacy_azure_resource", readResource.getName());
+        Assert.assertEquals(Resource.ResourceType.AZURE, readResource.getType());
+        readResource.readLock();
+        readResource.readUnlock();
+    }
+
+    private Resource readWrittenResource(Resource resource) throws IOException {
+        ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
+        DataOutputStream dos = new DataOutputStream(byteArrayOutputStream);
+        resource.write(dos);
+        dos.flush();
+        dos.close();
+
+        DataInputStream dis = new DataInputStream(new ByteArrayInputStream(byteArrayOutputStream.toByteArray()));
+        Resource readResource = Resource.read(dis);
+        dis.close();
+        return readResource;
+    }
+
+    private Resource readResourceFromJson(String json) throws IOException {
+        ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
+        DataOutputStream dos = new DataOutputStream(byteArrayOutputStream);
+        Text.writeString(dos, json);
+        dos.flush();
+        dos.close();
+
+        DataInputStream dis = new DataInputStream(new ByteArrayInputStream(byteArrayOutputStream.toByteArray()));
+        Resource readResource = Resource.read(dis);
+        dis.close();
+        return readResource;
     }
 }