Multiple-field key schema inference should be handled consistently between Protobuf and other formats

Is your feature request related to a problem? Please describe. Key schema inference between Protobuf and other formats like Avro is not currently consistent. For example, with a simple Protobuf message containing two fields:

ksql> create stream test (val string) with (kafka_topic='test', key_format='protobuf', value_format='json');
The key schema for topic test contains multiple columns, which is not supported by ksqlDB at this time.
Schema:syntax = "proto3";
package com.mycorp.mynamespace;

message Key {
  string k1 = 1;
  string k2 = 2;
}

With essentially the same schema in Avro:

{
  "fields": [
    {
      "name": "k1",
      "type": "string"
    },
    {
      "name": "k2",
      "type": "string"
    }
  ],
  "name": "key",
  "namespace": "com.mycorp.mynamespace",
  "type": "record"
}

it works:

ksql> create stream test (val string) with (kafka_topic='test2', key_format='avro', value_format='json');

 Message
----------------
 Stream created
----------------

ksql> describe test;

Name                 : TEST
 Field  | Type
---------------------------------------------------------------
 ROWKEY | STRUCT<K1 VARCHAR(STRING), K2 VARCHAR(STRING)> (key)
 VAL    | VARCHAR(STRING)
---------------------------------------------------------------

The difference appears to be that with Avro, the key schema is wrapped in a single struct field, while with Protobuf, the top level message is unwrapped into separate fields.

Describe the solution you'd like Make behavior for multi-field key schema inference consistent.

Describe alternatives you've considered Defining the key schema explicitly is a workaround for Protobuf:

create stream test (k1 string key, k2 string key, val string) with (kafka_topic='test', key_format='protobuf', value_format='json');

confluentinc / ksql

Multiple-field key schema inference should be handled consistently between Protobuf and other formats #8222