[INLONG-9038][Sort] Fix failure of parsing InlongMessage when exporti…

…ng Pulsar
vernedeng · Oct 10, 2023 · 687bab3 · 687bab3
1 parent fafe05a
commit 687bab3
Showing 1 changed file with 3 additions and 7 deletions.
diff --git a/...e/src/main/java/org/apache/inlong/sort/standalone/sink/pulsar/PulsarFederationWorker.java b/...e/src/main/java/org/apache/inlong/sort/standalone/sink/pulsar/PulsarFederationWorker.java
@@ -131,14 +131,10 @@ public void run() {
      * 
      * @param currentRecord
      */
-    private String fillTopic(Event currentRecord) {
-        Map<String, String> headers = currentRecord.getHeaders();
-        String inlongGroupId = headers.get(Constants.INLONG_GROUP_ID);
-        String inlongStreamId = headers.get(Constants.INLONG_STREAM_ID);
-        String uid = InlongId.generateUid(inlongGroupId, inlongStreamId);
-        String topic = this.context.getTopic(uid);
+    private String fillTopic(ProfileEvent currentRecord) {
+        String topic = this.context.getTopic(currentRecord.getUid());
         if (!StringUtils.isBlank(topic)) {
-            headers.put(Constants.TOPIC, topic);
+            currentRecord.getHeaders().put(Constants.TOPIC, topic);
             return topic;
         }
         return "-";