[jindo] Unify the Jindo User-Agent format

paimon-filesystems/paimon-jindo/pom.xml

@@ -163,6 +163,19 @@
 
     <build>
         <plugins>
+            <plugin>
+                <groupId>org.codehaus.mojo</groupId>
+                <artifactId>templating-maven-plugin</artifactId>
+                <executions>
+                    <execution>
+                        <id>filter-sources</id>
+                        <goals>
+                            <goal>filter-sources</goal>
+                        </goals>
+                    </execution>
+                </executions>
+            </plugin>
+
             <plugin>
                 <groupId>org.apache.maven.plugins</groupId>
                 <artifactId>maven-jar-plugin</artifactId>

paimon-filesystems/paimon-jindo/src/main/java-templates/org/apache/paimon/jindo/JindoBuildVersions.java

@@ -0,0 +1,27 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.jindo;
+
+/** Versions written in at build time, so reading them needs no class loader lookup. */
+final class JindoBuildVersions {
+
+    static final String PAIMON = "${project.version}";
+
+    private JindoBuildVersions() {}
+}

paimon-filesystems/paimon-jindo/src/main/java/org/apache/paimon/jindo/JindoFileIO.java

@@ -29,7 +29,6 @@
 import org.apache.paimon.utils.IOUtils;
 import org.apache.paimon.utils.Pair;
 import org.apache.paimon.utils.SensitiveConfigUtils;
-import org.apache.paimon.utils.StringUtils;
 
 import com.aliyun.jindodata.common.JindoHadoopSystem;
 import com.aliyun.jindodata.dls.JindoDlsFileSystem;
@@ -72,10 +71,7 @@ public class JindoFileIO extends HadoopCompliantFileIO implements HadoopOptionsP
     private static final String OSS_ACCESS_KEY_ID = "fs.oss.accessKeyId";
     private static final String OSS_ACCESS_KEY_SECRET = "fs.oss.accessKeySecret";
     private static final String OSS_SECURITY_TOKEN = "fs.oss.securityToken";
-    private static final String OSS_USER_AGENT_EXTENDED = "fs.oss.user.agent.extended";
     private static final String OSS_SHOW_DIR_TIMESTAMP = "fs.oss.show-dir-timestamp";
-    private static final String DLF_ACCESS_TRACKING_EXTENDED_INFO =
-            "dlf.access-tracking.extended-info";
 
     private static final Map<String, String> CASE_SENSITIVE_KEYS =
             new HashMap<String, String>() {
@@ -154,17 +150,7 @@ public void configure(CatalogContext context) {
             hadoopOptions.set(OSS_SHOW_DIR_TIMESTAMP, "false");
         }
 
-        String dlfAccessTrackingExtendedInfo =
-                context.options().get(DLF_ACCESS_TRACKING_EXTENDED_INFO);
-        if (!StringUtils.isNullOrWhitespaceOnly(dlfAccessTrackingExtendedInfo)) {
-            LOG.info("Adding DLF access tracking extended info: {}", dlfAccessTrackingExtendedInfo);
-            String existedUserAgentExtended = hadoopOptions.get(OSS_USER_AGENT_EXTENDED);
-            hadoopOptions.set(
-                    OSS_USER_AGENT_EXTENDED,
-                    StringUtils.isNullOrWhitespaceOnly(existedUserAgentExtended)
-                            ? dlfAccessTrackingExtendedInfo
-                            : existedUserAgentExtended + " " + dlfAccessTrackingExtendedInfo);
-        }
+        JindoUserAgent.apply(context.options(), hadoopOptions);
 
         // another config when enable cache
         hadoopOptionsWithCache = new Options(hadoopOptions.toMap());

paimon-filesystems/paimon-jindo/src/main/java/org/apache/paimon/jindo/JindoUserAgent.java

@@ -0,0 +1,101 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.jindo;
+
+import org.apache.paimon.options.Options;
+import org.apache.paimon.utils.StringUtils;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+/**
+ * Fills Paimon's unified User-Agent {@code module(transport;features) extended} into the {@code
+ * fs.<scheme>.user.agent.*} keys, from which Jindo builds the header.
+ */
+final class JindoUserAgent {
+
+    static final String MODULE = "user-agent.module";
+    static final String FEATURES = "user-agent.features";
+    static final String EXTENDED = "user-agent.extended";
+    static final String DLF_ACCESS_TRACKING_EXTENDED_INFO = "dlf.access-tracking.extended-info";
+
+    static final String PAIMON = "Paimon/" + JindoBuildVersions.PAIMON;
+
+    private static final String[] SCHEMES = {"oss", "dls"};
+
+    private JindoUserAgent() {}
+
+    /**
+     * Per part, {@code fs.<scheme>.user.agent.*} wins over the catalog-wide {@code user-agent.*}.
+     */
+    static void apply(Options catalogOptions, Options hadoopOptions) {
+        for (String scheme : SCHEMES) {
+            String prefix = "fs." + scheme + ".user.agent.";
+            String module = first(hadoopOptions.get(prefix + "module"), catalogOptions.get(MODULE));
+            if (module != null) {
+                hadoopOptions.set(prefix + "module", module);
+            }
+            hadoopOptions.set(
+                    prefix + "features",
+                    features(
+                            first(
+                                    hadoopOptions.get(prefix + "features"),
+                                    catalogOptions.get(FEATURES))));
+            // Access tracking info is appended, so a user-set extended value is kept.
+            String extended =
+                    join(
+                            first(
+                                    hadoopOptions.get(prefix + "extended"),
+                                    catalogOptions.get(EXTENDED)),
+                            catalogOptions.get(DLF_ACCESS_TRACKING_EXTENDED_INFO));
+            if (!extended.isEmpty()) {
+                hadoopOptions.set(prefix + "extended", extended);
+            }
+        }
+    }
+
+    private static String features(String configured) {
+        List<String> features = new ArrayList<>();
+        if (configured != null) {
+            features.addAll(Arrays.asList(configured.split("\\s+")));
+        }
+        if (features.stream().noneMatch(f -> f.equals("Paimon") || f.startsWith("Paimon/"))) {
+            features.add(0, PAIMON);
+        }
+        return String.join(" ", features);
+    }
+
+    private static String first(String preferred, String fallback) {
+        if (!StringUtils.isNullOrWhitespaceOnly(preferred)) {
+            return preferred.trim();
+        }
+        return StringUtils.isNullOrWhitespaceOnly(fallback) ? null : fallback.trim();
+    }
+
+    private static String join(String first, String second) {
+        List<String> parts = new ArrayList<>();
+        for (String part : Arrays.asList(first, second)) {
+            if (!StringUtils.isNullOrWhitespaceOnly(part)) {
+                parts.add(part.trim());
+            }
+        }
+        return String.join(" ", parts);
+    }
+}

paimon-filesystems/paimon-jindo/src/test/java/org/apache/paimon/jindo/TestJindoUserAgent.java

@@ -0,0 +1,113 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.jindo;
+
+import org.apache.paimon.catalog.CatalogContext;
+import org.apache.paimon.options.Options;
+
+import org.junit.jupiter.api.Test;
+
+import java.lang.reflect.Field;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for {@link JindoUserAgent}. */
+public class TestJindoUserAgent {
+
+    private static final String[] SCHEMES = {"oss", "dls"};
+
+    @Test
+    public void testDefault() {
+        assertThat(JindoBuildVersions.PAIMON).matches("\\d+\\.\\d+\\S*");
+        Options hadoopOptions = configure(new Options());
+        for (String scheme : SCHEMES) {
+            assertThat(hadoopOptions.get(key(scheme, "features"))).isEqualTo(JindoUserAgent.PAIMON);
+            assertThat(hadoopOptions.get(key(scheme, "module"))).isNull();
+            assertThat(hadoopOptions.get(key(scheme, "extended"))).isNull();
+        }
+    }
+
+    @Test
+    public void testCatalogWideKeys() {
+        Options options = new Options();
+        options.set(JindoUserAgent.MODULE, "MyApp/1.0");
+        options.set(JindoUserAgent.FEATURES, " Flink  Spark ");
+        options.set(JindoUserAgent.EXTENDED, "vvr");
+        Options hadoopOptions = configure(options);
+        for (String scheme : SCHEMES) {
+            assertThat(hadoopOptions.get(key(scheme, "module"))).isEqualTo("MyApp/1.0");
+            assertThat(hadoopOptions.get(key(scheme, "features")))
+                    .isEqualTo(JindoUserAgent.PAIMON + " Flink Spark");
+            assertThat(hadoopOptions.get(key(scheme, "extended"))).isEqualTo("vvr");
+        }
+    }
+
+    @Test
+    public void testSchemeKeysOverrideCatalogWideKeys() {
+        Options options = new Options();
+        options.set(JindoUserAgent.FEATURES, "Flink");
+        options.set(JindoUserAgent.EXTENDED, "vvr");
+        options.set(key("oss", "features"), "morax/2.6.0");
+        options.set(key("oss", "extended"), "bennett/2.7.0");
+        Options hadoopOptions = configure(options);
+        assertThat(hadoopOptions.get(key("oss", "features")))
+                .isEqualTo(JindoUserAgent.PAIMON + " morax/2.6.0");
+        assertThat(hadoopOptions.get(key("oss", "extended"))).isEqualTo("bennett/2.7.0");
+        assertThat(hadoopOptions.get(key("dls", "features")))
+                .isEqualTo(JindoUserAgent.PAIMON + " Flink");
+        assertThat(hadoopOptions.get(key("dls", "extended"))).isEqualTo("vvr");
+    }
+
+    @Test
+    public void testAccessTrackingIsAppendedToCatalogWideExtended() {
+        Options options = new Options();
+        options.set(JindoUserAgent.EXTENDED, "vvr");
+        options.set(JindoUserAgent.DLF_ACCESS_TRACKING_EXTENDED_INFO, "uid/123 user/alice");
+        Options hadoopOptions = configure(options);
+        for (String scheme : SCHEMES) {
+            assertThat(hadoopOptions.get(key(scheme, "extended")))
+                    .isEqualTo("vvr uid/123 user/alice");
+        }
+    }
+
+    @Test
+    public void testPaimonFeatureIsNotDuplicated() {
+        Options options = new Options();
+        options.set(JindoUserAgent.FEATURES, "Flink Paimon");
+        assertThat(configure(options).get(key("oss", "features"))).isEqualTo("Flink Paimon");
+    }
+
+    private static String key(String scheme, String part) {
+        return "fs." + scheme + ".user.agent." + part;
+    }
+
+    private static Options configure(Options options) {
+        options.set("fs.oss.accessKeyId", "testAk");
+        options.set("fs.oss.accessKeySecret", "testSk");
+        JindoFileIO fileIO = new JindoFileIO();
+        fileIO.configure(CatalogContext.create(options));
+        try {
+            Field field = JindoFileIO.class.getDeclaredField("hadoopOptions");
+            field.setAccessible(true);
+            return (Options) field.get(fileIO);
+        } catch (ReflectiveOperationException e) {
+            throw new RuntimeException(e);
+        }
+    }
+}

pom.xml

@@ -980,6 +980,12 @@ under the License.
                     <version>1.7</version>
                 </plugin>
 
+                <plugin>
+                    <groupId>org.codehaus.mojo</groupId>
+                    <artifactId>templating-maven-plugin</artifactId>
+                    <version>3.0.0</version>
+                </plugin>
+
                 <plugin>
                     <groupId>org.apache.maven.plugins</groupId>
                     <artifactId>maven-compiler-plugin</artifactId>