Kafka Connect:Java . lang . illegalstateexception:没有分区的当前赋值

Question

我在库伯内特斯（8-16个节点，自动缩放）上运行Kafka Connect。我总共定义了44个连接器，每个Kafka主题一个（每个主题一个分区）。这些主题是由Debezium/Postgreql生成的。有3个Kafka节点。每个连接器tasks.max设置为4。我的大多数连接器（但不是每个！）有一个（总是一个）失败任务，由于java.lang.IllegalStateExc的：分区-0没有当前分配。

这里不是Kafka专家，注意；)我假设有3个Kafka节点，所以3个工人做得很好，而第4个任务没有任何连接，所以它失败了。但是为什么有时有4个任务运行良好呢？

此外，我经常遇到“由于重新平衡而导致的冲突操作”问题，这可能会发生几分钟甚至几个小时。最近我删除了所有pod，它们自己重新启动，问题消失了，但这不是长期的解决方案。

任务是什么。最大推荐值？提前感谢！

例外情况：

java.lang.IllegalStateException: No current assignment for partition table-0
    at org.apache.kafka.clients.consumer.internals.SubscriptionState.assignedState(SubscriptionState.java:259)
    at org.apache.kafka.clients.consumer.internals.SubscriptionState.seek(SubscriptionState.java:264)
    at org.apache.kafka.clients.consumer.KafkaConsumer.seek(KafkaConsumer.java:1501)
    at org.apache.kafka.connect.runtime.WorkerSinkTask.rewind(WorkerSinkTask.java:601)
    at org.apache.kafka.connect.runtime.WorkerSinkTask.access$1200(WorkerSinkTask.java:70)
    at org.apache.kafka.connect.runtime.WorkerSinkTask$HandleRebalance.onPartitionsAssigned(WorkerSinkTask.java:675)
    at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.onJoinComplete(ConsumerCoordinator.java:291)
    at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.joinGroupIfNeeded(AbstractCoordinator.java:406)
    at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureActiveGroup(AbstractCoordinator.java:340)
    at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.poll(ConsumerCoordinator.java:341)
    at org.apache.kafka.clients.consumer.KafkaConsumer.updateAssignmentMetadataIfNeeded(KafkaConsumer.java:1214)
    at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1179)
    at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1164)
    at org.apache.kafka.connect.runtime.WorkerSinkTask.pollConsumer(WorkerSinkTask.java:445)
    at org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:318)
    at org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:226)
    at org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:194)
    at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:175)
    at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:219)
    at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
    at java.util.concurrent.FutureTask.run(FutureTask.java:266)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
    at java.lang.Thread.run(Thread.java:748

接收器连接器配置：

connector.class com.wepay.kafka.connect.bigquery.BigQuerySinkConnector
autoUpdateSchemas   true
sanitizeTopics  true
autoCreateTables    true
topics  <topic-name>
tasks.max   3
schemaRegistryLocation  http://<ip>:8081
project <big-query-project>
maxWriteSize    10000
datasets    .*=<big-query-dataset>
task.class  com.wepay.kafka.connect.bigquery.BigQuerySinkTask
keyfile /credentials/<credentials-file>.json
name    <connector-name>
schemaRetriever com.wepay.kafka.connect.bigquery.schemaregistry.schemaretriever.SchemaRegistrySchemaRetriever
tableWriteWait  1000
bufferSize  100000

并且它抛出了上面的异常＜code＞java.lang.IllegalStateException：没有当前的〔…〕赋值＜/code＞

Answer 1

属性任务的值.max取决于几个因素。最重要的是特定的连接器。特定连接器取决于其逻辑和任务值.max计算将要创建的任务数。例如，FileStreamSourceConnector 始终创建 1 个任务，因此即使您传递的值大于 1，它也只会创建一个。与PostgresConnector平行的情况相同。

tasks.max值还应取决于其他因素，如：Kafka Connect模式、您拥有多少Kafka Connect实例、机器的CPU等。

我怎么理解你在用source connector(< code > PostgresConnector )。源连接器不轮询来自Kafka的数据。您发布的异常与某个< code>SinkConnector有关。如果使用的是< code>SinkConnector，则< code>tasks.max不应超过分区数。如果您启动的任务多于分区数量，一些分区将处于空闲状态(状态为正在运行，但它们不处理数据),重新平衡可能会发生。