0

我得到了这个例外

org.apache.kafka.common.protocol.types.SchemaException

当 kafka 重新平衡时

这些是详细信息：

  1. 使用 Azure 事件中心。使用 kafka API 访问它

  2. “已启用 Kafka”= 是，在 Azure 门户中

  3. 使用：编译组：'org.apache.kafka'，名称：'kafka-clients'，版本：'1.0.2'

  4. 使用消费者组

  5. Properties properties = new Properties();
properties.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
properties.put(SaslConfigs.SASL_JAAS_CONFIG, saslJaasConfig);
properties.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_SSL");
properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, String.format("%s.servicebus.windows.net:9093", this.namespace));
properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, LongDeserializer.class.getName());
properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, MeasurementDeSerializer.class.getName());
properties.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 120000);
properties.put(ConsumerConfig.GROUP_ID_CONFIG, this.groupName);
properties.put(ConsumerConfig.CLIENT_ID_CONFIG, clientId);

我在 2 台不同的 PC 上有 2 个客户端

当它们都运行时，每个都从可用的 32 个分区中获得 16 个分区。

当我关闭其中一个时，所有部件都会重新平衡到另一个。

在仍在运行的实例上，我将得到：

  1. 已撤销的分区 [16, 17, 18, 19, 20, 21, 22, 23, 24, 25, 26, 27, 28, 29, 30, 31]

  2. 然后，从池循环中我会得到这个异常：

    org.apache.kafka.common.protocol.types.SchemaException：读取字段“leader_id”时出错：org.apache.kafka.common.protocol.types.Schema.read（Schema.java:76）的字符串长度-1不能为负) 在 org.apache.kafka.clients.NetworkClient.parseStructMaybeUpdateThrottleTimeMetrics(NetworkClient.java:586) 在 org.apache.kafka.clients 的 org.apache.kafka.common.protocol.ApiKeys.parseResponse(ApiKeys.java:279)。 NetworkClient.handleCompletedReceives(NetworkClient.java:686) at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:469) at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java: 258) 在 org.apache.kafka.clients 的 org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:230)。consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:190) at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.joinGroupIfNeeded(AbstractCoordinator.java:364) at org.apache.kafka.clients.consumer.internals。 AbstractCoordinator.ensureActiveGroup(AbstractCoordinator.java:316) at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.poll(ConsumerCoordinator.java:295) at org.apache.kafka.clients.consumer.KafkaConsumer.pollOnce(KafkaConsumer. java:1146) 在 org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1111)internals.AbstractCoordinator.ensureActiveGroup(AbstractCoordinator.java:316) at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.poll(ConsumerCoordinator.java:295) at org.apache.kafka.clients.consumer.KafkaConsumer.pollOnce( KafkaConsumer.java:1146) 在 org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1111)internals.AbstractCoordinator.ensureActiveGroup(AbstractCoordinator.java:316) at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.poll(ConsumerCoordinator.java:295) at org.apache.kafka.clients.consumer.KafkaConsumer.pollOnce( KafkaConsumer.java:1146) 在 org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1111)

另一方面，当走另一条路时，没有问题

  1. 启动第一个实例

  2. 实例 1 获取所有 32 个分区

  3. 启动实例 2

  4. 重新平衡开始

  5. 实例 1 丢失 16 个零件

  6. 实例 2 得到 16 个部分

知道什么可能导致此异常吗？

4

1 回答 1

0

对于未来的读者 - 问题已修复。 https://github.com/Azure/azure-event-hubs-for-kafka/issues/41

于 2019-10-16T22:58:00.117 回答