[ISSUE apache#4596]Fix SourceWorker#convertRecordToEvent method conve…

…rts ConnectRecord to CloudEvent throw NPE
mxsm · Dec 2, 2023 · f45438d · f45438d
1 parent 5998926
commit f45438d
Showing 1 changed file with 7 additions and 6 deletions.
diff --git a/...entmesh-openconnect-java/src/main/java/org/apache/eventmesh/openconnect/SourceWorker.java b/...entmesh-openconnect-java/src/main/java/org/apache/eventmesh/openconnect/SourceWorker.java
@@ -244,12 +244,13 @@ private CloudEvent convertRecordToEvent(ConnectRecord connectRecord) {
             .withData(Objects.requireNonNull(JsonUtils.toJSONString(connectRecord.getData())).getBytes(StandardCharsets.UTF_8))
             .withExtension("ttl", 10000);
 
-        for (String key : connectRecord.getExtensions().keySet()) {
-            if (CloudEventUtil.validateExtensionType(connectRecord.getExtensionObj(key))) {
-                cloudEventBuilder.withExtension(key, connectRecord.getExtension(key));
+        if (connectRecord.getExtensions() != null) {
+            for (String key : connectRecord.getExtensions().keySet()) {
+                if (CloudEventUtil.validateExtensionType(connectRecord.getExtensionObj(key))) {
+                    cloudEventBuilder.withExtension(key, connectRecord.getExtension(key));
+                }
             }
         }
-
         return cloudEventBuilder.build();
     }
 
@@ -329,15 +330,15 @@ public boolean commitOffsets() {
             log.info("{} Committing offsets for {} acknowledged messages", this, committableOffsets.numCommittableMessages());
             if (committableOffsets.hasPending()) {
                 log.debug("{} There are currently {} pending messages spread across {} source partitions whose offsets will not be committed. "
-                    + "The source partition with the most pending messages is {}, with {} pending messages",
+                        + "The source partition with the most pending messages is {}, with {} pending messages",
                     this,
                     committableOffsets.numUncommittableMessages(),
                     committableOffsets.numDeques(),
                     committableOffsets.largestDequePartition(),
                     committableOffsets.largestDequeSize());
             } else {
                 log.debug("{} There are currently no pending messages for this offset commit; "
-                    + "all messages dispatched to the task's producer since the last commit have been acknowledged",
+                        + "all messages dispatched to the task's producer since the last commit have been acknowledged",
                     this);
             }
         }