[FLINK-38205][format][pb] Discard unknown fields by default

Author: X-czh

flink-formats/flink-protobuf/src/main/java/org/apache/flink/formats/protobuf/PbConstant.java

@@ -21,7 +21,7 @@
 /** Keeps protobuf constants separately. */
 public class PbConstant {
     public static final String PB_METHOD_GET_DESCRIPTOR = "getDescriptor";
-    public static final String PB_METHOD_PARSE_FROM = "parseFrom";
+    public static final String PB_METHOD_PARSER = "parser";
     public static final String GENERATED_DECODE_METHOD = "decode";
     public static final String GENERATED_ENCODE_METHOD = "encode";
     public static final String PB_MAP_KEY_NAME = "key";

flink-formats/flink-protobuf/src/main/java/org/apache/flink/formats/protobuf/deserialize/ProtoToRowConverter.java

@@ -37,6 +37,8 @@
 import com.google.protobuf.ByteString;
 import com.google.protobuf.Descriptors;
 import com.google.protobuf.Descriptors.FileDescriptor.Syntax;
+import com.google.protobuf.DiscardUnknownFieldsParser;
+import com.google.protobuf.Parser;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -53,8 +55,8 @@
  */
 public class ProtoToRowConverter {
     private static final Logger LOG = LoggerFactory.getLogger(ProtoToRowConverter.class);
-    private final Method parseFromMethod;
     private final Method decodeMethod;
+    private final Parser<?> protoParser;
     private boolean isCodeSplit = false;
 
     public ProtoToRowConverter(RowType rowType, PbFormatConfig formatConfig)
@@ -124,14 +126,16 @@ public ProtoToRowConverter(RowType rowType, PbFormatConfig formatConfig)
                             codegenAppender.code());
             decodeMethod =
                     generatedClass.getMethod(PbConstant.GENERATED_DECODE_METHOD, messageClass);
-            parseFromMethod = messageClass.getMethod(PbConstant.PB_METHOD_PARSE_FROM, byte[].class);
+            Method parserMethod = messageClass.getMethod(PbConstant.PB_METHOD_PARSER);
+            Parser originalProtoParser = (Parser) parserMethod.invoke(null);
+            protoParser = DiscardUnknownFieldsParser.wrap(originalProtoParser);
         } catch (Exception ex) {
             throw new PbCodegenException(ex);
         }
     }
 
     public RowData convertProtoBinaryToRow(byte[] data) throws Exception {
-        Object messageObj = parseFromMethod.invoke(null, data);
+        Object messageObj = protoParser.parseFrom(data);
         return (RowData) decodeMethod.invoke(null, messageObj);
     }