DBZ-7554 AvroConverter fails to parse events

[schampilomatis]

…columns

Adjust schema names for cells to be unique. Avro uses these names to identify sub schema declarations and will not redefine a sub schema that was already detected. In this case, if we have cells with multiple types, debezium will fail to serialize the event, since the stored avro schema does not match the cells.

[github-actions]

Welcome as a new contributor to Debezium, @schampilomatis. Reviewers, please add missing author name(s) and alias name(s) to the COPYRIGHT.txt and Aliases.txt respectively.

[schampilomatis]

Sharing an example schema produced for
CREATE TABLE IF NOT EXISTS test_ks.table_cdc (k text PRIMARY KEY, t text, i int) WITH cdc=true;

"fields": [
          {
            "name": "k",
            "type": [
              "null",
              {
                "type": "record",
                "name": "cell_value",
                "fields": [
                  {
                    "name": "value",
                    "type": [
                      "null",
                      "string"
                    ],
                    "default": null
                  },
                  {
                    "name": "deletion_ts",
                    "type": [
                      "null",
                      "long"
                    ],
                    "default": null
                  },
                  {
                    "name": "set",
                    "type": "boolean"
                  }
                ],
                "connect.version": 1,
                "connect.name": "cell_value"
              }
            ],
            "default": null
          },
          {
            "name": "i",
            "type": [
              "null",
              "cell_value"
            ],
            "default": null
          },
          {
            "name": "t",
            "type": [
              "null",
              "cell_value"
            ],
            "default": null
          },
          {
            "name": "_range_start",
            "type": [
              "null",
              {
                "type": "record",
                "name": "_range_start",
                "fields": [
                  {
                    "name": "method",
                    "type": "string"
                  },
                  {
                    "name": "values",
                    "type": {
                      "type": "array",
                      "items": {
                        "type": "record",
                        "name": "clustering_value",
                        "fields": [
                          {
                            "name": "name",
                            "type": "string"
                          },
                          {
                            "name": "value",
                            "type": "string"
                          },
                          {
                            "name": "type",
                            "type": "string"
                          }
                        ],
                        "connect.version": 1,
                        "connect.name": "clustering_value"
                      },
                      "connect.version": 1,
                      "connect.name": "clustering_values"
                    }
                  }
                ],
                "connect.version": 1,
                "connect.name": "_range_start"
              }
            ],
            "default": null
          },
          {
            "name": "_range_end",
            "type": [
              "null",
              {
                "type": "record",
                "name": "_range_end",
                "fields": [
                  {
                    "name": "method",
                    "type": "string"
                  },
                  {
                    "name": "values",
                    "type": {
                      "type": "array",
                      "items": "clustering_value",
                      "connect.version": 1,
                      "connect.name": "clustering_values"
                    }
                  }
                ],
                "connect.version": 1,
                "connect.name": "_range_end"
              }
            ],
            "default": null
          }
        ],

We see here that all of value for k, t, i are of type [null, "cell_value"], with "cell_value" being a named schema of "string". By setting the name of the schemas to something unique (column name) we avoid this.

This is a partial revert of commit:

ec9e889a

[github-actions]

Hi @schampilomatis, thanks for your contribution. Please prefix the commit message(s) with the DBZ-xxx JIRA issue key.

[schampilomatis]

Duplicate of #121