[INLONG-11349][Sort] Integrate opentelemetry for sort-connectors-v1.15

inlong-sort/sort-end-to-end-tests/sort-end-to-end-tests-v1.15/src/test/java/org/apache/inlong/sort/tests/Kafka2StarRocksTest.java

@@ -20,6 +20,7 @@
 import org.apache.inlong.sort.tests.utils.FlinkContainerTestEnvJRE8;
 import org.apache.inlong.sort.tests.utils.JdbcProxy;
 import org.apache.inlong.sort.tests.utils.MySqlContainer;
+import org.apache.inlong.sort.tests.utils.OpenTelemetryContainer;
 import org.apache.inlong.sort.tests.utils.PlaceholderResolver;
 import org.apache.inlong.sort.tests.utils.StarRocksContainer;
 import org.apache.inlong.sort.tests.utils.TestUtils;
@@ -34,6 +35,7 @@
 import org.testcontainers.containers.KafkaContainer;
 import org.testcontainers.containers.output.Slf4jLogConsumer;
 import org.testcontainers.utility.DockerImageName;
+import org.testcontainers.utility.MountableFile;
 
 import java.io.IOException;
 import java.net.URI;
@@ -111,6 +113,15 @@ public class Kafka2StarRocksTest extends FlinkContainerTestEnvJRE8 {
                     .withNetworkAliases("mysql")
                     .withLogConsumer(new Slf4jLogConsumer(MYSQL_LOG));
 
+    @ClassRule
+    public static final OpenTelemetryContainer OPEN_TELEMETRY_CONTAINER =
+            (OpenTelemetryContainer) new OpenTelemetryContainer()
+                    .withCopyFileToContainer(MountableFile.forClasspathResource("/env/otel-config.yaml"),
+                            "/otel-config.yaml")
+                    .withCommand("--config=/otel-config.yaml")
+                    .withNetwork(NETWORK)
+                    .withNetworkAliases("logcollector");
+
     @Before
     public void setup() {
         waitUntilJobRunning(Duration.ofSeconds(30));
@@ -152,6 +163,10 @@ public static void teardown() {
         if (STAR_ROCKS != null) {
             STAR_ROCKS.stop();
         }
+
+        if (OPEN_TELEMETRY_CONTAINER != null) {
+            OPEN_TELEMETRY_CONTAINER.stop();
+        }
     }
 
     private void initializeKafkaTable(String topic) {
@@ -223,5 +238,10 @@ public void testKafkaWithSqlFile() throws Exception {
                 "test_output1",
                 3,
                 60000L);
+        // check log appender
+        String logs = OPEN_TELEMETRY_CONTAINER.getLogs();
+        if (!logs.contains("OpenTelemetryLogger installed")) {
+            throw new Exception("Failure to append logs to OpenTelemetry");
+        }
     }
 }

inlong-sort/sort-end-to-end-tests/sort-end-to-end-tests-v1.15/src/test/java/org/apache/inlong/sort/tests/Mysql2StarRocksTest.java

@@ -20,6 +20,7 @@
 import org.apache.inlong.sort.tests.utils.FlinkContainerTestEnvJRE8;
 import org.apache.inlong.sort.tests.utils.JdbcProxy;
 import org.apache.inlong.sort.tests.utils.MySqlContainer;
+import org.apache.inlong.sort.tests.utils.OpenTelemetryContainer;
 import org.apache.inlong.sort.tests.utils.StarRocksContainer;
 import org.apache.inlong.sort.tests.utils.TestUtils;
 
@@ -30,6 +31,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.testcontainers.containers.output.Slf4jLogConsumer;
+import org.testcontainers.utility.MountableFile;
 
 import java.net.URISyntaxException;
 import java.nio.file.Path;
@@ -85,6 +87,15 @@ public class Mysql2StarRocksTest extends FlinkContainerTestEnvJRE8 {
                     .withNetworkAliases("mysql")
                     .withLogConsumer(new Slf4jLogConsumer(LOG));
 
+    @ClassRule
+    public static final OpenTelemetryContainer OPEN_TELEMETRY_CONTAINER =
+            (OpenTelemetryContainer) new OpenTelemetryContainer()
+                    .withCopyFileToContainer(MountableFile.forClasspathResource("/env/otel-config.yaml"),
+                            "/otel-config.yaml")
+                    .withCommand("--config=/otel-config.yaml")
+                    .withNetwork(NETWORK)
+                    .withNetworkAliases("logcollector");
+
     @Before
     public void setup() {
         waitUntilJobRunning(Duration.ofSeconds(30));
@@ -121,6 +132,9 @@ public static void teardown() {
         if (STAR_ROCKS != null) {
             STAR_ROCKS.stop();
         }
+        if (OPEN_TELEMETRY_CONTAINER != null) {
+            OPEN_TELEMETRY_CONTAINER.stop();
+        }
     }
 
     /**
@@ -161,6 +175,11 @@ public void testMysqlUpdateAndDelete() throws Exception {
                 expectResult,
                 "test_output1",
                 3,
-                60000L);
+                80000L);
+        // check log appender
+        String logs = OPEN_TELEMETRY_CONTAINER.getLogs();
+        if (!logs.contains("OpenTelemetryLogger installed")) {
+            throw new Exception("Failure to append logs to OpenTelemetry");
+        }
     }
 }

inlong-sort/sort-end-to-end-tests/sort-end-to-end-tests-v1.15/src/test/java/org/apache/inlong/sort/tests/Pulsar2StarRocksTest.java

@@ -19,6 +19,7 @@
 
 import org.apache.inlong.sort.tests.utils.FlinkContainerTestEnvJRE8;
 import org.apache.inlong.sort.tests.utils.JdbcProxy;
+import org.apache.inlong.sort.tests.utils.OpenTelemetryContainer;
 import org.apache.inlong.sort.tests.utils.StarRocksContainer;
 import org.apache.inlong.sort.tests.utils.TestUtils;
 
@@ -35,6 +36,7 @@
 import org.testcontainers.containers.PulsarContainer;
 import org.testcontainers.containers.output.Slf4jLogConsumer;
 import org.testcontainers.utility.DockerImageName;
+import org.testcontainers.utility.MountableFile;
 
 import java.net.URI;
 import java.net.URISyntaxException;
@@ -91,6 +93,15 @@ public class Pulsar2StarRocksTest extends FlinkContainerTestEnvJRE8 {
                     .withNetworkAliases(INTER_CONTAINER_STAR_ROCKS_ALIAS)
                     .withLogConsumer(new Slf4jLogConsumer(STAR_ROCKS_LOG));
 
+    @ClassRule
+    public static final OpenTelemetryContainer OPEN_TELEMETRY_CONTAINER =
+            (OpenTelemetryContainer) new OpenTelemetryContainer()
+                    .withCopyFileToContainer(MountableFile.forClasspathResource("/env/otel-config.yaml"),
+                            "/otel-config.yaml")
+                    .withCommand("--config=/otel-config.yaml")
+                    .withNetwork(NETWORK)
+                    .withNetworkAliases("logcollector");
+
     @Before
     public void setup() {
         waitUntilJobRunning(Duration.ofSeconds(30));
@@ -119,6 +130,9 @@ public static void teardown() {
         if (STAR_ROCKS != null) {
             STAR_ROCKS.stop();
         }
+        if (OPEN_TELEMETRY_CONTAINER != null) {
+            OPEN_TELEMETRY_CONTAINER.stop();
+        }
     }
 
     @Test
@@ -144,6 +158,11 @@ public void testPulsarToStarRocks() throws Exception {
                 "1,Alice,Hello, Pulsar",
                 "2,Bob,Goodbye, Pulsar");
         proxy.checkResultWithTimeout(expectedResult, "test_output1", 3, 60000L);
+        // check log appender
+        String logs = OPEN_TELEMETRY_CONTAINER.getLogs();
+        if (!logs.contains("OpenTelemetryLogger installed")) {
+            throw new Exception("Failure to append logs to OpenTelemetry");
+        }
     }
 
 }

inlong-sort/sort-end-to-end-tests/sort-end-to-end-tests-v1.15/src/test/java/org/apache/inlong/sort/tests/utils/FlinkContainerTestEnvJRE8.java

@@ -45,6 +45,7 @@ public static void before() {
                         .withNetworkAliases(INTER_CONTAINER_TM_ALIAS)
                         .withExposedPorts(DEBUG_PORT)
                         .withEnv("FLINK_PROPERTIES", FLINK_PROPERTIES)
+                        .withEnv("OTEL_EXPORTER_ENDPOINT", "logcollector:4317")
                         .dependsOn(jobManager)
                         .withLogConsumer(new Slf4jLogConsumer(TM_LOG));
 

inlong-sort/sort-end-to-end-tests/sort-end-to-end-tests-v1.15/src/test/java/org/apache/inlong/sort/tests/utils/OpenTelemetryContainer.java

@@ -0,0 +1,30 @@
+/*
+ * 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.inlong.sort.tests.utils;
+
+import org.testcontainers.containers.GenericContainer;
+
+public class OpenTelemetryContainer extends GenericContainer {
+
+    public static final String IMAGE = "otel/opentelemetry-collector-contrib:0.110.0";
+    public static final Integer PORT = 4317;
+    public OpenTelemetryContainer() {
+        super(IMAGE);
+        addExposedPort(PORT);
+    }
+}

inlong-sort/sort-end-to-end-tests/sort-end-to-end-tests-v1.15/src/test/resources/env/otel-config.yaml

@@ -0,0 +1,35 @@
+#
+# 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.
+#
+
+receivers:
+  otlp:
+    protocols:
+      grpc:
+        endpoint: logcollector:4317
+processors:
+  batch:
+
+exporters:
+  debug:
+    verbosity: detailed
+
+service:
+  pipelines:
+    logs:
+      receivers: [otlp]
+      processors: [batch]
+      exporters: [debug]

inlong-sort/sort-end-to-end-tests/sort-end-to-end-tests-v1.15/src/test/resources/flinkSql/mysql_test.sql

@@ -10,7 +10,7 @@ CREATE TABLE test_input1 (
     'password' = 'inlong',
     'database-name' = 'test',
     'table-name' = 'test_input1',
-    'scan.incremental.snapshot.enabled' = 'false',
+    'scan.incremental.snapshot.enabled' = 'true',
     'jdbc.properties.useSSL' = 'false',
     'jdbc.properties.allowPublicKeyRetrieval' = 'true'
     );

inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/iceberg/pom.xml

@@ -151,6 +151,8 @@
                             <createDependencyReducedPom>false</createDependencyReducedPom>
                             <artifactSet>
                                 <includes>
+                                    <include>io.opentelemetry*</include>
+                                    <include>com.squareup.*</include>
                                     <include>org.apache.inlong:*</include>
                                     <include>com.google.protobuf:*</include>
                                     <include>org.apache.kafka:*</include>

inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/source/reader/IcebergSourceReader.java

@@ -17,6 +17,8 @@
 
 package org.apache.inlong.sort.iceberg.source.reader;
 
+import org.apache.inlong.sort.base.util.OpenTelemetryLogger;
+
 import org.apache.flink.annotation.Internal;
 import org.apache.flink.api.connector.source.SourceReaderContext;
 import org.apache.flink.connector.base.source.reader.SingleThreadMultiplexSourceReaderBase;
@@ -25,6 +27,7 @@
 import org.apache.iceberg.flink.source.split.IcebergSourceSplit;
 import org.apache.iceberg.flink.source.split.SplitRequestEvent;
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.logging.log4j.Level;
 
 import java.util.Collection;
 import java.util.Collections;
@@ -40,6 +43,7 @@ public class IcebergSourceReader<T>
             SingleThreadMultiplexSourceReaderBase<RecordAndPosition<T>, T, IcebergSourceSplit, IcebergSourceSplit> {
 
     private final InlongIcebergSourceReaderMetrics<T> metrics;
+    private final OpenTelemetryLogger openTelemetryLogger;
     public IcebergSourceReader(
             InlongIcebergSourceReaderMetrics<T> metrics,
             ReaderFunction<T> readerFunction,
@@ -50,17 +54,28 @@ public IcebergSourceReader(
                 context.getConfiguration(),
                 context);
         this.metrics = metrics;
+        this.openTelemetryLogger = new OpenTelemetryLogger.Builder()
+                .setLogLevel(Level.ERROR)
+                .setServiceName(this.getClass().getSimpleName())
+                .setLocalHostIp(this.context.getLocalHostName()).build();
     }
 
     @Override
     public void start() {
+        this.openTelemetryLogger.install();
         // We request a split only if we did not get splits during the checkpoint restore.
         // Otherwise, reader restarts will keep requesting more and more splits.
         if (getNumberOfCurrentlyAssignedSplits() == 0) {
             requestSplit(Collections.emptyList());
         }
     }
 
+    @Override
+    public void close() throws Exception {
+        super.close();
+        openTelemetryLogger.uninstall();
+    }
+
     @Override
     protected void onSplitFinished(Map<String, IcebergSourceSplit> finishedSplitIds) {
         requestSplit(Lists.newArrayList(finishedSplitIds.keySet()));

inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/kafka/pom.xml

@@ -76,6 +76,8 @@
                             <createDependencyReducedPom>false</createDependencyReducedPom>
                             <artifactSet>
                                 <includes>
+                                    <include>io.opentelemetry*</include>
+                                    <include>com.squareup.*</include>
                                     <include>org.apache.inlong:*</include>
                                     <include>org.apache.kafka:*</include>
                                     <include>org.apache.flink:flink-connector-kafka</include>

inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/kafka/src/main/java/org/apache/inlong/sort/kafka/source/reader/KafkaSourceReader.java

@@ -17,6 +17,7 @@
 
 package org.apache.inlong.sort.kafka.source.reader;
 
+import org.apache.inlong.sort.base.util.OpenTelemetryLogger;
 import org.apache.inlong.sort.kafka.table.DynamicKafkaDeserializationSchema;
 
 import org.apache.flink.annotation.Internal;
@@ -37,6 +38,7 @@
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.clients.consumer.OffsetAndMetadata;
 import org.apache.kafka.common.TopicPartition;
+import org.apache.logging.log4j.Level;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -66,6 +68,7 @@ public class KafkaSourceReader<T>
     private final KafkaSourceReaderMetrics kafkaSourceReaderMetrics;
     private final boolean commitOffsetsOnCheckpoint;
     private final KafkaDeserializationSchema<RowData> metricSchema;
+    private final OpenTelemetryLogger openTelemetryLogger;
 
     public KafkaSourceReader(
             FutureCompletingBlockingQueue<RecordsWithSplitIds<ConsumerRecord<byte[], byte[]>>> elementsQueue,
@@ -87,6 +90,22 @@ public KafkaSourceReader(
                     "Offset commit on checkpoint is disabled. "
                             + "Consuming offset will not be reported back to Kafka cluster.");
         }
+        this.openTelemetryLogger = new OpenTelemetryLogger.Builder()
+                .setLogLevel(Level.ERROR)
+                .setServiceName(this.getClass().getSimpleName())
+                .setLocalHostIp(this.context.getLocalHostName()).build();
+    }
+
+    @Override
+    public void start() {
+        this.openTelemetryLogger.install();
+        super.start();
+    }
+
+    @Override
+    public void close() throws Exception {
+        super.close();
+        openTelemetryLogger.uninstall();
     }
 
     @Override

inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/mysql-cdc/pom.xml

@@ -95,6 +95,8 @@
                             <createDependencyReducedPom>false</createDependencyReducedPom>
                             <artifactSet>
                                 <includes>
+                                    <include>io.opentelemetry*</include>
+                                    <include>com.squareup.*</include>
                                     <include>org.apache.inlong:*</include>
                                     <include>io.debezium:debezium-api</include>
                                     <include>io.debezium:debezium-embedded</include>