Debezium Connector for Oracle does not read its configuration options

Hey people!

I am here trying to attach a Debezium Source Connector to my Oracle database. And I end up with this exception, when try to deploy my connector configuration to Kafka Connect:

Unable to connect: Failed to resolve Oracle database version

The exception looks really weird to me, because other connectors work with the same database perfectly fine (though, they all are sink connectors, not source, but I don’t think that this should make any difference).

I am providing my connector config here under a spoiler. I used the official docs very carefully to compose it, so I don’t really understand what could be wrong about it.

Connector config
{
    "name":"source-local-debezium-revision-1",
    "config":{
        "connector.class":"io.debezium.connector.oracle.OracleConnector",
        "key.converter":"io.confluent.connect.avro.AvroConverter",
        "key.converter.schema.registry.url":"http://localhost:8081",
        "key.converter.schemas.enable":"true",
        "value.converter":"io.confluent.connect.avro.AvroConverter",
        "value.converter.schema.registry.url":"http://localhost:8081",
        "value.converter.schemas.enable":"true",
        "tasks.max":"1",
        "topics":"my-dest-topic",
        "database.hostname":"localhost",
        "database.port":"1521",
        "database.user":"c##debezium_user",
        "database.password":"top_secret",
        "database.dbname":"XE",
        "database.pdb.name":"XEPDB1",
        "database.server.name":"local_sandbox_oracle",
        "database.connection.adapter":"logminer",
        "database.history.kafka.topic":"schema_changes",
        "database.history.kafka.bootstrap.servers":"localhost:9092",
        "database.tablename.case.insensitive":"true",
        "snapshot.mode":"initial",
        "table.include.list":"users.my_source_table",
        "tombstones.on.delete":"true",
        "message.key.columns":"users.my_source_table.id",
        "sanitize.field.names":"true"
    }
}

Furthermore, when I tried to validate the config.

Validation query
curl --location --request PUT 'http://localhost:8083/connector-plugins/io.debezium.connector.oracle.OracleConnector/config/validate' \
--header 'Content-Type: application/json' \
--data-raw '{
    "name": "source-local-debezium-revision-1",
    "connector.class": "io.debezium.connector.oracle.OracleConnector",
    "key.converter": "io.confluent.connect.avro.AvroConverter",
    "key.converter.schema.registry.url": "http://localhost:8081",
    "key.converter.schemas.enable": "true",
    "value.converter": "io.confluent.connect.avro.AvroConverter",
    "value.converter.schema.registry.url": "http://localhost:8081",
    "value.converter.schemas.enable": "true",
    "tasks.max": "1",
    "topics": "my-dest-topic",
    "database.hostname": "localhost",
    "database.port": "1521",
    "database.user": "c##debezium_user",
    "database.password": "top_secret",
    "database.dbname": "XE",
    "database.pdb.name": "XEPDB1",
    "database.server.name": "local_sandbox_oracle",
    "database.connection.adapter": "logminer",
    "database.history.kafka.topic": "schema_changes",
    "database.history.kafka.bootstrap.servers": "localhost:9092",
    "database.tablename.case.insensitive": "true",
    "snapshot.mode": "initial",
    "table.include.list": "users.my_source_table",
    "tombstones.on.delete": "true",
    "message.key.columns": "users.my_source_table:id",
    "sanitize.field.names": "true"
}'

I got this very weird result.

Validation result
{
    "name": "io.debezium.connector.oracle.OracleConnector",
    "error_count": 1,
    "groups": [
        "Common",
        "Transforms",
        "Predicates",
        "Error Handling",
        "Topic Creation",
        "Oracle",
        "Connector",
        "History Storage",
        "Events"
    ],
    "configs": [
        {
            "definition": {
                "name": "name",
                "type": "STRING",
                "required": true,
                "default_value": null,
                "importance": "HIGH",
                "documentation": "Globally unique name to use for this connector.",
                "group": "Common",
                "width": "MEDIUM",
                "display_name": "Connector name",
                "dependents": [],
                "order": 1
            },
            "value": {
                "name": "name",
                "value": "source-local-debezium-revision-1",
                "recommended_values": [],
                "errors": [],
                "visible": true
            }
        },
        {
            "definition": {
                "name": "connector.class",
                "type": "STRING",
                "required": true,
                "default_value": null,
                "importance": "HIGH",
                "documentation": "Name or alias of the class for this connector. Must be a subclass of org.apache.kafka.connect.connector.Connector. If the connector is org.apache.kafka.connect.file.FileStreamSinkConnector, you can either specify this full name,  or use \"FileStreamSink\" or \"FileStreamSinkConnector\" to make the configuration a bit shorter",
                "group": "Common",
                "width": "LONG",
                "display_name": "Connector class",
                "dependents": [],
                "order": 2
            },
            "value": {
                "name": "connector.class",
                "value": "io.debezium.connector.oracle.OracleConnector",
                "recommended_values": [],
                "errors": [],
                "visible": true
            }
        },
        {
            "definition": {
                "name": "tasks.max",
                "type": "INT",
                "required": false,
                "default_value": "1",
                "importance": "HIGH",
                "documentation": "Maximum number of tasks to use for this connector.",
                "group": "Common",
                "width": "SHORT",
                "display_name": "Tasks max",
                "dependents": [],
                "order": 3
            },
            "value": {
                "name": "tasks.max",
                "value": "1",
                "recommended_values": [],
                "errors": [],
                "visible": true
            }
        },
        {
            "definition": {
                "name": "key.converter",
                "type": "CLASS",
                "required": false,
                "default_value": null,
                "importance": "LOW",
                "documentation": "Converter class used to convert between Kafka Connect format and the serialized form that is written to Kafka. This controls the format of the keys in messages written to or read from Kafka, and since this is independent of connectors it allows any connector to work with any serialization format. Examples of common formats include JSON and Avro.",
                "group": "Common",
                "width": "SHORT",
                "display_name": "Key converter class",
                "dependents": [],
                "order": 4
            },
            "value": {
                "name": "key.converter",
                "value": "io.confluent.connect.avro.AvroConverter",
                "recommended_values": [],
                "errors": [],
                "visible": true
            }
        },
        {
            "definition": {
                "name": "value.converter",
                "type": "CLASS",
                "required": false,
                "default_value": null,
                "importance": "LOW",
                "documentation": "Converter class used to convert between Kafka Connect format and the serialized form that is written to Kafka. This controls the format of the values in messages written to or read from Kafka, and since this is independent of connectors it allows any connector to work with any serialization format. Examples of common formats include JSON and Avro.",
                "group": "Common",
                "width": "SHORT",
                "display_name": "Value converter class",
                "dependents": [],
                "order": 5
            },
            "value": {
                "name": "value.converter",
                "value": "io.confluent.connect.avro.AvroConverter",
                "recommended_values": [],
                "errors": [],
                "visible": true
            }
        },
        {
            "definition": {
                "name": "header.converter",
                "type": "CLASS",
                "required": false,
                "default_value": null,
                "importance": "LOW",
                "documentation": "HeaderConverter class used to convert between Kafka Connect format and the serialized form that is written to Kafka. This controls the format of the header values in messages written to or read from Kafka, and since this is independent of connectors it allows any connector to work with any serialization format. Examples of common formats include JSON and Avro. By default, the SimpleHeaderConverter is used to serialize header values to strings and deserialize them by inferring the schemas.",
                "group": "Common",
                "width": "SHORT",
                "display_name": "Header converter class",
                "dependents": [],
                "order": 6
            },
            "value": {
                "name": "header.converter",
                "value": null,
                "recommended_values": [],
                "errors": [],
                "visible": true
            }
        },
        {
            "definition": {
                "name": "transforms",
                "type": "LIST",
                "required": false,
                "default_value": "",
                "importance": "LOW",
                "documentation": "Aliases for the transformations to be applied to records.",
                "group": "Transforms",
                "width": "LONG",
                "display_name": "Transforms",
                "dependents": [],
                "order": 7
            },
            "value": {
                "name": "transforms",
                "value": "",
                "recommended_values": [],
                "errors": [],
                "visible": true
            }
        },
... some deleted rows due to max message size ...
        {
            "definition": {
                "name": "database.hostname",
                "type": "STRING",
                "required": false,
                "default_value": null,
                "importance": "HIGH",
                "documentation": "Resolvable hostname or IP address of the database server.",
                "group": "Oracle",
                "width": "MEDIUM",
                "display_name": "Hostname",
                "dependents": [],
                "order": 1
            },
            "value": {
                "name": "database.hostname",
                "value": null,
                "recommended_values": [],
                "errors": [
                    "Unable to connect: Failed to resolve Oracle database version"
                ],
                "visible": true
            }
        },
        {
            "definition": {
                "name": "database.port",
                "type": "INT",
                "required": false,
                "default_value": "1528",
                "importance": "HIGH",
                "documentation": "Port of the database server.",
                "group": "Oracle",
                "width": "SHORT",
                "display_name": "Port",
                "dependents": [],
                "order": 2
            },
            "value": {
                "name": "database.port",
                "value": null,
                "recommended_values": [],
                "errors": [],
                "visible": true
            }
        },
        {
            "definition": {
                "name": "database.user",
                "type": "STRING",
                "required": false,
                "default_value": null,
                "importance": "HIGH",
                "documentation": "Name of the database user to be used when connecting to the database.",
                "group": "Oracle",
                "width": "SHORT",
                "display_name": "User",
                "dependents": [],
                "order": 3
            },
            "value": {
                "name": "database.user",
                "value": null,
                "recommended_values": [],
                "errors": [],
                "visible": true
            }
        },
... some deleted rows due to max message size ...
        {
            "definition": {
                "name": "table.whitelist",
                "type": "LIST",
                "required": false,
                "default_value": null,
                "importance": "LOW",
                "documentation": "The tables for which changes are to be captured (deprecated, use \"table.include.list\" instead)",
                "group": "Events",
                "width": "LONG",
                "display_name": "Deprecated: Include Tables",
                "dependents": [],
                "order": 12
            },
            "value": {
                "name": "table.whitelist",
                "value": null,
                "recommended_values": [],
                "errors": [],
                "visible": false
            }
        },
        {
            "definition": {
                "name": "table.include.list",
                "type": "LIST",
                "required": false,
                "default_value": null,
                "importance": "HIGH",
                "documentation": "The tables for which changes are to be captured",
                "group": "Events",
                "width": "LONG",
                "display_name": "Include Tables",
                "dependents": [],
                "order": 13
            },
            "value": {
                "name": "table.include.list",
                "value": null,
                "recommended_values": [],
                "errors": [],
                "visible": true
            }
        },
        {
            "definition": {
                "name": "table.blacklist",
                "type": "LIST",
                "required": false,
                "default_value": null,
                "importance": "LOW",
                "documentation": "A comma-separated list of regular expressions that match the fully-qualified names of tables to be excluded from monitoring (deprecated, use \"table.exclude.list\" instead)",
                "group": "Events",
                "width": "LONG",
                "display_name": "Deprecated: Exclude Tables",
                "dependents": [],
                "order": 14
            },
            "value": {
                "name": "table.blacklist",
                "value": null,
                "recommended_values": [],
                "errors": [],
                "visible": false
            }
        },
... some deleted rows due to max message size ...
    ]
}

The most interesting part in it is the part, which shows what causes the issue:

{
    "definition": {
        "name": "database.hostname",
        "type": "STRING",
        "required": false,
        "default_value": null,
        "importance": "HIGH",
        "documentation": "Resolvable hostname or IP address of the database server.",
        "group": "Oracle",
        "width": "MEDIUM",
        "display_name": "Hostname",
        "dependents": [],
        "order": 1
    },
    "value": {
        "name": "database.hostname",
        "value": null,
        "recommended_values": [],
        "errors": [
            "Unable to connect: Failed to resolve Oracle database version"
        ],
        "visible": true
    }
}

And as you see, apparently, the value of the database.hostname configuration option is null, when it was perfectly fine provided in the config: "database.hostname":"localhost".

If you look deeper, you will also see that some other configuration options did not get their values as well (i.e. database.port, table.include.list, …), whereas some different ones were filled correctly (i.e. connector.class or key.converter).

I already tried different ordering of the configuration options (put ones which are now not filled in the beginning of the config file), tested this with Debezium 1.5.0 Final, 1.5.1 Final and 1.6.0 Beta. The issue persists in all these versions.

Does anybody know what could be the root cause for the issue? Or am I missing something?

Or should I address my question to the Debezium team instead?

Thank you!

Any ideas, guys?
Did it work for anybody with Oracle? Can you share your configuration probably?

@whatsupbros can you verify that the DB is reachable on ‘localhost:1521’? By chance are you running in Docker which might affect the availability of localhost ?

Hi, @rick!

The database is indeed reachable on localhost:1521 - as I described in the original topic, there are other connectors (JDBC Sink), which are RUNNING perfectly fine, and they are connected to the very same database. This means that the db is not just reachable, but 100% reachable from the Kafka Connect Cluster.

There is no Docker installed on my machine. Both Oracle db and Kafka operate locally on localhost.

And, as I also demonstrated above, it really seems that the issue is not because the db host localhost is unreachable, but because the connector didn’t read the database.host parameter value, and it tries to use null as the database host, which is of course unreachable for the connector.

The question is why it doesn’t see the provided properties values?

So, no ideas, colleagues?

Should I address my question to the Debezium team probably? I am just not sure where the problem is - on the Confluent Platform, or the Debezium Connector itself.

Thanks!

This topic was automatically closed 30 days after the last reply. New replies are no longer allowed.