fix unknown fields error count metric, change exception message

src/main/java/com/gojek/beast/converter/RowMapper.java

@@ -6,8 +6,8 @@
 import com.gojek.beast.converter.fields.ProtoField;
 import com.gojek.beast.exception.UnknownProtoFieldFoundException;
 import com.gojek.beast.models.ConfigurationException;
-import com.gojek.beast.protomapping.UnknownProtoFields;
 import com.gojek.beast.stats.Stats;
+import com.gojek.beast.util.DynamicMessageUtil;
 import com.google.protobuf.Descriptors;
 import com.google.protobuf.DynamicMessage;
 import lombok.AllArgsConstructor;
@@ -34,22 +34,23 @@ public Map<String, Object> map(DynamicMessage message) {
         if (mapping == null) {
             throw new ConfigurationException("BQ_PROTO_COLUMN_MAPPING is not configured");
         }
+
+        if (DynamicMessageUtil.hasUnknownField(message)) {
+            statsClient.count("kafka.error.records.count,type=unknownfields," + statsClient.getBqTags(), 1);
+            String serialisedProtoMessage = message.toString();
+            log.warn(String.format("unknown fields found in proto : %s", serialisedProtoMessage));
+            if (failOnUnknownFields) {
+                throw new UnknownProtoFieldFoundException(serialisedProtoMessage);
+            }
+        }
+
         return getMappings(message, mapping);
     }
 
     private Map<String, Object> getMappings(DynamicMessage message, ColumnMapping columnMapping) {
         if (message == null || columnMapping == null || columnMapping.isEmpty()) {
             return new HashMap<>();
         }
-        if (message.getUnknownFields().asMap().size() > 0) {
-            statsClient.count("kafka.error.records.count,type=unknownfields," + statsClient.getBqTags(), 1);
-            String serializedUnknownFields = message.getUnknownFields().asMap().keySet().toString();
-            String serializedMessage = UnknownProtoFields.toString(message.toByteArray());
-            log.warn(String.format("[%s] unknown fields found in proto [%s], either update mapped protobuf or disable FAIL_ON_UNKNOWN_FIELDS",
-                    serializedUnknownFields, serializedMessage));
-            if (failOnUnknownFields)
-                throw new UnknownProtoFieldFoundException(serializedUnknownFields, serializedMessage);
-        }
         Descriptors.Descriptor descriptorForType = message.getDescriptorForType();
 
         Map<String, Object> row = new HashMap<>(columnMapping.size());

src/main/java/com/gojek/beast/exception/UnknownProtoFieldFoundException.java

@@ -1,8 +1,7 @@
 package com.gojek.beast.exception;
 
 public class UnknownProtoFieldFoundException extends RuntimeException {
-    public UnknownProtoFieldFoundException(String serializedUnknownFields, String serializedMessage) {
-        super(String.format("[%s] unknown fields found in proto [%s], either update mapped protobuf or disable FAIL_ON_UNKNOWN_FIELDS",
-                serializedUnknownFields, serializedMessage));
+    public UnknownProtoFieldFoundException(String serialisedProtoMessage) {
+        super(String.format("some unknown fields found, please check the published message, update mapped protobuf or disable FAIL_ON_UNKNOWN_FIELDS, full proto message : %s", serialisedProtoMessage));
     }
 }

src/main/java/com/gojek/beast/util/DynamicMessageUtil.java

@@ -0,0 +1,42 @@
+package com.gojek.beast.util;
+
+import com.google.protobuf.DynamicMessage;
+
+import java.util.Collections;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Queue;
+import java.util.stream.Collectors;
+
+public class DynamicMessageUtil {
+    public static boolean hasUnknownField(DynamicMessage root) {
+        List<DynamicMessage> dynamicMessageFields = collectNestedFields(root);
+        List<DynamicMessage> messageWithUnknownFields = getMessageWithUnknownFields(dynamicMessageFields);
+        return messageWithUnknownFields.size() > 0;
+    }
+
+    private static List<DynamicMessage> collectNestedFields(DynamicMessage node) {
+        List<DynamicMessage> output = new LinkedList<>();
+        Queue<DynamicMessage> stack = Collections.asLifoQueue(new LinkedList<>());
+        stack.add(node);
+        while (true) {
+            DynamicMessage current = stack.poll();
+            if (current == null) {
+                break;
+            }
+            List<DynamicMessage> nestedChildNodes = current.getAllFields().values().stream()
+                    .filter(field -> field instanceof DynamicMessage)
+                    .map(field -> (DynamicMessage) field)
+                    .collect(Collectors.toList());
+            stack.addAll(nestedChildNodes);
+
+            output.add(current);
+        }
+
+        return output;
+    }
+
+    private static List<DynamicMessage> getMessageWithUnknownFields(List<DynamicMessage> messages) {
+        return messages.stream().filter(message -> message.getUnknownFields().asMap().size() > 0).collect(Collectors.toList());
+    }
+}