[avro] Support record unions as JSON (#800)

CHANGELOG.md

@@ -20,6 +20,14 @@
   `union_null_string_int` schema from the incompatible to the compatible Avro test fixtures and adds
   feature-test coverage. (https://github.com/ClickHouse/clickhouse-kafka-connect/issues/799)
 
+* Support Avro record unions (e.g. `[TypeA, TypeB]`) mapping to a ClickHouse `JSON` column. Confluent's
+  Avro converter turns a union of records into a Connect union struct (`io.confluent.connect.avro.Union`)
+  keyed by branch record name; the connector serializes this to JSON in tagged form (which branch it was
+  is preserved). Writing to a `JSON` column in RowBinary requires `input_format_binary_read_json_as_string=1`,
+  so the Avro integration harness now routes JSON-target fixtures through a `setupAvroConnectorWithJson`
+  connector config. This promotes the `union_two_records` schema from the incompatible to the compatible
+  Avro test fixtures and adds feature-test coverage. (https://github.com/ClickHouse/clickhouse-kafka-connect/issues/800)
+
 ## Bug Fixes 
 
 * Fixed issue with Table Schema cache that can remember old schema instead of new one. Now if old 

src/integrationTest/java/com/clickhouse/kafka/connect/sink/ClickHouseSinkConnectorIntegrationTest.java

@@ -230,6 +230,14 @@ private void setupAvroConnector(String topicName) throws IOException, Interrupte
         confluentPlatform.createConnectorAndWaitUntilRunning(SINK_CONNECTOR_NAME, SinkConfigs.AVRO.getJsonPayload(1, topicName));
     }
 
+    // Same as setupAvroConnector but with input_format_binary_read_json_as_string=1, required for
+    // fixtures whose target table has a JSON column (RowBinary writes the JSON value as a string).
+    private void setupAvroConnectorWithJson(String topicName) throws IOException, InterruptedException {
+        LOGGER.info("Setting up Avro connector (JSON as string) for topic {}...", topicName);
+        confluentPlatform.deleteConnectors(SINK_CONNECTOR_NAME);
+        confluentPlatform.createConnectorAndWaitUntilRunning(SINK_CONNECTOR_NAME, SinkConfigs.AVRO_JSON.getJsonPayload(1, topicName));
+    }
+
     private void setupProtobufConnector(String topicName) throws IOException, InterruptedException {
         LOGGER.info("Setting up Protobuf connector for topic {}...", topicName);
         confluentPlatform.deleteConnectors(SINK_CONNECTOR_NAME);
@@ -323,8 +331,11 @@ public void avroSchemaTest(Path schemaPath) throws Exception {
                 .engine("MergeTree")
                 .orderByColumn(fixture.getString(clickhouseOrderByKey));
         JSONObject clickhouseColumns = fixture.getJSONObject(clickhouseColumnsKey);
+        boolean hasJsonColumn = false;
         for (String colName : clickhouseColumns.keySet()) {
-            tableStmt.column(colName, clickhouseColumns.getString(colName));
+            String colType = clickhouseColumns.getString(colName);
+            tableStmt.column(colName, colType);
+            hasJsonColumn |= colType.contains("JSON");
         }
         tableStmt.execute(chcNoProxy);
 
@@ -336,8 +347,12 @@ public void avroSchemaTest(Path schemaPath) throws Exception {
         );
         LOGGER.info("Produced {} records to topic {}", producedCount, topicName);
 
-        // 4. Setup sink connector with Avro converter
-        setupAvroConnector(topicName);
+        // 4. Setup sink connector with Avro converter (JSON-as-string variant when a JSON column is present)
+        if (hasJsonColumn) {
+            setupAvroConnectorWithJson(topicName);
+        } else {
+            setupAvroConnector(topicName);
+        }
 
         // 5. Wait for data to flow through
         ClickHouseTestHelpers.waitWhileCounting(chcNoProxy, topicName, 3);
@@ -440,6 +455,7 @@ private static enum SinkConfigs {
         BASE_SCHEMALESS("clickhouse_sink_schemaless.json"),
         PROTOBUF("clickhouse_sink_protobuf.json"),
         AVRO("clickhouse_sink_avro.json"),
+        AVRO_JSON("clickhouse_sink_avro_json.json"),
         JDBC_PROP("clickhouse_sink_with_jdbc_prop.json");
 
         final Path pathToJsonConfig;

src/integrationTest/resources/clickhouse_sink_avro_json.json

@@ -0,0 +1,20 @@
+{
+  "name": "%s",
+  "config": {
+    "name": "%s",
+    "connector.class": "com.clickhouse.kafka.connect.ClickHouseSinkConnector",
+    "tasks.max": "%d",
+    "topics": "%s",
+    "hostname": "%s",
+    "port": "%s",
+    "database": "default",
+    "username": "%s",
+    "password": "%s",
+    "ssl": "false",
+    "exactlyOnce": "false",
+    "value.converter": "io.confluent.connect.avro.AvroConverter",
+    "value.converter.schema.registry.url": "http://schema-registry:8081",
+    "client_version": "V2",
+    "clickhouseSettings": "input_format_binary_read_json_as_string=1"
+  }
+}

src/test/java/com/clickhouse/kafka/connect/sink/db/AvroRecordUnionJsonTest.java

@@ -0,0 +1,83 @@
+package com.clickhouse.kafka.connect.sink.db;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import com.clickhouse.kafka.connect.sink.data.Data;
+import com.clickhouse.kafka.connect.sink.data.StructToJsonMap;
+import com.clickhouse.kafka.connect.util.DataJson;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import java.util.Map;
+import org.apache.kafka.connect.data.Schema;
+import org.apache.kafka.connect.data.SchemaBuilder;
+import org.apache.kafka.connect.data.Struct;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Verifies how a record union {@code [TypeA, TypeB]} serializes to JSON for a ClickHouse {@code JSON}
+ * column (issue #800, from #726). Confluent's Avro converter turns a union of two records into a
+ * Connect struct named {@code io.confluent.connect.avro.Union} with one optional struct field per
+ * branch (keyed by the branch record name). Per the maintainer, the value is kept in this tagged
+ * form (which branch it was is preserved); ClickHouse's {@code JSON} type stores it as-is.
+ *
+ * <p>Asserted by parsing the output rather than string-matching, since field order in the tag object
+ * is not significant.
+ */
+public class AvroRecordUnionJsonTest {
+
+  private static final String AVRO_UNION_SCHEMA_NAME = "io.confluent.connect.avro.Union";
+  private static final ObjectMapper JSON = new ObjectMapper();
+
+  private static Schema recordUnionSchema() {
+    Schema typeA =
+        SchemaBuilder.struct().name("TypeA").field("label", Schema.STRING_SCHEMA).optional().build();
+    Schema typeB =
+        SchemaBuilder.struct().name("TypeB").field("count", Schema.INT32_SCHEMA).optional().build();
+    Schema union =
+        SchemaBuilder.struct()
+            .name(AVRO_UNION_SCHEMA_NAME)
+            .field("TypeA", typeA)
+            .field("TypeB", typeB)
+            .optional()
+            .build();
+    return SchemaBuilder.struct().field("id", Schema.INT32_SCHEMA).field("payload", union).build();
+  }
+
+  private static JsonNode serializePayload(Struct record) throws Exception {
+    Map<String, Data> data = StructToJsonMap.toJsonMap(record);
+    return JSON.readTree(DataJson.OBJECT_MAPPER.writeValueAsBytes(data.get("payload")));
+  }
+
+  @Test
+  public void recordUnion_typeABranch_serializesTaggedByBranchName() throws Exception {
+    Schema record = recordUnionSchema();
+    Schema union = record.field("payload").schema();
+    Schema typeA = union.field("TypeA").schema();
+    Struct value =
+        new Struct(record)
+            .put("id", 1)
+            .put("payload", new Struct(union).put("TypeA", new Struct(typeA).put("label", "foo")));
+
+    JsonNode json = serializePayload(value);
+
+    assertTrue(json.has("TypeA"), "payload should be tagged by branch name: " + json);
+    assertEquals("foo", json.get("TypeA").get("label").asText());
+  }
+
+  @Test
+  public void recordUnion_typeBBranch_serializesTaggedByBranchName() throws Exception {
+    Schema record = recordUnionSchema();
+    Schema union = record.field("payload").schema();
+    Schema typeB = union.field("TypeB").schema();
+    Struct value =
+        new Struct(record)
+            .put("id", 2)
+            .put("payload", new Struct(union).put("TypeB", new Struct(typeB).put("count", 42)));
+
+    JsonNode json = serializePayload(value);
+
+    assertTrue(json.has("TypeB"), "payload should be tagged by branch name: " + json);
+    assertEquals(42, json.get("TypeB").get("count").asInt());
+  }
+}

src/testFixtures/avro/schemas/compatible/union_two_records.json

@@ -0,0 +1,95 @@
+{
+  "description": "Union: union of two different record types [recordA, recordB] \u2192 JSON",
+  "schema": {
+    "type": "record",
+    "name": "TwoRecords",
+    "namespace": "com.clickhouse.kafka.connect.avro.test",
+    "fields": [
+      {
+        "name": "id",
+        "type": "int"
+      },
+      {
+        "name": "payload",
+        "type": [
+          {
+            "type": "record",
+            "name": "TypeA",
+            "fields": [
+              {
+                "name": "label",
+                "type": "string"
+              }
+            ]
+          },
+          {
+            "type": "record",
+            "name": "TypeB",
+            "fields": [
+              {
+                "name": "count",
+                "type": "int"
+              }
+            ]
+          }
+        ]
+      }
+    ]
+  },
+  "clickhouse_columns": {
+    "id": "Int32",
+    "payload": "JSON"
+  },
+  "clickhouse_order_by": "id",
+  "records": [
+    {
+      "id": 1,
+      "payload": {
+        "com.clickhouse.kafka.connect.avro.test.TypeA": {
+          "label": "foo"
+        }
+      }
+    },
+    {
+      "id": 3,
+      "payload": {
+        "com.clickhouse.kafka.connect.avro.test.TypeB": {
+          "count": 42
+        }
+      }
+    },
+    {
+      "id": 5,
+      "payload": {
+        "com.clickhouse.kafka.connect.avro.test.TypeA": {
+          "label": "bar"
+        }
+      }
+    },
+    {
+      "id": 6,
+      "payload": {
+        "com.clickhouse.kafka.connect.avro.test.TypeB": {
+          "count": 99
+        }
+      }
+    },
+    {
+      "id": 8,
+      "payload": {
+        "com.clickhouse.kafka.connect.avro.test.TypeA": {
+          "label": "baz"
+        }
+      }
+    },
+    {
+      "id": 9,
+      "payload": {
+        "com.clickhouse.kafka.connect.avro.test.TypeB": {
+          "count": 7
+        }
+      }
+    }
+  ],
+  "expected_row_count": 6
+}