问题标签 [confluent-cloud]

对于托管在 Confluent Cloud 中的 Kafka 集群，会创建一个 Audit Log 集群。似乎可以将 Sink 连接器连接到该集群并从“confluent-audit-log-events”主题中排出事件。

但是，当我运行连接器执行相同操作时，我遇到了以下错误。

在我的 connect-distributed.properties 文件中，我的设置为：

需要授予哪些额外权限，以便连接器可以在集群中创建所需的主题？connect-distributed.properties 文件中使用的密钥/密钥是与此集群的服务帐户关联的有效密钥/密钥。

此外，当我在控制台中使用相同的键（如上）运行消费者时，我能够很好地读取审计日志事件。

1. 它是以纪元毫秒还是以秒为单位？
2. 它是否代表消息到达kafka的时间？还是由 kafka-client-producer 发送的？或者是什么时候吃的？
3. 如何知道我的 VM 与 Confluent Cloud 中的 Kafka VM 之间的时钟时间差？

我需要计算 KSQL confluent.cloud 中两个时间戳的差异。

以下是我的查询，

在这里，我有两个时间戳，例如 ROWTIME 和 UpdatedDateFROMtblEntity。那么，如何计算这些时间戳的差异？我需要以毫秒为单位。请帮我解决这个问题。

我目前在 Confluent 云上使用基本版本的集群，我只有一个主题有 9 个分区。我有一个使用 AWS lambda 服务设置的 REST Api，该服务将消息发布到 Kafka。目前我正在以每秒 5k-10k 的请求进行压力测试管道，我发现延迟达到 20-30 秒才能发布大小为 1kb 的记录。单个请求通常为 300 毫秒。我将 linger.ms - 500 ms 和 batch.size 等生产者配置添加到 100kb。我看到了一些改进（每个请求 15-20 秒），但我觉得它仍然太高了。有什么我遗漏的东西，还是融合云上的基本集群有什么东西？集群上的所有配置都是默认的。

出于某种原因，有时，我们的 Confluent Kafka 消费者会收到关于特定主题的奇怪消息：

即使我吞下了所有错误并在失败后提交了偏移量，我也多次使用了完全相同的消息。

它只发生在一个特定的主题上。我们从 Scarth 删除并重新创建了这个主题，但它没有解决这个问题。

在以下屏幕截图中，我们看到一条消息为空，而其他消息不为空。这是什么意思？

生产者 - C#：

消费者 NodeJ：

我们无法使用ccloud使用有问题的消息：错误 - “恐慌：运行时错误：索引超出范围 [0]，长度为 0”

我们迷路了。任何帮助都会很棒！

NodeJS：v14.15.4 KafkaJS：1.16.0-beta.18

我正在遵循与此博客文章中类似的示例：

https://rmoff.net/2019/11/12/running-dockerised-kafka-connect-worker-on-gcp/

除了我不是在 GCP 上而是在本地运行 kafka connect worker 之外。

一切都很好，我运行了 docker-compose up 并启动​​了 kafka 连接，但是当我尝试通过 CURL 创建源连接器的实例时，我收到以下模棱两可的消息（注意：kafka 连接日志中实际上没有输出任何日志）：

我知道我可以连接到融合云，因为我看到正在创建主题：

我的 docker-compose.yml 看起来像这样：

我已经运行了 dockerized mongo 实例，我想创建 mongo 源连接器，这是我的 CURL 请求：

我错过了什么？

也许有人可以帮我解决一个小问题。有没有办法使用 Confluent 的 MQTT-Proxy 中内置的 Producer 配置密钥分区？

只有一个 Topic-Mapping-Property，但我还没有找到如何将 MQTT-Proxy 与 Key 分区的 Topic 一起使用。我可以在 MQTT 消息中包含 Kafka 协议的密钥吗？

多谢你们。

我在 confluent.cloud 中创建了主题。现在我想根据消费者滞后在 Azure 中创建自动缩放。

你能为我提供解决方案吗？

我在 Spark Streaming 中使用 Kafka Source 来接收使用 Confluent Cloud 中的 Datagen 生成的记录。我打算使用 Confluent Schema Registry，

目前，这是我面临的例外：*

线程“主”io.confluent.kafka.schemaregistry.client.rest.exceptions.RestClientException 中的异常：未授权；错误代码：401

融合云的架构注册表需要传递一些我不知道如何输入的身份验证数据：

我想我必须将此身份验证数据传递给 CachedSchemaRegistryClient 但我不确定是否如此以及如何。

如果我尝试将身份验证发送到架构注册表

我明白 Cannot resolve overloaded constructor CachedSchemaRegistryClient了，似乎只有 2 个参数要发送到 CachedSchemaRegistryClient。

我该如何解决？

我遇到了这篇文章，但在这里他们没有对融合云中的模式注册表应用任何身份验证。

我已经建立了一个融合的云多区域集群，它只用一个引导服务器创建。创建集群时没有选择服务器数量的设置。即使在创建之后，我也无法编辑引导服务器的数量。

我想知道如何增加融合云 kafka 集群中的服务器数量。