小编Gio*_*ous的帖子

这是我在 Stackoverflow 上的第一篇文章，希望我没有选错部分。

语境 ：

Kafka HEAP 大小在以下文件中配置：

/etc/systemd/system/kafka.service

使用以下参数：

Environment="KAFKA_HEAP_OPTS=-Xms6g -Xmx6g"

操作系统是“CentOS Linux 7.7.1908 版”。

Kafka 是"confluent-kafka-2.12-5.3.1-1.noarch"，从以下存储库安装：

# Confluent REPO
[Confluent.dist]
name=Confluent repository (dist)
baseurl=http://packages.confluent.io/rpm/5.3/7
gpgcheck=1
gpgkey=http://packages.confluent.io/rpm/5.3/archive.key
enabled=1

[Confluent]
name=Confluent repository
baseurl=http://packages.confluent.io/rpm/5.3
gpgcheck=1
gpgkey=http://packages.confluent.io/rpm/5.3/archive.key
enabled=1

几天前，我在 3 台机器 KAFKA 集群上激活了 SSL，突然，以下命令停止工作：

kafka-topics --bootstrap-server <the.fqdn.of.server>:9093 --describe --topic <TOPIC-NAME>

其中返回以下错误：

[2019-10-03 11:38:52,790] ERROR Uncaught exception in thread 'kafka-admin-client-thread | adminclient-1':(org.apache.kafka.common.utils.KafkaThread) 
java.lang.OutOfMemoryError: Java heap space
    at java.nio.HeapByteBuffer.<init>(HeapByteBuffer.java:57)
    at java.nio.ByteBuffer.allocate(ByteBuffer.java:335)
    at org.apache.kafka.common.memory.MemoryPool$1.tryAllocate(MemoryPool.java:30)
    at org.apache.kafka.common.network.NetworkReceive.readFrom(NetworkReceive.java:112)
    at org.apache.kafka.common.network.KafkaChannel.receive(KafkaChannel.java:424)
    at org.apache.kafka.common.network.KafkaChannel.read(KafkaChannel.java:385)
    at …

我们在测试和生产中得到了这个。它是连续的，我们不知道这些错误如何每隔几秒钟就会出现一次，至少我们在测试中似乎没有来自另一个系统的反馈。

我们的消息非常小，最多只有几百字节。

这是 1.2 GB。我尝试将：socket.request.max.bytes 设置为 1195725856，但随后内存不足，即使堆大小约为 2.5 GB 并且 OpenShift 容器设置为最大 32GB。

非常欢迎任何帮助！

org.apache.kafka.common.network.InvalidReceiveException: Invalid receive (size = 1195725856 larger than 104857600)
    at org.apache.kafka.common.network.NetworkReceive.readFromReadableChannel(NetworkReceive.java:132)
    at org.apache.kafka.common.network.NetworkReceive.readFrom(NetworkReceive.java:93)
    at org.apache.kafka.common.network.KafkaChannel.receive(KafkaChannel.java:235)
    at org.apache.kafka.common.network.KafkaChannel.read(KafkaChannel.java:196)
    at org.apache.kafka.common.network.Selector.attemptRead(Selector.java:545)
    at org.apache.kafka.common.network.Selector.pollSelectionKeys(Selector.java:483)
    at org.apache.kafka.common.network.Selector.poll(Selector.java:412)
    at kafka.network.Processor.poll(SocketServer.scala:551)
    at kafka.network.Processor.run(SocketServer.scala:468)
at java.lang.Thread.run(Thread.java:748)

我有一个具有以下配置的接收器连接器

{
    "name": "sink-test-mariadb-MY_TOPIC",
    "config": { 
                "connector.class":"io.confluent.connect.jdbc.JdbcSinkConnector",
                "tasks.max":"10",
                "topics":"MY_TOPIC",
                "connection.url":"jdbc:mariadb://localhost:3306/myschema?user=myuser&password=mypass",
                "auto.create":"false",
                "auto.evolve":"true",
                "table.name.format":"MY_TABLE",
                "pk.mode":"record_value",
                "pk.fields":"ID",
                "insert.mode":"upsert",
                "transforms":"ExtractField",
                "transforms.ExtractField.type":"org.apache.kafka.connect.transforms.ExtractField$Value",
                "transforms.ExtractField.field":"data"
        }
}

一段时间后，连接器的所有任务都会失败，并出现以下错误：

{
    "state": "FAILED",
    "trace": "org.apache.kafka.connect.errors.DataException: MY_TOPIC
                at io.confluent.connect.avro.AvroConverter.toConnectData(AvroConverter.java:95)
                at org.apache.kafka.connect.runtime.WorkerSinkTask.convertMessages(WorkerSinkTask.java:468)
                at org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:301)
                at org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:205)
                at org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:173)
                at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:170)
                at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:214)
                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)
            Caused by: org.apache.kafka.common.errors.SerializationException: Error retrieving Avro schema for id 802
            Caused by: io.confluent.kafka.schemaregistry.client.rest.exceptions.RestClientException: Schema not found; error code: 40403
                at io.confluent.kafka.schemaregistry.client.rest.RestService.sendHttpRequest(RestService.java:202)
                at io.confluent.kafka.schemaregistry.client.rest.RestService.httpRequest(RestService.java:229)
                at io.confluent.kafka.schemaregistry.client.rest.RestService.getId(RestService.java:409)
                at io.confluent.kafka.schemaregistry.client.rest.RestService.getId(RestService.java:402)
                at …

在所有情况下，如果任何一个 Kafka 代理启动并运行，我的生产者和消费者都应该正常运行。

里面的参数server.properties如下，用于集群的高可用。

transaction.state.log.min.isr=2
offsets.topic.replication.factor=3
transaction.state.log.replication.factor=3

1. 如果所有 3 个 Kafka 代理（K1、K2 和 K3）均已启动，则集群运行良好。
2. 如果 K1 和 K2 代理出现故障并且只有 K3 正常运行，则集群运行良好。
3. 如果所有服务器都关闭并且我们启动 K1 Broker，集群将停止接收消息，消费者也会停止。

如果我遗漏了什么，请告诉我，如果我设置了怎么办transaction.state.log.min.isr=3？

提前致谢。

是否可以在 Kafka 控制台消费者中检索特定时间戳范围的消息？

例如，昨天 08:00 到 09:00 之间的 kafka 消息。

Kafka允许客户端通过SSL连接.默认情况下,SSL已禁用,但我通过引用以下链接启用了该功能. http://docs.confluent.io/2.0.0/kafka/ssl.html

完成所有配置后,Producer/Consumer无法生成/使用该消息.

    [2016-02-29 09:20:49,189] ERROR Error when sending message to topic ssltopic with key: null, value: 2 bytes with error: Failed to update metadata after 60000 ms. (org.apache.kafka.clients.producer.internals.ErrorLoggingCallback)
sas
[2016-02-29 09:21:16,031] WARN Failed to send SSL Close message  (org.apache.kafka.common.network.SslTransportLayer)
java.io.IOException: Connection reset by peer

我正在写docker-compose文件到MySQL实例,并想从env文件中使用一些变量:这里的文件实际上是这样的:

泊坞窗,compose.yml

version: '3.3'
services:
  db:
    image: mysql
    restart: always
    env_file:
      - ./imran.env
    environment:
      MYSQL_ROOT_PASSWORD: ${PASS}
    ports:
      - ${PORT1}: ${PORT2}

imran.env

PASS=imran123
PORT1=3306
PORT2=3306

而不是正确的工作我得到以下错误:

WARNING: The PASS variable is not set. Defaulting to a blank string.
WARNING: The PORT2 variable is not set. Defaulting to a blank string.
ERROR: The Compose file './docker-compose.yml' is invalid because:
services.db.ports contains unsupported option: '${PORT1}

请帮忙

假设我想在小型部署的生产环境中拥有高度可用的 Kafka。我必须使用以下配置

min.insync.replicas=2 // Don't want to lose messages in case of 1 broker crash    
default.replication.factor=3 // Will let producer write in case of 1 replica disappear with broker crash

如果 1 个 broker 崩溃并且 1 个副本消失，Kafka 会开始制作新副本吗？

在任何情况下，我们是否必须至少有default.replication.factor数量的经纪人才能继续工作？

数据并不完全干净，但在使用熊猫时没有问题。pandas 库为 EDA 提供了许多非常有用的函数。

但是，当我对大数据（即 10 亿条记录和 10 列）使用分析时，从数据库表中读取它时，它没有完成并且我的笔记本电脑内存不足，csv 中的数据大小约为 6 GB，我的 RAM 为 14 GB 我的空闲使用量大约是 3 - 4 GB。

df = pd.read_sql_query("select * from table", conn_params)
profile = pandas.profiling.ProfileReport(df)
profile.to_file(outputfile="myoutput.html")

我也尝试过使用check_recoded = False选项。但这对完全分析没有帮助。有什么办法可以对数据进行分块读取，最终生成一个整体的汇总报告？或任何其他方法将此函数用于大型数据集。

我们使用 schema 注册表来存储 schema，消息被序列化到 avro 并推送到 kafka 主题。

想知道，当从消费者读取数据时，如何找到 avro 记录被序列化的 schema id。我们需要此架构 ID 来跟踪是否将新列添加到表中的更改。如果添加或删除新列，架构注册表中将生成新的架构 id，以及如何在消费者中获取该 id。

consumer = KafkaConsumer(bootstrap_servers = conf['BOOTSTRAP_SERVERS'],
                        auto_offset_reset = conf['AUTO_OFFSET'],
                        enable_auto_commit = conf['AUTO_COMMIT'],
                        auto_commit_interval_ms = conf['AUTO_COMMIT_INTERVAL']
                        )
consumer.subscribe(conf['KAFKA_TOPICS'])

for message in consumer:
    print(message.key)

从上面的代码中，message.key 打印该特定记录的键，以及我们如何找到消费者用来反序列化记录的相应模式 ID。

curl -X GET http://localhost:8081/subjects/helpkit_internal.helpkit_support.agents-value/versions/2

{"subject":"helpkit_internal.helpkit_support.agents-value","version":2,"id":33,"schema":"{\"type\":\"record\",\"name\":\"Envelope\",\"namespace\":\"helpkit_internal.helpkit_support.agents\",\"fields\":[{\"name\":\"before\",\"type\":[\"null\",{\"type\":\"record\",\"name\":\"Value\",\"fields\":[{\"name\":\"id\",\"type\":\"long\"},{\"name\":\"user_id\"

我们想要从消费者那里获取 id 值"id":33

请就此提出建议。