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,21 @@ public Map<String, Object> map(DynamicMessage message) {
         if (mapping == null) {
             throw new ConfigurationException("BQ_PROTO_COLUMN_MAPPING is not configured");
         }
+
+        if (DynamicMessageUtil.isUnknownFieldExist(message)) {
+            statsClient.count("kafka.error.records.count,type=unknownfields," + statsClient.getBqTags(), 1);
+            if (failOnUnknownFields) {
+                throw new UnknownProtoFieldFoundException(message.toString());
+            }
+        }
+
         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,33 @@
+package com.gojek.beast.util;
+
+import com.google.protobuf.DynamicMessage;
+
+import java.util.LinkedList;
+import java.util.List;
+import java.util.stream.Collectors;
+
+public class DynamicMessageUtil {
+    public static boolean isUnknownFieldExist(DynamicMessage dynamicMessage) {
+        if (dynamicMessage == null) {
+            return false;
+        }
+        List<DynamicMessage> dynamicMessageFields = new LinkedList<>();
+        collectNestedFields(dynamicMessage, dynamicMessageFields);
+        List<DynamicMessage> messageWithUnknownFields = getMessageWithUnknownFields(dynamicMessageFields);
+        return messageWithUnknownFields.size() > 0;
+    }
+
+    private static void collectNestedFields(DynamicMessage root, List<DynamicMessage> accumulator) {
+        List<DynamicMessage> nestedChild = root.getAllFields().values().stream()
+                .filter(v -> v instanceof DynamicMessage)
+                .map(o -> (DynamicMessage) o)
+                .collect(Collectors.toList());
+
+        nestedChild.forEach(m -> collectNestedFields(m, accumulator));
+        accumulator.add(root);
+    }
+
+    private static List<DynamicMessage> getMessageWithUnknownFields(List<DynamicMessage> messages) {
+        return messages.stream().filter(message -> message.getUnknownFields().asMap().size() > 0).collect(Collectors.toList());
+    }
+}