Merge pull request #79 from odpf/unknown_fields_add_metrics

Increment the error record count for messages with unknownfields
gojekfarm · Jul 22, 2021 · 002ef57 · 002ef57
2 parents 954e3d1 + 6266f8f
commit 002ef57
Showing 1 changed file with 6 additions and 3 deletions.
diff --git a/src/main/java/com/gojek/beast/converter/RowMapper.java b/src/main/java/com/gojek/beast/converter/RowMapper.java
@@ -41,11 +41,14 @@ private Map<String, Object> getMappings(DynamicMessage message, ColumnMapping co
         if (message == null || columnMapping == null || columnMapping.isEmpty()) {
             return new HashMap<>();
         }
-        if (failOnUnknownFields && message.getUnknownFields().asMap().size() > 0) {
-            statsClient.increment("kafka.protobuf.unknownfields.errors");
+        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());
-            throw new UnknownProtoFieldFoundException(serializedUnknownFields, serializedMessage);
+            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();