Skip to content
This repository has been archived by the owner on Feb 12, 2022. It is now read-only.

fix unknown fields error count metric, change exception message #80

Open
wants to merge 6 commits into
base: master
Choose a base branch
from
Open

fix unknown fields error count metric, change exception message #80

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
a929420
fix: fix unknown fields error count metric, change exception message
fzrvic Jul 23, 2021
6bd9f5a
fix: fix unknown fields checking, remove duplicated nested field list
fzrvic Jul 28, 2021
eae47fe
fix: fix clash class name with testing class
fzrvic Jul 28, 2021
748b224
feat: add log warning when unknown fields found
fzrvic Jul 29, 2021
302daa7
refactor: change implementation unknown fields checking
fzrvic Jul 30, 2021
b947b6f
refactor: rename function, remove redundant code
fzrvic Jul 30, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
39 changes: 29 additions & 10 deletions src/main/java/com/gojek/beast/converter/RowMapper.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,6 @@
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.google.protobuf.Descriptors;
import com.google.protobuf.DynamicMessage;
Expand All @@ -15,8 +14,10 @@

import java.util.ArrayList;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

@Slf4j
@AllArgsConstructor
Expand All @@ -34,22 +35,40 @@ public Map<String, Object> map(DynamicMessage message) {
if (mapping == null) {
throw new ConfigurationException("BQ_PROTO_COLUMN_MAPPING is not configured");
}

List<DynamicMessage> messageWithUnknownFields = getMessageWithUnknownFields(getDynamicMessageFields(message));
if (messageWithUnknownFields.size() > 0) {
statsClient.count("kafka.error.records.count,type=unknownfields," + statsClient.getBqTags(), 1);
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Log the message with a warn when when unknowfields exists

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

will add log

if (failOnUnknownFields) {
throw new UnknownProtoFieldFoundException(message.toString());
}
}

return getMappings(message, mapping);
}

private List<DynamicMessage> getDynamicMessageFields(DynamicMessage message) {
Copy link
Contributor

@sravankorumilli sravankorumilli Jul 28, 2021 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

List<DynamicMessage> fields = new LinkedList<>();
fields.add(message);

List<DynamicMessage> children = message.getAllFields().values().stream().filter(v -> v instanceof DynamicMessage).map(o -> (DynamicMessage) o).collect(Collectors.toList());
children.forEach(m -> {
List<DynamicMessage> messageFields = getDynamicMessageFields(m);
fields.addAll(messageFields);
});

return fields;
}

private List<DynamicMessage> getMessageWithUnknownFields(List<DynamicMessage> messages) {
return messages.stream().filter(message -> message.getUnknownFields().asMap().size() > 0).collect(Collectors.toList());
}


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());
Expand Down
5 changes: 2 additions & 3 deletions src/main/java/com/gojek/beast/exception/UnknownProtoFieldFoundException.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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));
}
}