apache-kafka - Confluent Cloud -> BigQuery - 如何诊断“不良记录”原因

Question

我可以将数据从 MSSql Server 推送到 Confluent Cloud 上的主题，但不能从主题推送到 BigQuery，它会引发错误“过去一小时的错误记录 - 65”

我可以将主题连接到 bigQuery，但无法摄取数据。

MSSQL 和 BigQuery 表格式相同 first(string) last(string) raj ram

我是否需要添加任何其他列来提取时间戳、偏移量等数据？

Answer 1

如果有消息无法发送到目标，它们将被写入死信队列，其中包含问题的详细信息。

在连接器屏幕中，您可以看到连接器的 ID

使用该 id 来定位具有相同名称和dlq-前缀的主题。

然后您可以浏览主题并使用标题信息来确定问题的原因

如果您愿意，可以使用 kafkacat 查看标题：

$ docker run --rm edenhill/kafkacat:1.5.0 \
         -X security.protocol=SASL_SSL -X sasl.mechanisms=PLAIN \
         -X ssl.ca.location=./etc/ssl/cert.pem -X api.version.request=true \
         -b ${CCLOUD_BROKER_HOST} \
         -X sasl.username="${CCLOUD_API_KEY}" \
         -X sasl.password="${CCLOUD_API_SECRET}" \
         -t dlq-lcc-emj3x \
         -C -c1 -o beginning \
         -f 'Topic %t[%p], offset: %o, Headers: %h'

Topic dlq-lcc-emj3x[0], offset: 12006, Headers: __connect.errors.topic=mysql-01-asgard.demo.transactions,__connect.errors.partition=5,__connect.errors.offset=90,__connect.errors.connector.name=lcc-emj3x,__connect.errors.task.id=0,__connect.errors.stage=VALUE_CONVERTER,__connect.errors.class.name=org.apache.kafka.connect.json.JsonConverter,__connect.errors.exception.class.name=org.apache.kafka.connect.errors.DataException,__connect.errors.exception.message=Converting byte[] to Kafka Connect data failed due to serialization error: ,__connect.errors.exception.stacktrace=org.apache.kafka.connect.errors.DataException: Converting byte[] to Kafka Connect data failed due to serialization error:
        at org.apache.kafka.connect.json.JsonConverter.toConnectData(JsonConverter.java:344)
        at org.apache.kafka.connect.runtime.WorkerSinkTask.lambda$convertAndTransformRecord$1(WorkerSinkTask.java:487)
        at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndRetry(RetryWithToleranceOperator.java:128)
        at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndHandleError(RetryWithToleranceOperator.java:162)
        at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execute(RetryWithToleranceOperator.java:104)
        at org.apache.kafka.connect.runtime.WorkerSinkTask.convertAndTransformRecord(WorkerSinkTask.java:487)
        at org.apache.kafka.connect.runtime.WorkerSinkTask.convertMessages(WorkerSinkTask.java:464)
        at org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:320)
        at org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:224)
        at org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:192)
        at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:177)
        at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:227)
        at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)
        at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
        at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
        at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
        at java.base/java.lang.Thread.run(Thread.java:834)
Caused by: org.apache.kafka.common.errors.SerializationException: com.fasterxml.jackson.core.JsonParseException: Illegal character ((CTRL-CHAR, code 0)): only regular white space (\r, \n, \t) is allowed between tokens
 at [Source: (byte[])"

从那里开始，这只是理解错误的一个例子。很多时候都归结为序列化问题，您可以在此处了解更多信息。