DataException: "Not a valid hex string" for Debezium BITS/bytes to Snowflake BINARY

Start by tracing the Streaming Ingest path from BufferedTopicPartitionChannel.java, especially the insertRecords and handleInsertRowsFailures locations named in the stack trace. Reproduce the Debezium io.debezium.data.Bits bytes-to-Snowflake BINARY case and verify that valid raw bytes are accepted without the “Not a valid hex string” failure.

We're experiencing an issue with the Snowflake Kafka Connector's Streaming Ingest API. When processing records captured by Debezium where a source BITS column (io.debezium.data.Bits, which is Kafka Connect bytes) is mapped to a Snowflake BINARY column, the connector fails.

The error is DataException: Not a valid hex string, raised from BufferedTopicPartitionChannel.java. This suggests the streaming ingest path might incorrectly expect hex-encoded strings instead of raw bytes when writing to BINARY columns. The full stack trace and Avro schema are provided below.

Caused by: org.apache.kafka.connect.errors.DataException: Error inserting Records using Streaming API with msg:The given row cannot be converted to the internal format due to invalid value: Value cannot be ingested into Snowflake column PERMISSIONS_PRESENT of type BINARY, rowIndex:0, reason: Not a valid hex string
	at com.snowflake.kafka.connector.internal.streaming.BufferedTopicPartitionChannel.handleInsertRowsFailures(BufferedTopicPartitionChannel.java:783)
	at com.snowflake.kafka.connector.internal.streaming.BufferedTopicPartitionChannel.insertRecords(BufferedTopicPartitionChannel.java:537)
	at com.snowflake.kafka.connector.internal.streaming.BufferedTopicPartitionChannel.insertRecord(BufferedTopicPartitionChannel.java:418)
	at com.snowflake.kafka.connector.internal.streaming.SnowflakeSinkServiceV2.insert(SnowflakeSinkServiceV2.java:410)
	at com.snowflake.kafka.connector.internal.streaming.SnowflakeSinkServiceV2.insert(SnowflakeSinkServiceV2.java:377)
	at com.snowflake.kafka.connector.SnowflakeSinkTask.put(SnowflakeSinkTask.java:311)
	at org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:601)

This is happening on a column of type BITS in the source database, io.debezium.data.Bits in debezium.

  {
    "subject": "dbz_avro.foobar-value",
    "version": 1,
    "id": 190,
    "schema": "{\"type\":\"record\",\"name\":\"Value\",\"namespace\":\"dbz_avro.foobar\",\"fields\":[{\"name\":\"provider_id\",\"type\":\"int\"},{\"name\":\"role\",\"type\":\"string\"},{\"name\":\"permissions\",\"type\":{\"type\":\"bytes\",\"connect.version\":1,\"connect.parameters\":{\"length\":\"128\"},\"connect.default\":\"\",\"connect.name\":\"io.debezium.data.Bits\"},\"default\":\"\"},{\"name\":\"permissions_present\",\"type\":{\"type\":\"bytes\",\"connect.version\":1,\"connect.parameters\":{\"length\":\"128\"},\"connect.default\":\"\",\"connect.name\":\"io.debezium.data.Bits\"},\"default\":\"\"},{\"name\":\"debezium_ids\",\"type\":{\"type\":\"string\",\"connect.version\":1,\"connect.default\":\"{}\",\"connect.name\":\"io.debezium.data.Json\"},\"default\":\"{}\"},{\"name\":\"debezium_ids_array\",\"type\":{\"type\":\"array\",\"items\":[\"null\",\"int\"]}},{\"name\":\"__deleted\",\"type\":[\"null\",\"string\"],\"default\":null},{\"name\":\"__dbz_table\",\"type\":[\"null\",\"string\"],\"default\":null},{\"name\":\"__dbz_lsn\",\"type\":[\"null\",\"long\"],\"default\":null},{\"name\":\"__dbz_op\",\"type\":[\"null\",\"string\"],\"default\":null},{\"name\":\"__dbz_ts_ms\",\"type\":[\"null\",\"long\"],\"default\":null},{\"name\":\"__dbz_source_ts_ms\",\"type\":[\"null\",\"long\"],\"default\":null}],\"connect.name\":\"dbz_avro.foobar.Value\"}"
},