[CRUNCH-363] Cogroup using Protobufs with WritableTypeFamily throws Proto Exception

• Type: Bug
• Status: Closed
• Priority: Major
• Resolution: Fixed
• Affects Version/s: 0.8.2
• Fix Version/s: 0.10.0, 0.8.3
• Component/s: Core
• Assignee: Josh Wills
• Reporter: Micah Whitacre
• Created: 12/Mar/14 21:13
• Updated: 20/Jun/14 03:49
• Resolved: 13/Mar/14 21:15

Description

If you have code like the following:

PTable<String, Proto> t1 = ...
PTable<String, Proto> t2 = ...
ti.cogroup(t2);

Where the PType of each table was created using:
ptf.tableOf(ptf.strings(), PTypes.protos(Person.class, ptf));

and "ptf" is an instance of WritableTypeFamily.

You will get an exception like the following.

org.apache.crunch.CrunchRuntimeException: com.google.protobuf.InvalidProtocolBufferException: Protocol message contained an invalid tag (zero).
at org.apache.crunch.types.PTypes$ProtoInputMapFn.map(PTypes.java:191)
at org.apache.crunch.types.PTypes$ProtoInputMapFn.map(PTypes.java:160)
at org.apache.crunch.fn.CompositeMapFn.map(CompositeMapFn.java:63)
at org.apache.crunch.types.writable.Writables$UWInputFn.map(Writables.java:611)
at org.apache.crunch.types.writable.Writables$UWInputFn.map(Writables.java:573)
at org.apache.crunch.types.PGroupedTableType$HoldLastIterator.next(PGroupedTableType.java:84)
at org.apache.crunch.lib.Cogroup$PostGroupFn.map(Cogroup.java:275)
at org.apache.crunch.lib.Cogroup$PostGroupFn.map(Cogroup.java:250)
at org.apache.crunch.fn.PairMapFn.map(PairMapFn.java:62)
at org.apache.crunch.fn.PairMapFn.map(PairMapFn.java:26)
at org.apache.crunch.MapFn.process(MapFn.java:34)
at org.apache.crunch.impl.mr.run.RTNode.process(RTNode.java:98)
at org.apache.crunch.impl.mr.emit.IntermediateEmitter.emit(IntermediateEmitter.java:56)
at org.apache.crunch.MapFn.process(MapFn.java:34)
at org.apache.crunch.impl.mr.run.RTNode.process(RTNode.java:98)
at org.apache.crunch.impl.mr.run.RTNode.processIterable(RTNode.java:113)
at org.apache.crunch.impl.mr.run.CrunchReducer.reduce(CrunchReducer.java:57)
at org.apache.hadoop.mapreduce.Reducer.run(Reducer.java:176)
at org.apache.hadoop.mapred.ReduceTask.runNewReducer(ReduceTask.java:650)
at org.apache.hadoop.mapred.ReduceTask.run(ReduceTask.java:418)
at org.apache.hadoop.mapred.LocalJobRunner$Job.run(LocalJobRunner.java:262)
Caused by: com.google.protobuf.InvalidProtocolBufferException: Protocol message contained an invalid tag (zero).
at com.google.protobuf.InvalidProtocolBufferException.invalidTag(InvalidProtocolBufferException.java:89)
at com.google.protobuf.CodedInputStream.readTag(CodedInputStream.java:108)
at org.apache.crunch.lib.PersonProtos$Person.<init>(PersonProtos.java:101)
at org.apache.crunch.lib.PersonProtos$Person.<init>(PersonProtos.java:65)
at org.apache.crunch.lib.PersonProtos$Person$1.parsePartialFrom(PersonProtos.java:153)
at org.apache.crunch.lib.PersonProtos$Person$1.parsePartialFrom(PersonProtos.java:148)
at org.apache.crunch.lib.PersonProtos$Person$Builder.mergeFrom(PersonProtos.java:484)
at org.apache.crunch.lib.PersonProtos$Person$Builder.mergeFrom(PersonProtos.java:369)
at com.google.protobuf.AbstractMessageLite$Builder.mergeFrom(AbstractMessageLite.java:196)
at com.google.protobuf.AbstractMessage$Builder.mergeFrom(AbstractMessage.java:898)
at com.google.protobuf.AbstractMessage$Builder.mergeFrom(AbstractMessage.java:267)
at org.apache.crunch.types.PTypes$ProtoInputMapFn.map(PTypes.java:189)
... 20 more

Note you don't get the same exception for AvroTypeFamily.