Reading composite types from Kafka is broken

[zilder]

CREATE TYPE abc AS (a INT, b TEXT, c TIMESTAMP);

CREATE FOREIGN TABLE kafka_json (
	part int OPTIONS (partition 'true'),
	offs bigint OPTIONS (offset 'true'),
	x abc)
SERVER kafka_server
OPTIONS (format 'json', topic 'json', batch_size '30', buffer_delay '100');

INSERT INTO kafka_json (x) VALUES ((1, 'test', current_timestamp));

SELECT * FROM kafka_json;
ERROR:  malformed record literal: "{"a":1,"b":"test","c":"2018-07-04T11:16:55.671986"}"
DETAIL:  Missing left parenthesis.

Postgres expects an input string formatted like:

(1,"test","2018-07-04T11:16:55.671986")

for composite type. The easy solution would be to just remove keys from input string and replace {} with (). But this won't work if JSON document has different key order or extra or missing keys. So it makes sense to write a simple JSON parser to collect individual key-value pairs, reorder them if needed and fill the gaps with NULLs. It's also possible to use built-in JSONB facilities from postgres to parse and manage JSON documents but I anticipate that it would be less efficient than custom designed solution.