with_metadata not working in Python SDK ReadFromKafka

[prasanthkn83]

Here's the error generated when with_metadata is set as True in ReadFromKafka. It seems the SchemaCoder is not matching with KafkaConsumer record which has an extra timestamp_type at position 4 which is missing in the below SchemaCoder. Please check.

generic::unknown: org.apache.beam.sdk.util.UserCodeException: java.lang.IllegalArgumentException: Unable to encode element 'org.apache.beam.sdk.io.kafka.KafkaIO$ByteArrayKafkaRecord@45b2487a' with coder 'SchemaCoder<Schema: Fields: Field{name=topic, description=, type=STRING NOT NULL, options={{}}} Field{name=partition, description=, type=INT32 NOT NULL, options={{}}} Field{name=offset, description=, type=INT64 NOT NULL, options={{}}} Field{name=timestamp, description=, type=INT64 NOT NULL, options={{}}} Field{name=key, description=, type=BYTES, options={{}}} Field{name=value, description=, type=BYTES, options={{}}} Field{name=headers, description=, type=ARRAY<ROW<key STRING NOT NULL, value BYTES NOT NULL> NOT NULL>, options={{}}} Field{name=timestampTypeId, description=, type=INT32 NOT NULL, options={{}}} Field{name=timestampTypeName, description=, type=STRING NOT NULL, options={{}}} Encoding positions: {headers=6, timestampTypeName=8, partition=1, offset=2, topic=0, value=5, key=4, timestamp=3, timestampTypeId=7}