连接到现有的 kafka 代理时，ksqlDB 服务器无法启动
ksqlDB server fails to start when connecting to existing kafka broker
我在 :9092 有一个现有的 kafka 代理。该代理是 运行 Apache Kafka 2.2.0 版。我想使用 ksqlDB 对来自该 kafka 代理上的主题的数据进行一些流处理。因此，根据 https://docs.confluent.io/platform/current/installation/versions-interoperability.html.
的兼容性 table，我正在使用 Confluent Platform 5.2 版的 ksqlDB
我在 ksql/ksql-server.properties 中设置了 bootstrap.servers=<myurl>:9092。
但是，当我尝试通过 运行 ksql-server-start etc/ksql/ksql-server.properties 启动 ksql-server 时，出现以下错误：
ERROR Failed to start KSQL (io.confluent.ksql.rest.server.KsqlServerMain:53)
org.apache.kafka.common.errors.SerializationException: Error deserializing key/value for partition _confluent-ksql-default__command_topic-0 at offset 0. If needed, please seek past the record to continue consumption.
Caused by: org.apache.kafka.common.errors.SerializationException: com.fasterxml.jackson.databind.exc.InvalidDefinitionException: Cannot construct instance of `io.confluent.ksql.rest.server.computation.Command`, problem: `java.lang.NullPointerException`
at [Source: (byte[])"("statement":"CREATE STREAM KSQL_PROCESSING_LOG (logger VARCHAR, level VARCHAR, time BIGINT, message STRUCT<type INT, deserializationError STRUCT<errorMessage VARCHAR, recordB64 VARCHAR, cause ARRAY<VARCHAR>, `topic` VARCHAR>, recordProcessingError STRUCT<errorMessage VARCHAR, record VARCHAR, cause ARRAY<VARCHAR>>, productionError STRUCT<errorMessage VARCHAR>>) WITH(KAFKA_TOPIC='default_ksql_processing_log', VALUE_FORMAT='JSON');","originalProperties":{"ksql.extension.dir":"ext","ksql.streams.ca"[truncated 3011 bytes]; line: 1, column: 3511]
Caused by: com.fasterxml.jackson.databind.exc.InvalidDefinitionException: Cannot construct instance of `io.confluent.ksql.rest.server.computation.Command`, problem: `java.lang.NullPointerException`
at [Source: (byte[])"("statement":"CREATE STREAM KSQL_PROCESSING_LOG (logger VARCHAR, level VARCHAR, time BIGINT, message STRUCT<type INT, deserializationError STRUCT<errorMessage VARCHAR, recordB64 VARCHAR, cause ARRAY<VARCHAR>, `topic` VARCHAR>, recordProcessingError STRUCT<errorMessage VARCHAR, record VARCHAR, cause ARRAY<VARCHAR>>, productionError STRUCT<errorMessage VARCHAR>>) WITH(KAFKA_TOPIC='default_ksql_processing_log', VALUE_FORMAT='JSON');","originalProperties":{"ksql.extension.dir":"ext","ksql.streams.ca"[truncated 3011 bytes]; line: 1, column: 3511]
...
如果我使用本地代理并设置 bootstrap.servers=localhost:9092，ksql-server 启动没有任何问题。
如何解决这个 null record/deserialization 问题，以便将 ksqldb 服务器连接到我现有的 kafka 代理？
正如@OneCricketeer 所指出的，问题是由集群上现有的命令主题引起的。您可以通过更改 ksql.service.id in the ksqlDB server properties. See Configuring ksqlDB server.
使用新的命令主题
我在
我在 ksql/ksql-server.properties 中设置了 bootstrap.servers=<myurl>:9092。
但是，当我尝试通过 运行 ksql-server-start etc/ksql/ksql-server.properties 启动 ksql-server 时，出现以下错误：
ERROR Failed to start KSQL (io.confluent.ksql.rest.server.KsqlServerMain:53)
org.apache.kafka.common.errors.SerializationException: Error deserializing key/value for partition _confluent-ksql-default__command_topic-0 at offset 0. If needed, please seek past the record to continue consumption.
Caused by: org.apache.kafka.common.errors.SerializationException: com.fasterxml.jackson.databind.exc.InvalidDefinitionException: Cannot construct instance of `io.confluent.ksql.rest.server.computation.Command`, problem: `java.lang.NullPointerException`
at [Source: (byte[])"("statement":"CREATE STREAM KSQL_PROCESSING_LOG (logger VARCHAR, level VARCHAR, time BIGINT, message STRUCT<type INT, deserializationError STRUCT<errorMessage VARCHAR, recordB64 VARCHAR, cause ARRAY<VARCHAR>, `topic` VARCHAR>, recordProcessingError STRUCT<errorMessage VARCHAR, record VARCHAR, cause ARRAY<VARCHAR>>, productionError STRUCT<errorMessage VARCHAR>>) WITH(KAFKA_TOPIC='default_ksql_processing_log', VALUE_FORMAT='JSON');","originalProperties":{"ksql.extension.dir":"ext","ksql.streams.ca"[truncated 3011 bytes]; line: 1, column: 3511]
Caused by: com.fasterxml.jackson.databind.exc.InvalidDefinitionException: Cannot construct instance of `io.confluent.ksql.rest.server.computation.Command`, problem: `java.lang.NullPointerException`
at [Source: (byte[])"("statement":"CREATE STREAM KSQL_PROCESSING_LOG (logger VARCHAR, level VARCHAR, time BIGINT, message STRUCT<type INT, deserializationError STRUCT<errorMessage VARCHAR, recordB64 VARCHAR, cause ARRAY<VARCHAR>, `topic` VARCHAR>, recordProcessingError STRUCT<errorMessage VARCHAR, record VARCHAR, cause ARRAY<VARCHAR>>, productionError STRUCT<errorMessage VARCHAR>>) WITH(KAFKA_TOPIC='default_ksql_processing_log', VALUE_FORMAT='JSON');","originalProperties":{"ksql.extension.dir":"ext","ksql.streams.ca"[truncated 3011 bytes]; line: 1, column: 3511]
...
如果我使用本地代理并设置 bootstrap.servers=localhost:9092，ksql-server 启动没有任何问题。
如何解决这个 null record/deserialization 问题，以便将 ksqldb 服务器连接到我现有的 kafka 代理？
正如@OneCricketeer 所指出的，问题是由集群上现有的命令主题引起的。您可以通过更改 ksql.service.id in the ksqlDB server properties. See Configuring ksqlDB server.