我对kafka -> elasticsearch连接器有一个奇怪的问题。当我第一次开始的时候，一切都很棒，我在 elasticsearch 中收到了一个新数据并通过 kibana 仪表板进行了检查，但是当我使用相同的生产者应用程序将新数据生成到 kafka 并尝试再次启动连接器时，我没有在 elasticsearch 中获得任何新数据。 现在我收到这样的错误：
[2018-02-04 21:38:04,987] ERROR WorkerSinkTask{id=log-platform-elastic-0} Commit of offsets threw an unexpected exception for sequence number 14: null (org.apache.kafka.connect.runtime.WorkerSinkTask:233)
org.apache.kafka.connect.errors.ConnectException: Flush timeout expired with unflushed records: 15805
我正在使用下一个命令来运行连接器：
/usr/bin/connect-standalone /etc/schema-registry/connect-avro-standalone.properties log-platform-elastic.properties
connect-avro-standalone.properties：
bootstrap.servers=kafka-0.kafka-hs:9093,kafka-1.kafka-hs:9093,kafka-2.kafka-hs:9093
key.converter=io.confluent.connect.avro.AvroConverter
key.converter.schema.registry.url=http://localhost:8081
value.converter=io.confluent.connect.avro.AvroConverter
value.converter.schema.registry.url=http://localhost:8081
internal.key.converter=org.apache.kafka.connect.json.JsonConverter
internal.value.converter=org.apache.kafka.connect.json.JsonConverter
internal.key.converter.schemas.enable=false
internal.value.converter.schemas.enable=false
offset.storage.file.filename=/tmp/connect.offsets
# producer.interceptor.classes=io.confluent.monitoring.clients.interceptor.MonitoringProducerInterceptor
# consumer.interceptor.classes=io.confluent.monitoring.clients.interceptor.MonitoringConsumerInterceptor
#rest.host.name=
rest.port=8084
#rest.advertised.host.name=
#rest.advertised.port=
plugin.path=/usr/share/java
和log-platform-elastic.properties：
name=log-platform-elastic
key.converter=org.apache.kafka.connect.storage.StringConverter
connector.class=io.confluent.connect.elasticsearch.ElasticsearchSinkConnector
tasks.max=1
topics=member_sync_log, order_history_sync_log # ... and many others
key.ignore=true
connection.url=http://elasticsearch:9200
type.name=log
我检查了与 kafka 代理、弹性搜索和模式注册表的连接(模式注册表和连接器目前在同一主机上)，一切都很好。Kafka 代理在端口 9093 上运行，我能够使用 kafka-avro-console-consumer 从主题中读取数据。 我将不胜感激在这方面的任何帮助！
只需将flush.timeout.ms更新为大于 10000(默认值为 10 秒)
根据文档：
flush.timeout.ms用于定期的超时(以毫秒为单位) 刷新，以及等待缓冲区空间可用时 添加记录时已完成的请求。如果超过此超时 任务将失败。
类型：长 默认值：10000 重要性：低
请参阅文档
我们可以优化 Elastic 搜索配置来解决问题。有关配置参数，请参阅以下链接
https://docs.confluent.io/current/connect/kafka-connect-elasticsearch/configuration_options.html
以下是可以控制消息速率流以最终帮助解决问题的关键参数：
flush.timeout.ms：增加可能有助于在冲洗时间上提供更多的呼吸
用于定期刷新的超时(以毫秒为单位)，以及何时 等待已完成的请求提供缓冲区空间，因为 添加记录。如果超过此超时，任务将失败。
max.buffered.records：尝试减少缓冲区记录限制
每个任务在阻塞之前将缓冲的最大记录数 接受更多记录。此配置可用于限制 每个任务的内存使用情况
batch.size：尝试减小批处理大小
写入时要作为批处理处理的记录数 弹性搜索
任务.max：并行线程数(使用者实例)减少或增加。如果带宽无法处理减少任务可能会有所帮助，这将控制弹性搜索。
它通过调整上面的参数解决了我的问题