Kafka Stream GlobalKTable在特定字段上加入

Question

所以我有一个KStream，它被反序列化为类似的POJO

public class FinancialMessage {
public String user_id;
public String stock_symbol;
public String exchange_id;
}

以下是全球Ktable记录看起来像

public class CompanySectors {
public String company_id;
public String company_name;
public String tckr;
public String sector_cd;
}

我希望能够将KStream的stock_symbol字段与Ktable的tckr字段连接起来。这可能吗？我想在将一个新的EnrichedMessage对象流式传输到另一个主题之前创建它。我有下面这样的代码，但我似乎得到了一些空指针异常。

Exception in thread "trade-enrichment-stream-0c7e7782-4217-4450-8086-21871b4ebc45-StreamThread-1" java.lang.NullPointerException
at com.domain.EnrichedMessage.<init>(EnrichedMessage.java:51)
at com.domain.TradeEnrichmentTopology.lambda$3(TradeEnrichmentTopology.java:73)
at org.apache.kafka.streams.kstream.internals.KStreamKTableJoinProcessor.process(KStreamKTableJoinProcessor.java:79)
at org.apache.kafka.streams.processor.internals.ProcessorNode.lambda$process$2(ProcessorNode.java:142)
at org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.maybeMeasureLatency(StreamsMetricsImpl.java:806)
at org.apache.kafka.streams.processor.internals.ProcessorNode.process(ProcessorNode.java:142)
at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:201)
at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:180)
at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:133)
at org.apache.kafka.streams.processor.internals.SourceNode.process(SourceNode.java:101)
at org.apache.kafka.streams.processor.internals.StreamTask.lambda$process$3(StreamTask.java:383)
at org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.maybeMeasureLatency(StreamsMetricsImpl.java:801)
at org.apache.kafka.streams.processor.internals.StreamTask.process(StreamTask.java:383)
at org.apache.kafka.streams.processor.internals.AssignedStreamsTasks.process(AssignedStreamsTasks.java:475)
at org.apache.kafka.streams.processor.internals.TaskManager.process(TaskManager.java:550)
at org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:802)
at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:697)
at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:670)

以下是代码片段的样子。

KStream<String, FinancialMessage> financialMessageStream =
builder.stream(
INCOMING_TOPIC,
Consumed.with(Serdes.String(), financialMessageSerde)
);
GlobalKTable<String, CompanySectors> companySectorsStore = 
builder.globalTable(
KTABLE_TOPIC,
Consumed.with(Serdes.String(), companySectorsSerde)
);

KStream<String, EnrichedMessage> enrichedStream = financialMessageStream.leftJoin(
companySectorsStore,
(financialMessageKey, financialMessageValue) -> financialMessageValue.stock_symbol,
(financialMessageValue, companySectorsValue) -> new EnrichedMessage(financialMessageValue, companySectorsValue)
);

enrichedStream.to(
OUTGOING_TOPIC,
Produced.with(Serdes.String(), enrichedMessageSerde));

我想我的leftJoin逻辑中可能有一些错误。

Answer 1

执行左联接时，可以假设左流的记录不是null；但是，您不能假设正确的GlobalKTable将有一个与给定密钥匹配的记录，因此生成的记录可能为null。在您的情况下，当您实例化new EnrichedMessage(financialMessageValue, companySectorsValue)时，您确定companySectorsValue不是null吗？如果它为空，您是否正确处理了它？您的NPE似乎发生在EnrichedMessage的构造函数中，所以只需确保您知道companySectorsValue可以为null。

此外，请确保在发生任何联接逻辑之前预先填充GlobalKTable。