Merge pull request #106 from salesforce/upgradekafka2

Upgrade kafka version to 3.3.2
salesforce · Oct 23, 2023 · 95e9a22 · 95e9a22
2 parents 6f5e32c + 0bcf7f1
commit 95e9a22
Show file tree

Hide file tree

Showing 3 changed files with 23 additions and 13 deletions.
diff --git a/pom.xml b/pom.xml
@@ -4,7 +4,7 @@
 
     <groupId>com.salesforce.mirus</groupId>
     <artifactId>mirus</artifactId>
-    <version>0.6.10-SNAPSHOT</version>
+    <version>0.6.11-SNAPSHOT</version>
 
     <name>Mirus</name>
     <description>Apache Kafka data replication tool based on Kafka Connect</description>
@@ -13,8 +13,9 @@
         <dist.dir>${project.build.directory}/dist</dist.dir>
         <jackson.version>2.14.0</jackson.version>
         <java.numeric.version>11</java.numeric.version>
-        <kafka.version>2.6.2</kafka.version>
+        <kafka.version>3.3.2</kafka.version>
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+        <slf4j.version>2.0.9</slf4j.version>
     </properties>
 
     <dependencies>
@@ -57,6 +58,16 @@
                 </exclusion>
             </exclusions>
         </dependency>
+        <dependency>
+            <groupId>org.eclipse.jetty</groupId>
+            <artifactId>jetty-io</artifactId>
+            <version>11.0.17</version>
+        </dependency>
+        <dependency>
+            <groupId>org.slf4j</groupId>
+            <artifactId>slf4j-reload4j</artifactId>
+            <version>${slf4j.version}</version>
+        </dependency>
         <dependency>
             <groupId>junit</groupId>
             <artifactId>junit</artifactId>
@@ -285,7 +296,7 @@
                         <configuration>
                             <target>
                                 <chmod dir="${dist.dir}" perm="700">
-                                    <include name="**/*.sh"/>
+                                    <include name="**/*.sh" />
                                 </chmod>
                             </target>
                         </configuration>
@@ -385,7 +396,7 @@
                                                 </goals>
                                             </pluginExecutionFilter>
                                             <action>
-                                                <ignore/>
+                                                <ignore />
                                             </action>
                                         </pluginExecution>
                                         <pluginExecution>
@@ -398,7 +409,7 @@
                                                 </goals>
                                             </pluginExecutionFilter>
                                             <action>
-                                                <ignore/>
+                                                <ignore />
                                             </action>
                                         </pluginExecution>
                                     </pluginExecutions>
@@ -452,10 +463,10 @@
                                 <phase>package</phase>
                                 <configuration>
                                     <target>
-                                        <copy file="${project.build.directory}/${project.artifactId}-${project.version}-shaded.jar" tofile="${dist.dir}/mirus.jar"/>
+                                        <copy file="${project.build.directory}/${project.artifactId}-${project.version}-shaded.jar" tofile="${dist.dir}/mirus.jar" />
                                         <!-- Grant execute permission to all shell scripts in package dir by setting it to 755 -->
                                         <chmod dir="${dist.dir}" perm="755">
-                                            <include name="**/*.sh"/>
+                                            <include name="**/*.sh" />
                                         </chmod>
                                     </target>
                                 </configuration>
@@ -484,5 +495,4 @@
             </build>
         </profile>
     </profiles>
-
 </project>
diff --git a/src/main/java/com/salesforce/mirus/metrics/ConnectorJmxReporter.java b/src/main/java/com/salesforce/mirus/metrics/ConnectorJmxReporter.java
@@ -17,7 +17,7 @@
 import java.util.stream.Collectors;
 import org.apache.kafka.common.MetricName;
 import org.apache.kafka.common.metrics.Metrics;
-import org.apache.kafka.common.metrics.stats.Total;
+import org.apache.kafka.common.metrics.stats.CumulativeSum;
 import org.apache.kafka.common.metrics.stats.Value;
 import org.apache.kafka.common.utils.Time;
 import org.apache.kafka.connect.runtime.Herder;
@@ -165,13 +165,13 @@ private void ensureMetricsCreated(String connectorName) {
     if (!metrics.metrics().containsKey(totalAttemptsPerConnectorMetric)) {
       metrics
           .sensor(FAILED_TASK_ATTEMPTS_METRIC_NAME + connectorName)
-          .add(totalAttemptsPerConnectorMetric, new Total());
+          .add(totalAttemptsPerConnectorMetric, new CumulativeSum());
     }
 
     if (!metrics.metrics().containsKey(restartAttemptsPerConnectorMetric)) {
       metrics
           .sensor(FAILED_CONNECTOR_ATTEMPTS_METRIC_NAME + connectorName)
-          .add(restartAttemptsPerConnectorMetric, new Total());
+          .add(restartAttemptsPerConnectorMetric, new CumulativeSum());
     }
   }
 

diff --git a/src/main/java/com/salesforce/mirus/metrics/TaskJmxReporter.java b/src/main/java/com/salesforce/mirus/metrics/TaskJmxReporter.java
@@ -15,7 +15,7 @@
 import org.apache.kafka.common.MetricName;
 import org.apache.kafka.common.metrics.Metrics;
 import org.apache.kafka.common.metrics.Sensor;
-import org.apache.kafka.common.metrics.stats.Total;
+import org.apache.kafka.common.metrics.stats.CumulativeSum;
 import org.apache.kafka.common.utils.Time;
 import org.apache.kafka.connect.runtime.TaskStatus;
 import org.apache.kafka.connect.runtime.rest.entities.ConnectorStateInfo.TaskState;
@@ -64,7 +64,7 @@ private void ensureMetricsCreated(ConnectorTaskId taskId) {
 
     if (!metrics.metrics().containsKey(taskMetric)) {
       Sensor sensor = getSensor(taskId.toString());
-      sensor.add(taskMetric, new Total());
+      sensor.add(taskMetric, new CumulativeSum());
       logger.info("Added the task {} to the list of JMX metrics", taskId);
       logger.debug("Updated set of JMX metrics is {}", metrics.metrics());
     }