fix unknown fields error count metric, change exception message

[fzrvic]

No description provided.

[sravankorumilli]

As discussed over call, using an accumulator will help in collecting the messages properly

final List<DynamicMessage> allNestedMessagesInProto = getAllNestedMessages(message, List.of(message));
        List<DynamicMessage> messageWithUnknownFields = getMessageWithUnknownFields(allNestedMessagesInProto);
        if (messageWithUnknownFields.size() > 0) {
            statsClient.count("kafka.error.records.count,type=unknownfields," + statsClient.getBqTags(), 1);
            if (failOnUnknownFields) {
                throw new UnknownProtoFieldFoundException(message.toString());
            }
        }
        return getMappings(message, mapping);
    }

private List<DynamicMessage> getAllNestedMessages(DynamicMessage message, List<DynamicMessage> nestedMessages) {
        List<DynamicMessage> children = message.getAllFields().values().stream()
                .filter(v -> v instanceof DynamicMessage)
                .map(o -> (DynamicMessage) o)
                .collect(Collectors.toList());
        nestedMessages.addAll(children);
        children.forEach(m -> getAllNestedMessages(m, nestedMessages));
        return nestedMessages;
    }

[sravankorumilli]

the root message should be there in the dynamicMessageFields

[fzrvic]

it automatically add every DynamicMessage to accumulator when traversing the each of the DynamicMessage trees, including the very first node (root)

   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);
    }

[sravankorumilli]

Log the message with a warn when when unknowfields exists

[fzrvic]

will add log