[ISSUE #4596]Fix SourceWorker#convertRecordToEvent method converts Co…

…nnectRecord to CloudEvent throw NPE (#4597) * [ISSUE #4596]Fix SourceWorker#convertRecordToEvent method converts ConnectRecord to CloudEvent throw NPE * fix CloudEventUtil#convertRecordToEvent method converts ConnectRecord to CloudEvent throw NPE
apache · Dec 3, 2023 · 2a46b82 · 2a46b82
1 parent d463fa7
commit 2a46b82
Show file tree

Hide file tree

Showing 2 changed files with 18 additions and 17 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);
             }
         }

diff --git a/...-openconnect-java/src/main/java/org/apache/eventmesh/openconnect/util/CloudEventUtil.java b/...-openconnect-java/src/main/java/org/apache/eventmesh/openconnect/util/CloudEventUtil.java
@@ -23,6 +23,7 @@
 import java.net.URISyntaxException;
 import java.time.OffsetDateTime;
 import java.util.Objects;
+import java.util.Optional;
 
 import io.cloudevents.CloudEvent;
 import io.cloudevents.core.builder.CloudEventBuilder;
@@ -33,32 +34,31 @@
 public class CloudEventUtil {
 
     public static CloudEvent convertRecordToEvent(ConnectRecord connectRecord) {
-        CloudEventBuilder cloudEventBuilder = CloudEventBuilder.v1()
-            .withData((byte[]) connectRecord.getData());
-        connectRecord.getExtensions().keySet().forEach(s -> {
-            switch (s) {
+        final CloudEventBuilder cloudEventBuilder = CloudEventBuilder.v1().withData((byte[]) connectRecord.getData());
+        Optional.ofNullable(connectRecord.getExtensions()).ifPresent((extensions) -> extensions.keySet().forEach(key -> {
+            switch (key) {
                 case "id":
-                    cloudEventBuilder.withId(connectRecord.getExtension(s));
+                    cloudEventBuilder.withId(connectRecord.getExtension(key));
                     break;
                 case "topic":
-                    cloudEventBuilder.withSubject(connectRecord.getExtension(s));
+                    cloudEventBuilder.withSubject(connectRecord.getExtension(key));
                     break;
                 case "source":
                     try {
-                        cloudEventBuilder.withSource(new URI(connectRecord.getExtension(s)));
+                        cloudEventBuilder.withSource(new URI(connectRecord.getExtension(key)));
                     } catch (URISyntaxException e) {
                         throw new RuntimeException(e);
                     }
                     break;
                 case "type":
-                    cloudEventBuilder.withType(connectRecord.getExtension(s));
+                    cloudEventBuilder.withType(connectRecord.getExtension(key));
                     break;
                 default:
-                    if (validateExtensionType(connectRecord.getExtensionObj(s))) {
-                        cloudEventBuilder.withExtension(s, connectRecord.getExtension(s));
+                    if (validateExtensionType(connectRecord.getExtensionObj(key))) {
+                        cloudEventBuilder.withExtension(key, connectRecord.getExtension(key));
                     }
             }
-        });
+        }));
         return cloudEventBuilder.build();
     }