Kafka表中聚合记录超时？
Kafka表中聚合记录超时是指在Kafka中进行数据聚合操作时，某个记录的处理时间超过了预设的超时时间。这种情况可能会导致数据处理的延迟和性能下降。
Kafka是一种分布式流处理平台，用于高吞吐量、低延迟的数据传输和处理。它采用了发布-订阅模式，将数据以消息的形式进行传输和处理。在Kafka中，可以通过消费者组对消息进行聚合操作，例如将多个消息合并为一个批次进行处理。
当在Kafka表中进行聚合记录时，如果某个记录的处理时间超过了预设的超时时间，就会发生聚合记录超时的情况。这可能是由于数据量过大、处理逻辑复杂或者处理节点负载过高等原因导致的。
为了解决聚合记录超时的问题，可以采取以下措施：
- 调整超时时间：根据实际情况，适当调整聚合记录的超时时间，使其能够容纳更多的处理时间。
- 增加处理节点：通过增加处理节点的数量，可以提高整体的处理能力，减少聚合记录超时的概率。
- 优化处理逻辑：对聚合记录的处理逻辑进行优化，减少不必要的计算和IO操作，提高处理效率。
- 监控和调优：使用监控工具对Kafka集群进行监控，及时发现聚合记录超时的情况，并进行相应的调优操作。
腾讯云提供了一系列与Kafka相关的产品和服务，包括消息队列 CKafka、流数据分析平台 DataWorks、流计算平台流计算 Oceanus 等。这些产品可以帮助用户搭建高可靠、高性能的消息队列系统，实现实时数据处理和分析。
更多关于腾讯云的产品和服务信息，请访问腾讯云官方网站：https://cloud.tencent.com/
相关·内容
1回答
我使用Kafka来处理消息。消息可以分成几个部分(它是一个复合消息)。例如，在流中，我可以有一个被分成三个部分的复合消息。换句话说，它将是Kafka流中的三个记录，但这是一个大消息。我想使用Kafka表在一个Kafka记录中合并复合消息的各个部分。合并后，将在数据库(Postgres)中插入一封邮件。每个部件都有部件数和总部件数。例如，如果我在流中有一条消息的三个部分(三个Kafka记录)-每个部分都有字段total number of part，值为3。
我的理解是，任务在积极的场景中很简单:在表中聚合部件，从表中创建流，并过滤具有等于聚合部件大小和部件总数的记录，映射在一个合并的消息中过滤
浏览 5提问于2018-08-01得票数 0
我有ApacheBeam2.2.0实时作业，它运行在平台上。这项工作基本上是从Kafka读取json事件，并将其转换并写入BigQuery。问题是，Google不断关闭现有的Kafka用户，然后创建新的用户，其信息如下：
记录器："com.google.cloud.dataflow.worker.ReaderCache“
讯息：“关闭s4-0000000000000014的闲置阅读器”
舞台："S4“
步骤：“从Kafka/KafkaIO.Read/KafkaIO.Read/Read(UnboundedKafkaSource)/DataflowRunner.Streaming
浏览 0提问于2018-02-13得票数 1
回答已采纳
注意:使用的Kafka版本：kafka_2.11-0.9.0.1
我有一个名为test-kafka的主题，分别是3 partitions和1 replication factor，这个主题在每个分区中都有一些字符串数据，即key and value pair。
当我试图通过消费者api获取记录时，consumer.poll函数不会获取任何记录并提供records.count =0，但是当我对single parition使用相同的逻辑时，它会很好地工作并获取记录。
我认为Kafka将在内部处理多个分区，并将从每个分区逐个为我获取记录，我只需要subscribe到一个topic。
这里有人能帮上
浏览 2提问于2016-06-22得票数 0
2回答
Kafka微服务适当用例
、、、、
在我新工作的项目中，我发现微服务不是直接从一个微服务到另一个微服务进行post/put API调用，而是向kafka生成一条消息，然后由单个微服务使用该消息。
例如，Order微服务将向"pending-order“主题发布一条记录，然后由库存微服务(没有其他消费者)使用该记录。反过来，在使用记录并做了一些处理之后，库存微服务将产生一个记录到“已处理订单”，然后只由订单微服务使用。
这是一个正确的用例吗？或者在这种情况下，只在微服务之间进行API调用更好？
浏览 0提问于2019-06-19得票数 1
我在使用卡夫卡-风暴集成。Kafka将将数据加载到队列中，Kafka Spout将提取数据和进程。我有下面的设计。
Kafka -> Queue -> KafkaSpout -> Process1 Bolt -> Process2 Bolt
问题是，如果Process2 Bolt需要较长的时间来处理KafkaSpout正在失败的数据，并且再次尝试从队列中读取数据，这将导致重复记录。
如果博尔特处理得很慢，为什么KafkaSpout把它当作失败呢？解决办法是什么？有什么时间或类似的财产，我必须在风暴中？
浏览 6提问于2015-05-14得票数 2
我有apache访问日志文件，我想在cassandra表中存储每个页面的访问计数(总数/每天/每小时)。
我正在尝试使用kafka连接从日志文件流到kafka主题。为了在Cassandra中增加指标计数器，我可以再次使用Kafka Connect吗？否则，这里应该使用哪些其他工具，例如kafka streams，spark，flink，kafka connect等？
浏览 0提问于2017-10-11得票数 0
我有一个奇怪的问题卡夫卡-> 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 se
浏览 0提问于2018-02-04得票数 4
Kafka消费者轮询api未将记录返回到低超时。如果我在轮询中增加超时值，那么记录就会出现。我不能理解这个逻辑。请帮助，遵循以下代码：
public ConsumerRecords<String, Map<String, String>> subscribeToQueue(String topic, QueueListener q) {
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
pro
浏览 0提问于2017-06-23得票数 2
我正在尝试使用输入主题作为KTable对记录进行重复数据删除，并将它们下沉到输出主题。但KTable仍在将重复记录下沉到输出主题。不确定我哪里错了。 这是我的application.yml spring:
cloud:
stream:
function:
bindings:
process-in-0: input.topic
process-out-0: output.topic
definition: process
kafka:
streams:
浏览 38提问于2021-05-07得票数 1
回答已采纳
有一个包含80到100个记录的ArrayList，试图将每个单独的记录(POJO，而不是整个列表)发送到Kafka主题(事件中心)。像每小时一样安排一个cron作业，将这些记录(POJO)发送到事件中心。
能够看到消息被发送到eventhub，但在3到4成功运行之后，得到以下异常(这包括发送了几条消息，还有几条在下面的异常中失败)
Expiring 14 record(s) for eventhubname: 30125 ms has passed since batch creation plus linger time
下面是使用的生产者的配置，
props.put(Pr
浏览 3提问于2019-09-19得票数 1
回答已采纳
我试图写一个数据框架，它有大约2.3亿条记录给卡夫卡。更具体地说，，但我不确定这是否真的是我的问题的根源。
EH_SASL = 'kafkashaded.org.apache.kafka.common.security.plain.PlainLoginModule required username="$ConnectionString" password="Endpoint=sb://myeventhub.servicebus.windows.net/;SharedAccessKeyName=RootManageSharedAccessKey;SharedAc
浏览 1提问于2018-12-13得票数 5
回答已采纳
1回答
我已经成功地获得了一个用于消费、转换和生成数据的流应用程序，但是我注意到，定期地，流处理器将转换到ERROR状态，并且进程将处于不退出状态。
给我看原木如下：
All stream threads have died. The instance will be in error state and should be closed.
有什么方法可以告诉流应用程序一旦到达ERROR状态就退出吗？也许是某种监控线程？
我在Kafka代码的注释中看到了，当应用程序达到这种状态时，用户需要关闭它，但是，我在文档中还没有找到对这个任务的提及。
有简单的方法来完成这个关机步骤吗？
可能不正确的方式可能关
浏览 0提问于2018-03-06得票数 10
回答已采纳
我有一个问题，卡夫卡制片人在生产中，我看到下面的错误。
“发布失败，myTopic的过期2条记录-1:120004 ms自批创建以来已过[ org.apache.kafka.common.errors.TimeoutException:到期2条记录的myTopic-1:120004 ms自批创建以来”
Kafka经纪人的版本为5.3.2，而kafka-client为Apache2.3.1。生产者配置(在我的代码中显式指定)在下面，其余是默认设置。
batch.size = 102400 linger.ms = 100 compression.type = lz4 ack = all
示例Jav
浏览 6提问于2022-06-03得票数 1
我的Scs应用程序有两个Kafka生产者，其配置如下：
spring:
cloud:
function:
definition: myProducer1;myProducer2
stream:
bindings:
myproducer1-out-0:
destination: topic1
producer:
useNativeEncoding: true
myproducer2-out-0:
destination: topic2
浏览 4提问于2020-04-14得票数 5
1回答
我有一个java服务器端应用程序，它侦听tcp连接，读取消息，处理和发布到kafka主题。我们使用带有kafka的事务性kafka模板来实现一次，只有当我们通过semantics.This进行记录时，record.We才能很好地提交一批记录(检查消息是否具有某个参数集，然后提交)或基于时间间隔。这是卡夫卡template.We使用SpringKafka2.1的可能性吗？
浏览 0提问于2018-09-14得票数 1
回答已采纳
我有一个基于DSL的流，它使用split遍历对象列表并发送Kafka消息：
.transform(...)
.split()
.channel(KAFKA_OUT_CHANNEL)
发送完所有消息后，我需要调用服务，还需要记录处理了多少消息。我知道一种方法是使用publishSubscribeChannel，其中第一个subscribe执行实际的Kafka发送，然后aggregate执行服务调用：
.transform(...)
.split().
.publishSubscribeChannel(pubSub -> pubSub
.subscribe(f -> f
浏览 1提问于2018-06-27得票数 0
我是原型，计算聚合在一个星火结构流(Spark3.0)作业和发布的更新卡夫卡。我需要计算的最大日期和最大百分比的所有时间(没有窗口)的每一组。代码看起来很好，除非在源流中有Kafka墓碑记录(删除)。流接收一个具有有效键和空值的Kafka记录，但是最大聚合继续在计算中包含该记录。什么是最好的选择，让这个重新计算，而没有删除记录时，删除是从卡夫卡消费？
示例
产生的信息：
<"user1|1", {"user": "user1", "pct":30, "timestamp":"2021-01-01 01
浏览 6提问于2021-01-19得票数 0
5回答
我们有一个微服务体系结构，使用Kafka作为服务之间的通信机制.有些服务有自己的数据库。假设用户调用服务A，这将导致在该服务的数据库中创建记录(或记录集)。此外，这一事件应报告给其他服务，作为卡夫卡主题的一个项目。只有在Kafka主题成功更新(基本上围绕数据库更新和Kafka更新创建分布式事务)时，才能确保编写数据库记录的最佳方式是什么？
我们正在考虑使用 (在Spring服务中)，我可以看到它有一个，但据我所知，这更多是关于卡夫卡交易本身(确保卡夫卡生产者和消费者之间的一致性)，而不是在两个系统之间同步交易(参见：“Kafka不支持XA，您必须处理DB可能在Kafka tx回滚时提交的可能性
浏览 0提问于2018-09-06得票数 63
回答已采纳
1回答
1)是需要执行聚合函数的Stream上的Key。我读了几篇博客，还有来自Confluent的建议，即聚合功能工作所需的密钥
CREATE STREAM Employee (EmpId BIGINT, EmpName VARCHAR,
DeptId BIGINT, SAL BIGINT) WITH (KAFKA_TOPIC='EmpTopic',
VALUE_FORMAT='JSON');
在定义上面的Stream时，我没有定义任何键(ROWKEY为空)。基础主题'EmpTopic‘也不是键。
我正在对Stream执
浏览 3提问于2019-06-23得票数 0
1回答
我只是在学习Apache Kafka。我当前的ETL在batch process上运行，现在我希望它在流进程上运行，以便用于报告的数据始终是最新的。据我所知，我可以使用MongoDB连接器来捕获mongodb中的数据变化，然后将其发送到kafka主题。但在我的ETL中，我需要将处理后的数据存储到SQL数据库中。如何以及在哪里处理从mongodb发送到主题的数据，然后从该主题创建到另一个数据库的记录？我是否可以使用AWS lambda函数来执行处理和记录创建？但是，我如何在kafka中调用这个函数呢？
浏览 4提问于2020-06-17得票数 0
如果我实现Kafka Connect与Source and Sink连接器同步2个表，是否有可能在以后更改Kafka平台而不会对表数据产生任何影响。
我使用Kafka连接在微服务数据库之间同步数据，我预测会有一些实例需要从一个Kafka实例移动到另一个实例。
场景:如果我在表A中有100条记录，那么在表B中也会有同样的100条记录可用，并且使用主题T1进行同步。如果我丢弃现有的Kafka topic T1并为同一任务创建一个新的topic T2 (或者如果我从一个Kafka集群移到另一个集群)，是否会再次将表A中的100条记录带到表B中，从而导致重复？
浏览 0提问于2020-03-11得票数 1
1回答
如何让系统中的警报卡夫卡连接器是与雪花数据库？
当前producer.properties是：
bootstrap.servers =XX.XX.XX.XX0: 9092,XX.XX.XX.XX1: 9092,XX.XX.XX.XX2: 9092
compression.type = none
日志目录中的错误: connect.logs中的一些dates=
第一个
[2022 - 06 - 20 14: 52: 10, 239] WARN[Producer clientId = producer - 2] Received invalid metadata error in produce
浏览 21提问于2022-04-15得票数 0
我已经使用Spring Kafka starter创建了一个Kafka消费者(使用KafkaListener接口)。我在containerFactory中为我的消费者设置了pollTimeout。我将其设置为5秒。当我一次发送一批100条消息时，我注意到我的消费者将从该批消息中抓取一些消息，但不是所有消息。我预计消费者会等待5秒，等待额外的记录？
@Bean
public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
ConcurrentKa
浏览 0提问于2020-05-27得票数 0
4回答
由于卡夫卡制作人的各种原因，我经常得到超时异常。目前，我正在使用生产者配置的所有默认值。
我看到了以下超时异常：
60000 ms后，org.apache.kafka.common.errors.TimeoutException:未能更新元数据。
主题的org.apache.kafka.common.errors.TimeoutException:到期1(S)记录-1-0: 30001 ms自上一次追加以来已过。
我有以下问题：
这些超时异常的一般原因是什么？
1. Temporary network issue
2. Server issue? if yes then w
浏览 2提问于2019-02-20得票数 8
回答已采纳
我使用kafka流组件在30分钟的滑动窗口上构建聚合(和)，宽限期为2分钟。我正在按10,000个时间序列(组)进行处理。聚合使用禁用日志记录的持久状态存储。若要在聚合间隔结束时仅输出最终结果，我将使用被抑制的运算符。
对于大多数时间序列，聚合计算都是正确的，但在没有计算的情况下，有很小的百分比。在这些情况下，聚合值在许多情况下正好反映了输入流中的一个记录值。最初，我使用的是启用默认记录缓存(cache.max.bytes.buffering为10 MB)的聚合和禁用的withNoBound()选项，以便在需要时继续分配更多内存。
基于以下建议：
抑制缓冲区内存与流的记录缓存无关，因此，除了
浏览 3提问于2021-04-28得票数 1
1回答
我有一个Kafka消费者项目，它使用来自特定Kafka主题的数据。我一得到90%的记录就进行了处理，但我延迟了对其中一些记录的处理(10%)。
这些记录需要延迟，我无法提交记录，因此可能会导致Kafka将分区重新分配给新节点。为了避免这种情况，我可以将同一主题读两次，并在第二个使用者中延迟获取数据部分，但它需要两次反序列化，因此会带来开销。
有没有可能使用单个消费者读取记录，但与Kafka消费者有两个单独的提交？就提交而言，这基本上类似于有两个不同的消费者，consumer.poll将从单个消费者调用，但每个批处理将有两个consumer.commitSync。我将帮助我避免额外的反序列化和网
浏览 0提问于2017-12-27得票数 0
我想测试一个kafka示例:生产者：
object ProducerApp extends App {
val topic = "topicTest"
val props = new Properties()
props.put("bootstrap.servers", "localhost:9092")
props.put(ConsumerConfig.GROUP_ID_CONFIG, "consumer")
props.put("key.serializer", "org.apache.kafka.
浏览 0提问于2017-11-03得票数 0
我有一个应用程序，它使用KafkaStreams并将更新写入下游数据库(Redis)。当下游数据库服务关闭时，应用程序试图优雅地关闭；但是，它会到达main()的末尾，并由于Kafka的未关闭网络线程而挂起。我可以从日志中确认KafkaStreams是使用kafkaStreams.close()关闭的：
2020-08-07 14:40:54.723 INFO 1 --- [ shutdown-hook] org.apache.kafka.streams.KafkaStreams : stream-client [imc1-retrospect-processor-v34-31881
浏览 3提问于2020-08-07得票数 0
我们今天遇到了一个问题，一个Kakfa消费者无法连接到任何代理，日志中没有错误消息。当它通过SSL连接并且指向错误的信任库时，它未能连接。但是，虽然在这种情况下，这是由于SSL失败，但我也曾见过其他情况，即使用者无法连接，日志中没有错误消息。
是否有一种方法来配置卡夫卡使用者，以便在使用者无法连接到代理？时记录错误消息--我正在寻找一个通用解决方案，而不是特定于SSL连接问题的解决方案。
(我们使用的是：org.apache.kafka:kafka-clients:0.9.0.1和org.springframework.kafka:spring-kafka:1.0.4.RELEASE)
浏览 1提问于2018-11-07得票数 3
1回答
我们有5个节点的kafka-cluster，2.3.0 Kafka版本kafka_2.12-2.3.0运行OpenJDK11。 从上周开始，集群中的一个代理突然出现故障。 我们注意到这个节点的controller.log中出现了下面的错误，不能完全确定这是什么意思，可以有人解释一下吗，谢谢 2021-01-27 11:26:59,471 INFO kafka.controller.ZkPartitionStateMachine: [PartitionStateMachine controllerId=5] Stopped partition state machine
2021-01-2
浏览 86提问于2021-02-01得票数 0
1回答
我对kafka还是个新手。我已经用java创建了一个示例生产者和消费者。使用producer，我可以将数据发送到kafka主题，但无法使用以下消费者代码获取主题中的记录数量。
public class ConsumerTests {
public static void main(String[] args) throws Exception {
BasicConfigurator.configure();
String topicName = "MobileData";
Stri
浏览 18提问于2017-08-11得票数 1
在我目前的用例中，我使用Spark从minute读取数据，并对数据进行一些处理并每1分钟发送一次给Kafka，我使用Spark和菲尼克斯维护HBase表中的CDC信息。
但是这个设计有一些问题，例如如果MS记录的激增，火花处理需要比批处理时间更长的时间，并最终将重复的记录发送给Kafka。
作为替代，我正在考虑使用Kafka读取来自MS的消息，并将记录发送到Kafka主题，并在Kafka中维护MS。星火流将读取卡夫卡主题的记录，并将记录和存储处理到HBase，并发送到其他卡夫卡主题。
为了实现这个体系结构，我有几个问题：
我能用开源的Kafka连接器和ApacheKafka0.9版本来实现
浏览 0提问于2017-09-11得票数 1
回答已采纳
1回答
我正在使用SparkSession.readStream进行结构化流处理，并将其写入蜂窝表，但它似乎不允许我使用基于时间的微批，也就是说，我需要一批5秒的时间。所有的消息都应该形成一个5秒的批处理，并且批处理数据应该被写入到hive表中。
现在，当他们被发布到Kafka主题时，它会读取这些信息，每条信息都是表的一条记录。
工作代码
def hive_write_batch_data(data, batchId):
data.write.format("parquet").mode("append").saveAsTable("test.my_ta
浏览 12提问于2022-04-25得票数 0
回答已采纳
我正在阅读Connect API上的Kafka文档，并试图将其与我的问题域联系起来。我有多个数据库，其中有公共表，我需要在任何更新/插入/删除时同步这些表。一个例子是：
1)有人在order_history中更新表“DB1”--我希望将更新传递给DB2/DB3等。
2)有人将记录插入到"purchase_order“中--我希望将插入传递给DB2/DB3等，以便在那些DB2/DB3等中进行相同的插入。
3)表将出现在所有DBs中，因此不存在缺少的表问题。
这些只能在一组特定的表上完成，而不是整个数据库。我从Connect API文档中了解到，我需要提供以下内容：
1)源连接器将数据从S
浏览 2提问于2017-09-11得票数 0
我正在使用confluent kafka python库来使用消息，我希望在民意测验中获得多条记录。但不知怎么的，我一次只得到一张唱片。
from confluent_kafka import Consumer
c = Consumer({
'bootstrap.servers': '127.0.0.1:9092',
'group.id': 'mygroup',
'session.timeout.ms': 6000,
'auto.offset.reset':
浏览 0提问于2018-12-26得票数 0
1回答
我已经将我的Postgres DB连接到Kafka，以读取表中的新记录，并将它们推送到elasticsearch。数据库已经有一些记录，当我将kafka连接到数据库时，这些记录在kafka主题中可见(使用./kafka-console-consumer.sh --topic postgres.public.table --bootstrap-server kafka:9092 --from-beginning)。 但是使用下面的代码片段，我只能读取表中的新记录。 if __name__ == "__main__":
if es.indices.exists(
浏览 71提问于2021-10-07得票数 0
1回答
我们有一个源连接器，从rdbms读取并放入kafka。它使用模式注册表和avro模式。我在kafka连接日志和模式注册表日志中分别发现了以下异常。
1.
Committing offsets (org.apache.kafka.connect.runtime.WorkerSourceTask:426)
WorkerSourceTask{id=A-0} flushing 0 outstanding messages for offset commit (org.apache.kafka.connect.runtime.WorkerSourceTask:443)
Task threw an
浏览 19提问于2021-03-15得票数 0
为了复制这个问题，运行docker run zookeeper，然后在下面的yaml文件上运行docker-compose up。
我正在使用动物园管理员的最新图片，wurstmeister/kafka:0.9.0.0-1和。我运行了docker-compose up，并最终使它正常工作，但现在我得到了以下错误：
我搜索了git，并在流中堆叠，但没有结果。在我保存集群之前一切看起来都很好。在卡夫卡日志中我得到：
警告o.a.z.ClientCnxn -会话0x0服务器空，意外错误，关闭套接字连接并尝试重新连接kafka-manager_1 _1\ java.net.ConnectE
浏览 3提问于2018-01-15得票数 10
我正在做一些kafka流聚合，并为主题编写聚合记录，并获得以下错误。我正在为聚合助手类使用自定义的json。我在一些博客上找到的解决这个问题的方法是增加max.request.size。
虽然我将max.request大小从默认增加到401391899，但序列化的聚合消息大小在后续写入主题时不断增加。
运行流10分钟后，将显示下面的错误。不确定问题是否与我的服务器有关，或者是否应该更改除max.request.size之外的任何配置设置来解决此问题。
写到主题的信息；
{A=5, B=1, C=0, D=87, E=1, F=0.4482758620689655 }
{A=6, B=1,
浏览 2提问于2017-04-28得票数 0
回答已采纳
我正在使用Debezium SQL Server Source connector通过CDC将数据推送到Kafka。SQL Server表中的数据是连续插入的。当Kafka宕机时，数据还会插入到该表中吗？如果是这样，当我的Kafka打开时，如何获得这些记录？
浏览 13提问于2020-05-22得票数 0
回答已采纳
我是从一个Flink流媒体应用程序中的Kafka集群中阅读的。在获取源流之后，我希望通过复合键和timeEvent滚动窗口聚合事件，然后将结果写入表中。问题是，在应用我的aggregateFunction (只计算clientId的点击次数)之后，我找不到获得每个输出记录的键的方法，因为api返回一个累积结果的实例，而不是相应的键。
DataStream<Event> stream = environment.addSource(mySource)
stream.keyBy(new KeySelector<Event,Integer>() {
浏览 0提问于2019-07-08得票数 1
回答已采纳
要求:我有不同的微服务生产数据给卡夫卡。我想收集信息，进行聚合，并为报告构建表。
构建用于报告的新表将要求我从Kafka读取数据，将其与现有事实表进行比较，并在数据库中创建/更新新的roes。
我想的是使用星火流从Kafka主题中读取，并与RDBMS数据库连接。
寻找帮助来理解连接的最佳方式。
浏览 4提问于2019-12-03得票数 2
回答已采纳
我在我的应用程序中使用Apache Drill (1.14) JDBC驱动程序，它使用来自Kafka的数据。应用程序在一段时间内工作得很好，经过几次迭代后，由于以下Too many files open问题而无法执行。我确保我的代码中没有文件句柄泄漏，但仍然不确定为什么会发生这个问题？ 看起来这个问题是在构建Kafka消费者时在Apache drill库中发生的。有谁能指导我解决这个问题吗？ 当我重新启动Apache drillbit时，这个问题就消失了，但很快它又出现了。在钻取过程重启前后，我确实使用ulimit -a | wc -l & lsof -a -p <PID>
浏览 42提问于2019-06-12得票数 0
我有一个卡夫卡集群运行与3个经纪人。
<node-ip>:10092 //broker-0
<node-ip>:10093 //broker-1
<node-ip>:10094 //broker-2
broker-1 <node-ip>:10093处于未就绪状态(由于某些准备失败)。但其他两家经纪商运行良好。
但是，当我使用带有运行代理地址的脚本kafka-consumer-groups.sh作为引导服务器时，我会得到以下错误。
kafka@mirror-maker-0:/opt/kafka/bin$ /opt/kafka/bin/kafka-c
浏览 16提问于2022-03-14得票数 1
我正在使用来自具有多个分区的Kafka主题的数据来处理Flink。有没有可能在每个并行子任务/分区上都有一个窗口，而不必使用keyBy (因为我想避免混乱)。根据文档，我只能在键控窗口(需要混洗)或全局窗口(将并行度降低到1)之间进行选择。 这样做的动机是，我希望使用CountWindow来批处理具有自定义触发器的消息，该触发器也会在设置的处理时间后触发。因此，对于每个Kafka分区，我希望将N个记录一起批处理，或者在将批发送到下游之前等待X个量的处理时间。 谢谢!
浏览 25提问于2021-01-22得票数 0
1回答
我正在使用Kafka 0.10.2，为了接收Kafka的记录，我有一个消费者投票循环，如下所示：
while (true) {
ConsumerRecords<Long, String> records = consumer.poll(2000);
int count= 0;
long lasttimestamp = 0;
long lastOffset = 0;
for (ConsumerRecord<Long, String> record : records) {
if (record.timestamp()
浏览 2提问于2017-10-08得票数 0
当试图向Kafka发送消息时，计时器应用程序将引发每秒错误。
2017-08-03 10:20:58,805 ERROR kafka-producer-network-thread | producer-2 o.s.k.s.LoggingProducerListener:76 - Exception thrown when sending a message with key='null' and payload='{-1, 2, 11, 99, 111, 110, 116, 101, 110, 116, 84, 121, 112, 101, 0, 0, 0, 12,
浏览 0提问于2017-08-03得票数 0
我正在使用kafka 0.9.0.1 broker和0.9.0.1客户端。我的使用者实例正在消耗处理时间小于1秒的记录。和其他主要的秘密是
enable.auto.commit=false
session.timeout.ms=30000
heartbeat.interval.ms=25000
处理后，我将提交偏移。我得到了例外
为组UNKNOWN_MEMBER_ID提交偏移时发生错误kafka_to_s3
错误com.bsb.hike.analytics.consumer.Consumer -由于组重新平衡，无法提交retryCount=2 retryCount=2提交无法完成
一小
浏览 1提问于2016-09-28得票数 2
我正在使用Ubuntu 14.04。A nodejs ->kafkaesque(kafka nodejs客户端)->kafka_2.9.2-0.8.2.1->风暴.在Kafkaesque (kafka客户端)中，我注意到在api.js中，有一行可以创建套接字。
_socket = net.createConnection(_options.port, _options.host);
我想知道套接字超时的默认值是多少。如果我想要套接字连接总是在那里/永久的，我应该setTimeout(0)吗？
我还注意到代码后面的一行：
_socket.on('timeout'
浏览 3提问于2015-05-20得票数 0
回答已采纳
1回答
我有一个消费者，它消耗了一个话题。主题有6个分区。分配给组的单个使用者。我做投票，如下面的Consumer.poll(10000)，我退出消费者获取时，没有记录返回。
从文档中，我相信当没有要使用的记录和持续时间10000足以重新平衡和获取记录时，轮询返回为空。多数时间轮询使用所有分区的记录，但有时轮询从3个分区获取记录，然后返回空记录，而不消耗其他3个分区。
顺便说一下，我使用了2.0.1Kafka客户端，Kafka服务器版本是2.11-2.2.0。
任何人都知道为什么我的使用者跳过其他分区并返回空的records.what来使用所有分区。
浏览 1提问于2020-01-23得票数 1
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
腾讯会议
实时音视频活动推荐
腾讯云开发者
