[INLONG-12071][Manager] Fix the incorrect filtering logic that checks whether messages match (#12072)

* [INLONG-12071][Manager] Fix the filtering logic that incorrectly determines whether messages match

* [INLONG-12071][Manager] Fix the compile warning issues

Author: healchow

inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/consume/BriefMQMessage.java

@@ -72,7 +72,7 @@ public static class FieldInfo {
 
         private String fieldName;
 
-        private String FieldValue;
+        private String fieldValue;
 
     }
 

inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/message/DeserializeOperator.java

@@ -70,27 +70,41 @@ default DeserializationConfig getDeserializationConfig(InlongStreamInfo streamIn
                 streamInfo.getWrapType()));
     }
 
-    default Boolean checkIfFilter(QueryMessageRequest request, List<FieldInfo> streamFieldList) {
-        if (StringUtils.isBlank(request.getFieldName()) || StringUtils.isBlank(request.getOperationType())
-                || StringUtils.isBlank(request.getTargetValue())) {
+    /**
+     * Check whether the message should be filtered by the filter condition
+     *
+     * @param request query message request
+     * @param streamFieldList stream field list
+     * @return true if the message should be filtered
+     */
+    default boolean checkIfFilter(QueryMessageRequest request, List<FieldInfo> streamFieldList) {
+        String fieldName = request.getFieldName();
+        if (StringUtils.isAnyBlank(fieldName, request.getOperationType())) {
             return false;
         }
-        boolean ifFilter = false;
+
         FieldInfo fieldInfo = streamFieldList.stream()
-                .filter(v -> Objects.equals(v.getFieldName(), request.getFieldName())).findFirst()
+                .filter(field -> fieldName.equals(field.getFieldName()))
+                .findFirst()
                 .orElse(null);
-        if (fieldInfo != null) {
-            switch (request.getOperationType()) {
-                case "=":
-                    ifFilter = !Objects.equals(request.getTargetValue(), fieldInfo.getFieldValue());
-                    break;
-                case "!=":
-                    ifFilter = Objects.equals(request.getTargetValue(), fieldInfo.getFieldValue());
-                    break;
-                case "like":
-                    ifFilter = fieldInfo.getFieldValue() != null
-                            && !fieldInfo.getFieldValue().contains(request.getTargetValue());
-            }
+        if (fieldInfo == null) {
+            return false;
+        }
+
+        boolean ifFilter = false;
+        String fieldValue = fieldInfo.getFieldValue();
+        // support targetValue == blank string (null or "") or targetValue != blank string
+        String targetValue = request.getTargetValue();
+        switch (request.getOperationType()) {
+            case "=":
+                ifFilter = !Objects.equals(targetValue, fieldValue);
+                break;
+            case "!=":
+                ifFilter = Objects.equals(targetValue, fieldValue);
+                break;
+            case "like":
+                // fieldValue or targetValue is null, like operation is invalid, so this record should be filtered
+                ifFilter = fieldValue == null || targetValue == null || !fieldValue.contains(targetValue);
         }
         return ifFilter;
     }

inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/message/RawMsgDeserializeOperator.java

@@ -35,7 +35,6 @@
 import org.springframework.stereotype.Service;
 
 import java.nio.charset.Charset;
-import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 
@@ -65,6 +64,7 @@ public List<BriefMQMessage> decodeMsg(InlongStreamInfo streamInfo, List<BriefMQM
             if (checkIfFilter(request, fieldList)) {
                 return briefMQMessages;
             }
+
             BriefMQMessage briefMQMessage = BriefMQMessage.builder()
                     .id(index)
                     .inlongGroupId(groupId)
@@ -75,12 +75,12 @@ public List<BriefMQMessage> decodeMsg(InlongStreamInfo streamInfo, List<BriefMQM
                     .body(body)
                     .fieldList(fieldList)
                     .build();
-            briefMQMessages.addAll(Collections.singletonList(briefMQMessage));
+            briefMQMessages.add(briefMQMessage);
             return briefMQMessages;
         } catch (Exception e) {
-            String errMsg = String.format("decode msg failed for groupId=%s, streamId=%s", groupId, streamId);
+            String errMsg = String.format("Failed to decode msg for groupId=%s, streamId=%s", groupId, streamId);
             log.error(errMsg, e);
-            throw new BusinessException(errMsg);
+            throw new BusinessException(errMsg + ", message: " + e.getMessage());
         }
     }
 

inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/queue/pulsar/PulsarOperator.java

@@ -448,18 +448,21 @@ private List<BriefMQMessage> queryMessageFromPulsar(String topicPartition, Pulsa
             int index, InlongStreamInfo streamInfo, int messagePosition, QueryMessageRequest request) {
         List<BriefMQMessage> briefMQMessages = new ArrayList<>();
         try {
-            ResponseEntity<byte[]> httpResponse =
-                    PulsarUtils.examineMessage(restTemplate, pulsarClusterInfo, topicPartition, "latest",
-                            messagePosition);
+            ResponseEntity<byte[]> httpResponse = PulsarUtils.examineMessage(restTemplate, pulsarClusterInfo,
+                    topicPartition, "latest", messagePosition);
             PulsarMessageInfo messageInfo = PulsarUtils.getMessageFromHttpResponse(httpResponse, topicPartition);
+            if (messageInfo == null) {
+                return briefMQMessages;
+            }
+
             Map<String, String> headers = messageInfo.getProperties();
             if (headers == null) {
                 headers = new HashMap<>();
             }
             MessageWrapType messageWrapType = MessageWrapType.forType(streamInfo.getWrapType());
-            if (headers.get(InlongConstants.MSG_ENCODE_VER) != null) {
-                messageWrapType =
-                        MessageWrapType.valueOf(Integer.parseInt(headers.get(InlongConstants.MSG_ENCODE_VER)));
+            String encodeType = headers.get(InlongConstants.MSG_ENCODE_VER);
+            if (StringUtils.isNotBlank(encodeType)) {
+                messageWrapType = MessageWrapType.valueOf(Integer.parseInt(encodeType));
             }
             DeserializeOperator deserializeOperator = deserializeOperatorFactory.getInstance(messageWrapType);
             deserializeOperator.decodeMsg(streamInfo, briefMQMessages, messageInfo.getBody(), headers, index, request);