Support copying of JSON data to pubsub topic as JSON instead of stringified map.

[sriramkoopidai]

Problem: The issue is with the sink connector, If schema is not used, JSON data in topic is pushed to pubsub topic as string value of the map.
We need a feature so that we can force the output to be JSON and could be interpreted as JSON in pubsub.

Example for JSON object
{"name":"bob","id":45}

it is shown as on pubsub topic
{name=bob, id=45}

Add a configuration which could be used to force output to be JSON.

Alternativies considered:
Tried forcing the output to be string using configuration
value.converter=org.apache.kafka.connect.storage.StringConverter

pubsub topic gets the right value but the single message transforms didn't not work, for example if you want to copy json field to header, it does not work.

Example: Sample SMT

"headers.publish" : true,
"transforms": "insertAppIdHeader,moveFieldsToHeader",
"transforms.moveFieldsToHeader.type": "org.apache.kafka.connect.transforms.HeaderFrom$Value",
"transforms.moveFieldsToHeader.fields": "name,id",
"transforms.moveFieldsToHeader.headers": "name,id",

[lqiu96]

Do you have a repro that we could follow/ use?

CC: @anguillanneuf who may have more context regarding this issue.

[samarthsingal]

Could you share the sink connector config you are using?

[ThomasBarach]

Hello, I'm hitting this issue as well. Did you manage to do something about it?
Regards

[CatherineThompson]

Hi, I've also run into this issue. Is there a workaround?

[ian-q1]

Hello, also running into this problem, is there a reason this PR #203 was closed?

sample config

connector.class=com.google.pubsub.kafka.sink.CloudPubSubSinkConnector
cps.project=<project>
cps.topic=kafka-test
key.converter=org.apache.kafka.connect.converters.ByteArrayConverter
name=pubsub_sink
tasks.max=3
topics=product_viewed
value.converter=org.apache.kafka.connect.json.JsonConverter
value.converter.schemas.enable=false
transforms=ReplaceField
transforms.ReplaceField.type=org.apache.kafka.connect.transforms.ReplaceField$Value
transforms.ReplaceField.blacklist=datajson

If I use org.apache.kafka.connect.converters.ByteArrayConverter for the value converter, it's passed to pubsub as a json string, same as producer input. However the transform causes an error in the pubsub sink. If I use org.apache.kafka.connect.json.JsonConverter for the value converter, the transform works but the message is sent to pubsub as a stringified map e.g:

{data_base64=Cgdzb3VyY2UxEgdkZXZpY2UxGghzZXNzaW9uMSIENDM1Ng==, datacontenttype=binary, subject=subject, siteid=123, specversion=1.0, id=0196f31d-1f6f-7ce2-879d-3b63e557a53b, source=fasdfasfdsf, time=2025-05-21T13:51:34.255853609Z, type=test, dataschema=test.proto}

Ideally the message should be sent to pubsub as a json string. If the pubsub topic has a schema, this message will fail schema validation.