Use of Confluent.SchemaRegistry.Serdes.ProtobufDeserializer with Kafka Connect converter io.confluent.connect.protobuf.ProtobufConverter throws Google.Protobuf.InvalidProtocolBufferException

[CooperWolfe]

Description

When I use a Kafka Connect configuration with the io.confluent.connect.protobuf.ProtobufConverter converter and use the Confluent.SchemaRegistry.Serdes.ProtobufDeserializer when constructing my consumer, I get a Google.Protobuf.InvalidProtocolBufferException.

How to reproduce

• Set up a Kafka Connect connector with the io.confluent.connect.protobuf.ProtobufConverter, e.g.

  {
  "name": "example",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "plugin.name":"pgoutput",
    "slot.name": "example",
  
    "database.hostname": "postgres",
    "database.port": "5432",
    "database.user": "postgres",
    "database.password": "insecurepassword",
    "database.dbname" : "example",
    "database.server.name": "postgres",
    "table.include.list": "example.example",
  
    "key.converter":"org.apache.kafka.connect.storage.StringConverter",
    "value.converter":"io.confluent.connect.protobuf.ProtobufConverter",
    "value.converter.schema.registry.url": "http://schema-registry:8081"
  }
  }

• Construct a consumer like so:

  public IConsumer<string, T> Create<T>() where T : class, IMessage<T>, new()
  {
  var config = new ConsumerConfig
  {
      EnableAutoCommit = false,
      AutoOffsetReset = AutoOffsetReset.Earliest,
      AllowAutoCreateTopics = true,
      GroupId = AppDomain.CurrentDomain.FriendlyName,
      ClientId = AppDomain.CurrentDomain.FriendlyName,
      BootstrapServers = "kafka:9092"
  };
  return new ConsumerBuilder<string, T>(config)
      .SetKeyDeserializer(Deserializers.Utf8)
      .SetValueDeserializer(new ProtobufDeserializer<T>().AsSyncOverAsync())
      .Build();
  }

• Call consumer.Consume(cancellationToken)
• The following exception is thrown:

  Unhandled exception. Confluent.Kafka.ConsumeException: Local: Value deserialization error
  ---> Google.Protobuf.InvalidProtocolBufferException: While parsing a protocol message, the input ended unexpectedly in the middle of a field.  This could mean either that the input has been truncated or that an embedded message misreported its own length.
  at Google.Protobuf.ParsingPrimitives.ValidateCurrentLimit(ReadOnlySpan`1& buffer, ParserInternalState& state, Int32 size)
  at Google.Protobuf.ParsingPrimitives.ReadRawBytesSlow(ReadOnlySpan`1& buffer, ParserInternalState& state, Int32 size)
  at Google.Protobuf.ParsingPrimitives.ReadRawBytes(ReadOnlySpan`1& buffer, ParserInternalState& state, Int32 size)
  at Google.Protobuf.UnknownFieldSet.MergeFieldFrom(ParseContext& ctx)
  at Google.Protobuf.UnknownFieldSet.MergeFieldFrom(UnknownFieldSet unknownFields, ParseContext& ctx)
  at Google.Protobuf.WellKnownTypes.Timestamp.pb::Google.Protobuf.IBufferMessage.InternalMergeFrom(ParseContext& input)
  at Google.Protobuf.ParsingPrimitivesMessages.ReadRawMessage(ParseContext& ctx, IMessage message)
  at Google.Protobuf.ParsingPrimitivesMessages.ReadMessage(ParseContext& ctx, IMessage message)
  at Outbox.OutboxEntry.pb::Google.Protobuf.IBufferMessage.InternalMergeFrom(ParseContext& input)
  at Google.Protobuf.ParsingPrimitivesMessages.ReadRawMessage(ParseContext& ctx, IMessage message)
  at Google.Protobuf.CodedInputStream.ReadRawMessage(IMessage message)
  at Outbox.OutboxEntry.MergeFrom(CodedInputStream input)
  at Google.Protobuf.MessageExtensions.MergeFrom(IMessage message, Stream input, Boolean discardUnknownFields, ExtensionRegistry registry)
  at Google.Protobuf.MessageParser`1.ParseFrom(Stream input)
  at Confluent.SchemaRegistry.Serdes.ProtobufDeserializer`1.DeserializeAsync(ReadOnlyMemory`1 data, Boolean isNull, SerializationContext context)
  at Confluent.Kafka.SyncOverAsync.SyncOverAsyncDeserializer`1.Deserialize(ReadOnlySpan`1 data, Boolean isNull, SerializationContext context)
  at Confluent.Kafka.Consumer`2.Consume(Int32 millisecondsTimeout)
  --- End of inner exception stack trace ---
  at Confluent.Kafka.Consumer`2.Consume(Int32 millisecondsTimeout)
  at Confluent.Kafka.Consumer`2.Consume(CancellationToken cancellationToken)
  ...

Kafka doesn't seem to be logging anything interesting other than seeing the failure by my service and booting it. Zookeeper has nothing of interest either. Additionally, Kafka Connect (or more specifically, Debezium) seems to be producing its messages without a hitch.

Versions

Confluent.Kafka nuget version: 1.7.0 Scala version: 2.12 Apache Kafka version: 2.8.0 OS: Windows 10 Home

Checklist

Please provide the following information:

• [ ] A complete (i.e. we can run it), minimal program demonstrating the problem. No need to supply a project file.
• [x] Confluent.Kafka nuget version.
• [x] Apache Kafka version.
• [x] Client configuration.
• [x] Operating system.
• [x] Provide logs (with "debug" : "..." as necessary in configuration).
• [ ] Provide broker log excerpts.
• [x] Critical issue.

Thank you to any and all who can help.

[mhowlett]

I suspect you might be hitting the framing incompatibility between .net and java serdes. The format was changed mid-development of this feature in schema registry, and the message didn't get through to the clients team - the non-java clients implemented the original proposed format. we have changed this in v1.8 (not released yet).

https://github.com/confluentinc/confluent-kafka-dotnet/pull/1628/files#diff-635a634d86028a5efbc4d6913293a861c4ff487f5983e10684f56228560a85d5R42

[CooperWolfe]

Thank you for the quick answer @mhowlett. I will try making and using a v1.8 locally tomorrow with latest master.

[CooperWolfe]

I cloned this repository and packed a local nuget package (bumped the version to 1.8.0, so no collisions with the nuget.org source). Even while using this, there seems to be no change in the error. Trying another converter. I would imagine the community has used this with Kafka Connect before. Perhaps someone has had success with another converter, or even the same one?

[CooperWolfe]

I did some intense testing. I got a console application up in my environment to produce an "identical" message. That is, a message created from the same proto file as was used to put the data into the database that Debezium was listening to. After some creative use of kafka-console-consumer, I was able to produce the following two byte streams:

The Kafka Connect byte stream:

The Confluent.Kafka producer:

For the record, this was the .proto:

...

import "google/protobuf/timestamp.proto";

message OutboxEntry {
    string payload = 1;
    string id = 2;
    string aggregateId = 3;
    string eventType = 4;
    google.protobuf.Timestamp createdAt = 5;
}

Without even breaking down the bytes, you can see that in the first byte stream, the value of createdAt was serialized as a string, not a google.protobuf.Timestamp. Seems likely to be the cause, but just to be sure: is there any reason this would NOT be the cause?

[CooperWolfe]

Looking in schema registry, it seems the schema produced by Kafka Connect was:

While the schema produced by the Confluent.Kafka producer was:

I believe I am certain this is my issue, but will adjust and confirm

[CooperWolfe]

It seems Debezium by default converts timestamp columns in Postgres using string's. This causes complications when it detects the google.protobuf.Timestamp field in the schema registry. The solution I went with was to grit my teeth and change my proto to use a string for the time being. Once I did that, I reverted to version 1.7.0 and all was still well. Learned a lot from this, thanks again @mhowlett for your answer.