polish

apache · Nov 3, 2022 · bb3b491 · bb3b491
1 parent 21e95fd
commit bb3b491
Show file tree

Hide file tree

Showing 3 changed files with 10 additions and 10 deletions.
diff --git a/src/main/java/org/apache/rocketmq/flink/sink2/committer/RocketMQCommittable.java b/src/main/java/org/apache/rocketmq/flink/sink2/committer/RocketMQCommittable.java
@@ -23,19 +23,19 @@ public class RocketMQCommittable {
 
     private final String producerGroup;
 
-    private final String transactionId;
+    private final String msgId;
 
     public String getProducerGroup() {
         return producerGroup;
     }
 
-    public String getTransactionId() {
-        return transactionId;
+    public String getMsgId() {
+        return msgId;
     }
 
-    public RocketMQCommittable(String producerGroup, String transactionId) {
+    public RocketMQCommittable(String producerGroup, String msgId) {
         this.producerGroup = producerGroup;
-        this.transactionId = transactionId;
+        this.msgId = msgId;
     }
 
     @Override
@@ -45,7 +45,7 @@ public String toString() {
                 + producerGroup
                 + '\''
                 + ", transactionId='"
-                + transactionId
+                + msgId
                 + '\''
                 + '}';
     }
@@ -60,11 +60,11 @@ public boolean equals(Object o) {
         }
         RocketMQCommittable that = (RocketMQCommittable) o;
         return Objects.equals(producerGroup, that.producerGroup)
-                && Objects.equals(transactionId, that.transactionId);
+                && Objects.equals(msgId, that.msgId);
     }
 
     @Override
     public int hashCode() {
-        return Objects.hash(producerGroup, transactionId);
+        return Objects.hash(producerGroup, msgId);
     }
 }
diff --git a/src/main/java/org/apache/rocketmq/flink/sink2/committer/RocketMQCommittableSerializer.java b/src/main/java/org/apache/rocketmq/flink/sink2/committer/RocketMQCommittableSerializer.java
@@ -41,7 +41,7 @@ public byte[] serialize(RocketMQCommittable state) throws IOException {
         try (final ByteArrayOutputStream baos = new ByteArrayOutputStream();
                 final DataOutputStream out = new DataOutputStream(baos)) {
             out.writeUTF(state.getProducerGroup());
-            out.writeUTF(state.getTransactionId());
+            out.writeUTF(state.getMsgId());
             out.flush();
             return baos.toByteArray();
         }

diff --git a/src/main/java/org/apache/rocketmq/flink/sink2/committer/RocketMQCommitter.java b/src/main/java/org/apache/rocketmq/flink/sink2/committer/RocketMQCommitter.java
@@ -52,7 +52,7 @@ public void commit(Collection<CommitRequest<RocketMQCommittable>> requests)
             throws IOException, InterruptedException {
         for (CommitRequest<RocketMQCommittable> request : requests) {
             final RocketMQCommittable committable = request.getCommittable();
-            final String transactionalId = committable.getTransactionId();
+            final String transactionalId = committable.getMsgId();
             LOG.debug("Committing RocketMQ transaction {}", transactionalId);
             needToCommitMsgTransactionIds.addLast(transactionalId);
         }