Search code examples
apache-kafkavertica

Vertica-Kafka vkconfig connector

I'm trying to define a Vertica-Kafka scheduler. I ran the first few commands successfully, but failed on the following command:

$ /opt/vertica/packages/kafka/bin/vkconfig source --create --cluster kafka_nms_cluster --source test --partitions 1 --conf /home/vertica/vkconfig/vkconfig.conf

The error I got

Exception in thread "main" com.vertica.solutions.kafka.exception.ConfigurationException: ERROR: [[Vertica][VJDBC](5861) ERROR: Error calling processPartition() in User Function KafkaListTopics at [/data/qb_workspaces/jenkins2/ReleaseBuilds/Grader/REL-9_2_1-x_grader/build/udx/supported/kafka/KafkaUtil.cpp:163], error code: 0, message: Error getting metadata: [Local: Broker transport failure]]
        at com.vertica.solutions.kafka.model.StreamSource.validateConfiguration(StreamSource.java:248)
        at com.vertica.solutions.kafka.model.StreamSource.setFromMapAndValidate(StreamSource.java:194)
        at com.vertica.solutions.kafka.model.StreamModel.<init>(StreamModel.java:93)
        at com.vertica.solutions.kafka.model.StreamSource.<init>(StreamSource.java:44)
        at com.vertica.solutions.kafka.cli.SourceCLI.getNewModel(SourceCLI.java:62)
        at com.vertica.solutions.kafka.cli.SourceCLI.getNewModel(SourceCLI.java:13)
        at com.vertica.solutions.kafka.cli.CLI.run(CLI.java:59)
        at com.vertica.solutions.kafka.cli.CLI._main(CLI.java:141)
        at com.vertica.solutions.kafka.cli.SourceCLI.main(SourceCLI.java:29)
Caused by: java.sql.SQLNonTransientException: [Vertica][VJDBC](5861) ERROR: Error calling processPartition() in User Function KafkaListTopics at [/data/qb_workspaces/jenkins2/ReleaseBuilds/Grader/REL-9_2_1-x_grader/build/udx/supported/kafka/KafkaUtil.cpp:163], error code: 0, message: Error getting metadata: [Local: Broker transport failure]
        at com.vertica.util.ServerErrorData.buildException(Unknown Source)
        at com.vertica.dataengine.VResultSet.fetchChunk(Unknown Source)
        at com.vertica.dataengine.VResultSet.initialize(Unknown Source)
        at com.vertica.dataengine.VQueryExecutor.readExecuteResponse(Unknown Source)
        at com.vertica.dataengine.VQueryExecutor.handleExecuteResponse(Unknown Source)
        at com.vertica.dataengine.VQueryExecutor.execute(Unknown Source)
        at com.vertica.jdbc.common.SPreparedStatement.executeWithParams(Unknown Source)
        at com.vertica.jdbc.common.SPreparedStatement.executeQuery(Unknown Source)
        at com.vertica.solutions.kafka.model.StreamSource.validateConfiguration(StreamSource.java:227)
        ... 8 more
Caused by: com.vertica.support.exceptions.NonTransientException: [Vertica][VJDBC](5861) ERROR: Error calling processPartition() in User Function KafkaListTopics at [/data/qb_workspaces/jenkins2/ReleaseBuilds/Grader/REL-9_2_1-x_grader/build/udx/supported/kafka/KafkaUtil.cpp:163], error code: 0, message: Error getting metadata: [Local: Broker transport failure]
        ... 17 more

However, when I try to run KafkaListTopics using vsql, the resultset shows the test topic with 1 partition.

[root@dal_server1 ~]# /opt/vertica/bin/vsql -U vertica -c "SELECT KafkaListTopics(USING PARAMETERS brokers='10.22.2.38:9092') OVER ();"

       topic        | num_partitions
--------------------+----------------
 __consumer_offsets |             50
 test               |              1
 TutorialTopic      |              1
(3 rows)

What might be causing this error?

Thanks Avi

Solution

  • The issue may actually be with the cluster you created before you attempted to create the source. I have had this same issue when testing the Vertica/Kafka integration where the test Kafka cluster does not have a DNS entry, but a DNS name is stored in the stream_clsuters table.

    Query the <scheduler_config_schema>.stream_clusters table. If a DNS name is stored instead of just a plain IP address, then you can do two things.

    1. Do a manual update on the stream_clusters table to change it to <ip_address>:<port> if there is only one Kafka node, or<ip_address1>:<port>,...,<ip_addressN>:<port> if there are multiple.
    2. Or, add the domain name to your /etc/hosts on all Vertica nodes

    For example, in the stream_clusters table, you see domain_name_1:9092, run this UPDATE statement:

    UPDATE <scheduler_config_schema>.stream_clusters
SET hosts = '10.22.2.38:9092'
WHERE id = <some_id>

    Normally, I would advice NOT doing any kind of manual DML on these scheduler configuration tables, but I have done this specific update before and it is safe (especially in testing).

    Of course in a true production environment the Kafka clusters should have DNS entries in your network, and you will not have to worry about this error, but for testing with VMs or Docker containers, I've come across this several times and the suggestions above have done the trick.