title: Using protobuf with flink
body: <p>I'm using flink to read data from kafka and convert it to protobuf. The problem I'm facing is when I run the java application I get the below error. If I modify the <code>unknownFields</code> variable name to something else, it works but it's hard to make this change on all protobuf classes. </p>
  
  <p>I also tried to deserialize directly when reading from kafka but I'm not sure what should be the <code>TypeInformation</code> to be returned for <code>getProducedType()</code> method.</p>
  
  <pre><code>    public static class ProtoDeserializer implements DeserializationSchema{
  
      @Override
      public TypeInformation getProducedType() {
          // TODO Auto-generated method stub
          return PrimitiveArrayTypeInfo.BYTE_PRIMITIVE_ARRAY_TYPE_INFO;
      }
  </code></pre>
  
  <p>Appreciate all the help. Thanks.</p>
  
  <blockquote>
    <p>java.lang.RuntimeException: The field protected com.google.protobuf.UnknownFieldSet com.google.protobuf.GeneratedMessage.unknownFields is already contained in the hierarchy of the class com.google.protobuf.GeneratedMessage.Please use unique field names through your classes hierarchy
        at org.apache.flink.api.java.typeutils.TypeExtractor.getAllDeclaredFields(TypeExtractor.java:1594)
        at org.apache.flink.api.java.typeutils.TypeExtractor.analyzePojo(TypeExtractor.java:1515)
        at org.apache.flink.api.java.typeutils.TypeExtractor.privateGetForClass(TypeExtractor.java:1412)
        at org.apache.flink.api.java.typeutils.TypeExtractor.privateGetForClass(TypeExtractor.java:1319)
        at org.apache.flink.api.java.typeutils.TypeExtractor.createTypeInfoWithTypeHierarchy(TypeExtractor.java:609)
        at org.apache.flink.api.java.typeutils.TypeExtractor.privateCreateTypeInfo(TypeExtractor.java:437)
        at org.apache.flink.api.java.typeutils.TypeExtractor.getUnaryOperatorReturnType(TypeExtractor.java:306)
        at org.apache.flink.api.java.typeutils.TypeExtractor.getFlatMapReturnTypes(TypeExtractor.java:133)
        at org.apache.flink.streaming.api.datastream.DataStream.flatMap(DataStream.java:529)</p>
  </blockquote>
  
  <p>Code:</p>
  
  <pre><code>    FlinkKafkaConsumer09&lt;byte[]&gt; kafkaConsumer = new FlinkKafkaConsumer09&lt;&gt;("testArr",new ByteDes(),p);
  
      DataStream&lt;byte[]&gt; input = env.addSource(kafkaConsumer);
      DataStream&lt;PBAddress&gt; protoData = input.map(new RichMapFunction&lt;byte[], PBAddress&gt;() {
          @Override
          public PBAddress map(byte[] value) throws Exception {
              PBAddress addr = PBAddress.parseFrom(value);
              return addr;
          }
      });
  </code></pre>