如何处理Kafka streams中的处理异常？
Kafka Streams是一个用于构建实时流处理应用程序的客户端库。在处理Kafka Streams中的异常时，可以采取以下几种方法：
- 异常处理器（Exception Handler）：Kafka Streams提供了异常处理器接口，可以通过实现该接口来自定义异常处理逻辑。异常处理器可以捕获和处理在流处理过程中发生的异常，例如记录日志、发送警报或执行特定的错误处理逻辑。可以通过
StreamsBuilder#stream()方法的#withExceptionHandler()来设置异常处理器。 - 容错机制：Kafka Streams具有内置的容错机制，可以自动处理一些异常情况。例如，当Kafka集群中的一个或多个分区不可用时，Kafka Streams会自动尝试重新连接并恢复处理。此外，Kafka Streams还支持状态恢复，可以在应用程序重新启动后恢复之前的处理状态。
- 监控和报警：为了及时发现和处理异常情况，可以使用监控和报警工具来监控Kafka Streams应用程序的运行状态。例如，可以使用Prometheus和Grafana等工具来收集和可视化指标数据，并设置警报规则以便在异常情况发生时及时通知相关人员。
- 单元测试和集成测试：为了确保Kafka Streams应用程序的正确性和稳定性，可以编写单元测试和集成测试来覆盖各种场景和异常情况。通过模拟输入数据和预期输出，可以验证应用程序在处理异常情况时的行为是否符合预期。
总结起来，处理Kafka Streams中的处理异常可以通过自定义异常处理器、利用内置的容错机制、使用监控和报警工具以及编写测试来实现。这些方法可以帮助我们及时发现和处理异常情况，确保Kafka Streams应用程序的稳定性和可靠性。
腾讯云相关产品和产品介绍链接地址：
- 异常处理器（Exception Handler）：Kafka Streams异常处理器
- 监控和报警工具：腾讯云监控服务
- 单元测试和集成测试：腾讯云测试服务
相关·内容
我正在为spring kafka streams应用程序的定制而苦苦挣扎。我一直在尝试在我的KStreams中配置处理未捕获(运行时异常)。 参考文档https://docs.spring.io/spring-cloud-stream-binder-kafka/docs/3.1.0/reference/html/spring-cloud-stream-binder-kafka.html#_kafka_streams_binder -应该这样做： @Configuration
@Slf4j
public class CustomKafkaStreamsConfiguration {
浏览 65提问于2021-01-14得票数 0
使用kafka-streams 0.10.0.0，在转发消息时，我会周期性地在StreamTask中看到一个空指针异常。它在10%到50%的调用之间变化。NPE发生在这种方法中：
public <K, V> void forward(K key, V value) {
ProcessorNode thisNode = currNode;
try {
for (ProcessorNode childNode : (List<ProcessorNode<K, V>>) thisNode.children()) {
浏览 1提问于2016-08-21得票数 3
回答已采纳
我有一个流应用程序与下面的驱动程序代码的实时消息转换.
String topicName = ...
KStreamBuilder builder = new KStreamBuilder();
KStream<String, String> source = builder.stream(topicName);
source.transform(() -> new MyTransformer()).to(...);
KafkaStreams streams = new KafkaStreams(builder, appConfig);
streams.setUncaug
浏览 17提问于2017-02-22得票数 7
回答已采纳
我正在使用kafka-clients 2.5.1，并且有一个具有以下属性的kafka streams应用程序：
properties.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, "1");
properties.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, String.valueOf(Integer.MAX_VALUE));
properties.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "15000");
prop
浏览 2提问于2021-07-02得票数 0
在自定义Transformer中，我们尝试在init(..)方法中创建和转发一些记录。根据Kafka Streams JavaDoc的说法，这应该是可能的。但是，这不起作用，我们得到以下异常： Exception in thread "my-app-0.0.16-3be1aa47-d51d-4b0c-821a-a0b850359490-StreamThread-1" org.apache.kafka.streams.errors.StreamsException: failed to initialize processor KSTREAM-TRANSFORM-0000000
浏览 7提问于2019-03-16得票数 0
回答已采纳
问题描述：
我正在创建一个spring云Kafka流应用程序。我有一个输入主题和一个输出主题，我试图使用KStream函数为输入主题应用KStream.map键值转换操作。
如果转换为null的值，则函数抛出IllegalArgumentException
我的问题是：
1:例外的原因？尽管文档中写着：“忽略带有空键或空值的输入记录”
2:处理无状态/有状态操作中的异常的最佳实践？是否应该围绕整个处理计划进行一次尝试/捕捉就足够了？或者，我应该在每个转换函数中尝试/捕获(例如。过滤，地图，加入，减少)？
任何想法都很感激。
应用程序Config：
spring:
application:
浏览 8提问于2019-10-21得票数 0
回答已采纳
1回答
我有1分钟的窗口，一些活动通过我的主题之一。我正在寻找该窗口的结果，并将其再转换一次，以获得我想要推入水槽的最终结果。这些是我的DSL调用来完成我的聚合。
val sessionProcessorStream = builder.stream("collector-prod", Consumed.`with`(Serdes.String, Serdes.String)) //[String,String]
.filter((_, value) => filterRequest(value))
.transform(valTransformer,"valTr
浏览 0提问于2018-04-02得票数 0
回答已采纳
当kafka-streams应用程序运行，Kafka突然关闭时，应用程序进入“等待”模式，用户和生产者线程发送警告日志给他们无法连接，当Kafka回来时，一切都应该(理论上)恢复正常。我正试图对这种情况发出警告，但我无法找到捕捉它并发送日志/度量的位置。我尝试了以下几点：
streams.setUncaughtExceptionHandler，但是这只发生在异常(这里不是这样)，，，扩展ProductionExceptionHandler，并将default.production.exception.handler属性更改为我的类，该类扩展了这个接口。再一次，和setUncaughtExcep
浏览 2提问于2020-08-23得票数 5
回答已采纳
当使用处理器API和KTable函数创建全局addGlobalStore时，结果存储将填充OK。但是，随后对存储内容进行迭代的尝试会导致以下异常：
Exception in thread "main" java.lang.NullPointerException at org.apache.kafka.streams.state.internals.Serializ
浏览 0提问于2017-09-01得票数 1
回答已采纳
我正在运行一个Kafka应用程序，它使用来自2个主题的数据，并将连接/合并结果输出到3个主题。kafka主题有15个分区和3个复制因子。我们有5位kafka经纪人和5位动物园管理员。我正在运行15个Kafka应用程序实例，因此每个应用程序可以有一个分区。卡夫卡版本- 0.11.0.0
在kafka流应用程序中，我得到了以下异常：
无法完成org.apache.kafka.clients.consumer.CommitFailedException:提交，因为组已经重新平衡并将分区分配给另一个成员。这意味着随后调用轮询()之间的时间要比配置的max.poll.interval.ms长，这通常意
浏览 4提问于2017-08-09得票数 6
我们有一个带有多个avroregistry的KafkaAvroSerde。在某个时候，serde尝试在1 url上注册模式时超时，但是由于它向流应用程序抛出IO异常，流线程关闭了。从kafka流应用程序的角度来看，这种类型违背了在创建avro serdes时支持多个urls的功能，因为DSL堆栈中出现的运行时异常将关闭stream Thread。几个问题：
有什么好办法处理这件事吗？
我们是否需要在应用程序逻辑中执行重试(当您简单地将主题具体化到商店中时，这可能是很棘手的)？
否则是否有一个avroserde包装器
可以用实际配置的avroRegistry urls重试吗？
当
浏览 0提问于2018-11-02得票数 2
回答已采纳
我有这个问题：
当我使用处理器API从一个主题中消费时，当在处理器中使用方法context().forward(K, V)时，Kafka Streams抛出一个空指针异常。
这是它的堆栈跟踪：
Exception in thread "StreamThread-1" java.lang.NullPointerException
at org.apache.kafka.streams.processor.internals.StreamTask.forward(StreamTask.java:336)
at org.apache.kafka.streams.processor.i
浏览 3提问于2017-02-01得票数 1
回答已采纳
我正在尝试访问在同一个java程序中创建的inMemoryStore。但是，将异常作为“线程中的异常”org.apache.kafka.streams.errors.InvalidStateStoreException:返回，状态存储storeName可能已迁移到另一个实例。
当我使用persistentKeyValueStore时，它工作得很好，能够创建存储并返回值。
package com.bakdata.streams_store.demo;
import java.util.Collection;
import java.util.Properties;
import org.ap
浏览 0提问于2019-07-12得票数 0
在运行有状态的Kafka流应用程序时，我遇到了各种不同的RocksDB磁盘I/O Stalefile异常。只有当我至少有一个KTable实现并且发生在不同的时间时，异常才会发生。我试过无数次来复制它，但一直无法复制。
应用程序/环境详细信息：
运行时：JavaKafka库：org.apache.kafka:kafka-streams:2.5.1Deployment：OpenShiftVolume类型：NFSRAM：2000 - 8000 MiBCPU：200 Millicore to 2 CoresThreads：1Partitions：1 - many
遇到的异常：
由：org.apache
浏览 1提问于2021-09-17得票数 0
所以我有一个KStream，它被反序列化为POJO，如下所示 public class FinancialMessage {
public String user_id;
public String stock_symbol;
public String exchange_id;
} 下面是Global Ktable记录的样子 public class CompanySectors {
public String company_id;
public String company_name;
public String tckr;
public String sector_cd;
}
浏览 9提问于2020-08-03得票数 0
回答已采纳
当我尝试创建一个kafka实例时，我得到了下面的异常。
Exception in thread "main" java.lang.NoSuchMethodError: org.apache.kafka.common.metrics.Sensor.add(Lorg/apache/kafka/common/MetricName;Lorg/apache/kafka/common/metrics/MeasurableStat;)Z at org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl
浏览 0提问于2018-11-26得票数 10
回答已采纳
需要一些帮助，找出我收到的一个卡夫卡流消费者的例外。
我用低级别处理器API实现了Kafka流。对于从Kafka接收到的每一次更新，都会将其合并并更新到keystore，从而维护状态。最初，我们只运行了一个消费者，过了一段时间，我们尝试提出第二个消费者。但第二个消费者在再平衡期间抛出了一个例外，称其未能实现再平衡。这是因为更改日志的状态发生了更改(下面的异常共享)。我假设，当重新平衡发生时，第一个使用者收到了一些更新，因此更新被推送到相应的更改日志中。请帮帮忙。还共享相同的示例代码。我使用的是Kafka 2_11 0.10.2.1，主题有72个分区
异常
Exception in th
浏览 2提问于2017-08-30得票数 1
回答已采纳
对于我的Kafka流应用程序之一，我需要使用DSL和处理器API的特性。我的流媒体应用程序流
source -> selectKey -> filter -> aggregate (on a window) -> sink
聚合之后，我需要向接收器发送一条聚合消息。因此，我将我的拓扑定义如下
KStreamBuilder builder = new KStreamBuilder();
KStream<String, String> source = builder.stream(source_stream);
source.selectKey(new MyK
浏览 4提问于2016-10-24得票数 6
回答已采纳
Handling exceptions in Kafka streams是一个类似的问题，但公认的答案只涉及productionException。如何处理处理过程中发生的异常，从而控制手动偏置提交。
浏览 23提问于2020-08-31得票数 1
我试图使用Scala中的ProcessorAPI运行一个非常基本的流。
class KafkaProcessor extends Processor[String, GenericRecord] {
private var kvStore: KeyValueStore[String, GenericRecord] = _
override def init(processorContext: ProcessorContext): Unit = {
this.kvStore = Stores
.keyValueStoreBuilder(
浏览 1提问于2019-01-09得票数 0
回答已采纳
这是关于升级生产中的现有代码库，这些代码库使用来自kafka-clients，kafka-streams，spring-kafka 2.4.0到2.6.x的窗口，并将spring-boot-starter-parent2.2.2.RELEASE升级到2.3.x，因为2.2与kafka-streams 2.6不兼容。
现有的代码在下面的旧版本中提到了这些bean(2.4.0，2.2春季版本)：
@Bean("DataCompressionCustomTopology")
public Topology customTopology(@Qualifier("CustomFa
浏览 20提问于2021-05-15得票数 2
我试图将事件从KStream计数到时间段：
KStream<String, VehicleEventTO> stream = builder.stream("vehicle", Consumed.with(Serdes.String(), new JsonSerde<>(VehicleEventTO.class)));
KStream<String, VehicleEventTO> streamWithKey = stream.selectKey((key, value) -> value.getId_vehicle(
浏览 1提问于2019-07-01得票数 1
回答已采纳
我正在看这里提到的例子：
我正在以json字符串的形式生成输入数据。
关于主题- PageViews
{"industry":"eng","user":"bob","page":"index.html"}
关于主题- UserProfiles
{"experience":"some","region":"europe"}
我的课看起来像：
import com.google.gson.Gson;
import com.sohi.exam
浏览 0提问于2017-05-25得票数 1
回答已采纳
1回答
在一个简单的Kafka流程序中，当我使用下面的代码时，它不会引发任何错误：
KTable<String, Long> result= source.mapValues(textLine
->textLine.toLowerCase()) .flatMapValues(lowercasedTextLine ->
Arrays.asList(lowercasedTextLine.split(" "))) .selectKey((ignoredKey,word) ->
word) .groupByKey()
浏览 0提问于2019-03-25得票数 1
我编写了一个Streams Binder应用程序，该应用程序将多个Kafka输入主题复用到一个流中：
spring:
cloud:
stream:
bindings:
process-in-0:
destination: test.topic-a,test.topic-b
(资料来源：)
但是，每当我在输入目标中设置多个主题(用逗号分隔)时，就会发生以下错误：
2022-06-17 14:07:07.648 INFO --- [-StreamThread-1] o.a.k.clients.consumer.KafkaConsumer
浏览 16提问于2022-06-17得票数 0
回答已采纳
假设我们正在KStream和KTable之间进行内部连接，如下所示：
StreamsBuilder sb = new StreamsBuilder();
JsonSerde<SensorMetaData> sensorMetaDataJsonSerde = new JsonSerde<>(SensorMetaData.class);
KTable<String, String> kTable = sb.stream("sensorMetadata",
Consum
浏览 2提问于2020-06-06得票数 0
回答已采纳
在使用kafka流时获取以下错误堆栈跟踪
更新:如per@matthias-j-sax所示，我已经用WrapperSerde的默认构造函数实现了自己的WrapperSerde，但仍然得到以下异常
org.apache.kafka.streams.errors.StreamsException: stream-thread [streams-request-count-4c239508-6abe-4901-bd56-d53987494770-StreamThread-1] Failed to rebalance.
at org.apache.kafka.streams.processor
浏览 0提问于2018-06-27得票数 3
回答已采纳
1回答
我的任务是拆卸一个Dev环境，并将其重新从废品中设置，以验证我们的CI-CD进程；唯一的问题是，我创建了一个主题，因此Kafka Streams应用程序出现了一个错误。
我研究了一下，发现了这个问题，并纠正了它，但当我深入挖掘的时候，我遇到了另一个奇怪的小问题。
我实现了一个意外的异常处理程序如下：
streams.setUncaughtExceptionHandler((t, e) -> {
logger.fatal("Caught unhandled Kafka Streams Exception:", e);
// Do some exception
浏览 0提问于2018-09-27得票数 2
回答已采纳
有什么帮助吗?如何在Spring cloud上添加状态存储 我总是收到这样的错误：“嵌套异常是无效的，无法启动流：；嵌套异常是无效的拓扑: StateStore myStore is org.apache.kafka.streams.errors.TopologyException: org.springframework.kafka.KafkaException”。 下面是bean的定义，但是它永远不会工作 @Bean
public StoreBuilder storeBuilder() {
KeyValueBytesStoreSupplier storeSupplier =
浏览 13提问于2020-03-18得票数 0
回答已采纳
我是卡夫卡的新手，目前我正在研究卡夫卡流，尤其是加入两个流。
我浏览的示例使用的是相当简单的消息/文本消息。因此，我构建了另一个简单的示例，它更适用于传统的ETL。比方说，我们有两个“数据集”：合同(=Vertrag)和现金流，基数为1到n。
在我的示例中，我为每个对象创建了一个主题，并将对象(Vertrag，现金流)发送给每个对象。
我第一次加入了他们。
KStream<String, String> joined = srcVertrag.leftJoin(srcCashflow,
(leftValue, rightValue) -> &#
浏览 1提问于2019-06-16得票数 0
我试图将目前用Java代码编写的spring应用程序的配置具体化。是将ProducerConfig和ConsumerConfig值放入spring.kafka.streams.properties，还是通过spring.kafka.producer和spring.kafka.consumer正确配置它们？
到目前为止，看来我应该把我所有的配置都放到KafkaStreamsConfiguration类型的bean中，以便配置我的kafka-streams应用程序。目前，我是通过直接在代码中设置ProducerConfig和ConsumerConfig值来做到这一点的。
当我将此配置具体化时，从P
浏览 1提问于2019-06-13得票数 3
回答已采纳
有人知道如何为卡夫卡流配置spring.json.trusted.packages吗？我使用以下配置：
@Configuration
@EnableKafka
@EnableKafkaStreams
public class KafkaStreamsConfig {
@Bean(name = KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME)
public KafkaStreamsConfiguration kStreamsConfig(KafkaProperties kafkaProperti
浏览 10提问于2022-03-25得票数 0
回答已采纳
我现在正在尝试kafka streams。在构建我的应用程序的过程中，我流式传输了一些无效事件，这些事件现在会导致反序列化过程中出现异常。 我想按照here的建议使用kafka-streams- application-reset，但我不确定在哪里可以找到真正的kafka-streams应用程序-reset？
浏览 68提问于2019-01-13得票数 0
2回答
我正在尝试使用来自kafka主题的合流avro消息，作为带有spring 2.0的Kstream。
我能够将消息作为MessageChannel来使用，而不是以KStream的形式使用。
@Input(ORGANIZATION)
KStream<String, Organization> organizationMessageChannel();
@StreamListener
public void processOrganization(@Input(KstreamBinding.ORGANIZATION)KStream<String, Organizatio
浏览 2提问于2019-11-25得票数 1
回答已采纳
我用的是Spring和卡夫卡流。我试图模拟一个错误场景。正如您从代码中看到的，如果我将输入值作为0传递给其中一个输入主题，我将得到一个divide by 0异常。
我已经将导致错误的部分封装在一个try-catch块中，假设该错误将被catch块捕获。
@Component
@Slf4j
public class Topology {
private static final Serde<String> STRING_SERDE = Serdes.String();
@Autowired
void KStreamsTopology(StreamsBuil
浏览 12提问于2022-08-11得票数 0
回答已采纳
在使用@StreamListener收听kafka主题之后，在RuntimeException、全局erroChannel或主题特定errorChannel (topic.group.errors)上没有收到任何错误消息。@ServiceActivator没有收到任何信息。
POM依赖项: Greenwich.RELEASE
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream</artifactI
浏览 1提问于2019-06-07得票数 0
回答已采纳
我正在开发一个卡夫卡流应用程序，但在覆盖默认的ProductionExceptionHandler时遇到了困难。
我已经创建了一个实现ProductionExceptionHandler的类
public class RtaCustomProcessingExceptionHandler implements ProductionExceptionHandler {
private static final Logger log = LoggerFactory.getLogger(RtaCustomProcessingExceptionHandler.class);
priva
浏览 0提问于2018-10-10得票数 2
1回答
下面的错误是在流运行一定时间后给出的？我找不到谁负责创建.sst文件？
环境规划署：
Kafka版本0.10.0-cp1
scala 2.11.8
org.apache.kafka.streams.errors.ProcessorStateException: Error while executing flush from store agg
at org.apache.kafka.streams.state.internals.RocksDBStore.flushInternal(RocksDBStore.java:424)
at org.a
浏览 11提问于2016-06-24得票数 1
2回答
我正在以json字符串的形式生成输入数据。
用于主题-我的输入
{"time":"2017-11-28T09:42:26.776Z","name":"Lane1","oclass"
:"myClass","id":112553,"Scope":"198S"}
我的班级是这样的：
public class App {
static public class CountryMessage {
public String
浏览 4提问于2017-11-30得票数 2
我在TopologyTestDriver上遇到了一个问题，因为设置属性是必须的:当拓扑试图将记录发布到主题中时，它会转到配置中提供的"url“。我如何才能模拟指向MockedSchemaRegistry的访问？
要将avro中的记录写入主题，请使用mockSchemaRegistryClient.register
还有一个问题，我如何将stateStore加载到拓扑中？我正在初始化时创建stateStore (主题已经创建)
我的亲信：
testImplementation("org.junit.jupiter:junit-jupiter-api:5.3.1")
浏览 1提问于2018-10-17得票数 2
回答已采纳
我用kafka streams写了这个应用程序。它不时地抛出InvalidPidMappingException。
Caused by: org.apache.kafka.common.KafkaException: org.apache.kafka.common.errors.InvalidPidMappingException: The producer attempted to use a producer id which is not currently assigned to its transactional id.
我有这段代码片段，它为应用程序设置了一个未被重视的异常处理程序
浏览 6提问于2021-04-19得票数 0
我们在kafka中使用Ktabke进行聚合，它是非常基本的用法，并参考了kafka文档。 我只是试图调查一下，如果一些消息消费失败，同时聚合我们如何将这些消息移动到错误主题或dlq。 我为KStream找到了类似的东西，但没有找到用于ktable的东西，我也不能简单地将KStream解决方案扩展到KTable。 KStream Handling bad messages using Kafka's Streams API参考 我的用例非常简单，对于任何类型的异常，只需移动到错误主题并移动到不同的消息
浏览 11提问于2019-04-05得票数 0
使用的build.sbt文件如下所示：
scalaVersion := "2.12.1"
val kafka_streams_version = "2.0.0"
libraryDependencies ++= Seq(
"org.apache.kafka" % "kafka-streams" % kafka_streams_version % Provided,
"org.apache.kafka" % "kafka-clients" % kafka_streams_vers
浏览 4提问于2020-01-11得票数 0
在Kafka streams中，我的kv商店链接到一个接收器，该接收器将记录发送到输出主题。-如果由于某种原因，接收器不能将记录提交到主题，我们会得到什么异常？
浏览 0提问于2017-08-17得票数 0
2回答
我正在运行一个有7个节点和大量流处理的Kafka集群。现在我在我的Kafka Streams应用程序中看到了不常见的错误，比如在高输入率下：
[2018-07-23 14:44:24,351] ERROR task [0_5] Error sending record to topic topic-name. No more offsets will be recorded for this task and the exception will eventually be thrown (org.apache.kafka.streams.processor.internals.RecordC
浏览 0提问于2018-07-24得票数 1
1回答
我面临Spring的一个问题，那就是它不能从process事件访问状态存储，我将这个特定的存储添加到拓扑/流中。
方法1：
@Component
@RequiredArgsConstructor
@EnableKafkaStreams
@Order(2)
public class TimelineVersionUpdatedStream implements EventStream {
private static final Logger logger =
LoggerFactory.getLogger(TimelineVersionUpdatedStrea
浏览 7提问于2022-02-24得票数 0
回答已采纳
启动流应用程序(使用卡夫卡流)失败时使用"java.lang.IllegalStateException:这不应该发生，因为headers()应该只在处理记录时调用“
这似乎只有当我启动应用程序时，在主题中已经有数据时才会发生。如果主题是空的，并且我开始将数据推送给它，那么一切都很好。
有人知道为什么会发生这种事吗？
谢谢
This should not happen as headers() should only be called while a record is processed
java.lang.IllegalStateException: This should no
浏览 0提问于2019-05-29得票数 1
1回答
当我试图实现一个商店时，我收到了一个异常。我正在运行Kafka 1.0、Confluent的SchemaRegistry4.0和Avro 1.8.2。我已经用Avro的maven插件生成了Pojo，并将模式部署到具有Confluent插件的汇合服务器上。我能够为STREAM1主题生成一条消息。下面是设置流的代码：
Properties properties = new Properties();
properties.put(StreamsConfig.APPLICATION_ID_CONFIG, "streams-pipe");
properties.put(StreamsC
浏览 0提问于2018-03-02得票数 3
回答已采纳
1回答
卡夫卡流连接表太大
、、、
我想使用在KStream上执行一个左联接，并使用一个KTable将表的一些字段添加到流中。
对于包含所有相关条目的表的更小版本(大约1,300项)，一切都很好。
一旦我使用了整个表(大约200,000个条目)，就会在行中得到一个NullPointerException，其中包含KTable的Avro (GenericRecord)的相关字段。
当我在KSQL中执行相同的左联接操作时，表中添加的字段为NULL。表中存在相关的连接键，但是在KSQL中查询它们需要大约。20秒后他们才会出现。
表是否可能太大，无法执行左联接？如果是这样的话，我能做些什么让联接使用整个表吗？
出于开发目的，我目前使用的
浏览 0提问于2019-08-08得票数 0
我有两个broker 1.0.0 kafka集群，我正在针对这个kafka运行1.0.0 kafka流API应用程序，我将producer request.timeout.ms增加到5分钟来修复producer Timeoutexception。
目前，在运行了一段时间之后，我得到了两种类型的异常。我正试图按照中的建议修复这些异常，但这里有不完整的解决方案。此解决方案是否值得推荐(降低生产者batch.size)。请帮帮忙。异常1
2017-12-08 13:11:55,129 ERROR o.a.k.s.p.i.RecordCollectorImpl [sample-app-0.0.1-15
浏览 16提问于2017-12-11得票数 1
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
实时音视频
对象存储
即时通信 IM活动推荐
腾讯云开发者
