commit 916cbe30170c25e085b23f27af0e7ceef08eb364
author Git_Yang <30995057+Git-Yang@users.noreply.github.com> Mon Mar 22 14:38:36 2021 +0800
committer GitHub <noreply@github.com> Mon Mar 22 14:38:36 2021 +0800
tree 8b2ad645a469e9bf732a092a83fbe5f01825f9e6
parent 948c5ee2a5ee9e7f19da30e58c70ba0ecee4a5c7

[Replicator] Fix message duplication problem (#692)

Signed-off-by: zhangyang <Git_Yang@163.com>

rocketmq-replicator/src/main/java/org/apache/rocketmq/replicator/RmqSourceTask.java

diff --git a/rocketmq-replicator/src/main/java/org/apache/rocketmq/replicator/RmqSourceTask.java b/rocketmq-replicator/src/main/java/org/apache/rocketmq/replicator/RmqSourceTask.java
index 3e8d78b..da7013a 100644
--- a/rocketmq-replicator/src/main/java/org/apache/rocketmq/replicator/RmqSourceTask.java
+++ b/rocketmq-replicator/src/main/java/org/apache/rocketmq/replicator/RmqSourceTask.java
@@ -160,10 +160,10 @@
                             schema.getFields().add(new Field(0,
                                 FieldName.COMMON_MESSAGE.getKey(), FieldType.STRING));
 
-                            DataEntryBuilder dataEntryBuilder = new DataEntryBuilder(schema);
-                            dataEntryBuilder.timestamp(System.currentTimeMillis())
-                                .queue(this.config.getStoreTopic()).entryType(EntryType.CREATE);
                             for (MessageExt msg : msgs) {
+                                DataEntryBuilder dataEntryBuilder = new DataEntryBuilder(schema);
+                                dataEntryBuilder.timestamp(System.currentTimeMillis())
+                                        .queue(this.config.getStoreTopic()).entryType(EntryType.CREATE);
                                 dataEntryBuilder.putFiled(FieldName.COMMON_MESSAGE.getKey(), new String(msg.getBody()));
                                 SourceDataEntry sourceDataEntry = dataEntryBuilder.buildSourceDataEntry(
                                     ByteBuffer.wrap(RmqConstants.getPartition(