You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "Michał Kowol (Jira)" <ji...@apache.org> on 2020/07/15 07:42:00 UTC
[jira] [Comment Edited] (KAFKA-10205) NullPointerException in
StreamTask (Kafka Streams 2.5.0)
[ https://issues.apache.org/jira/browse/KAFKA-10205?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17157966#comment-17157966 ]
Michał Kowol edited comment on KAFKA-10205 at 7/15/20, 7:41 AM:
----------------------------------------------------------------
We had the same problem - we tried 2.4.0 and 2.5.0 with the same result.
I followed [~vvcephei] advice and we changed our code to have deterministic streams creation order.
Old code (we experienced that issue):
{code:java}
@Slf4j
@RequiredArgsConstructor
@Configuration
public class Streams {
private final StreamsBuilder builder;
private final EventsSerdeFactory serdeFactory;
@Bean
public KStream<UserId, String> eventsRaw() {
return builder.stream(EVENTS_RAW, serdeFactory.stringConsumer())
.peek((key, value) -> log.debug("Processing [topic={}, key={} value={}]", EVENTS_RAW, key, value));
}
@Bean
public KStream<UserId, String> events() {
return builder.stream(EVENTS, serdeFactory.stringConsumer())
.peek((key, value) -> log.debug("Processing [topic={}, key={} value={}]", EVENTS, key, value));
}
@Bean
public KStream<UserId, EventPackageContainer> eventsAvro() {
return builder.stream(EVENTS_AVRO, serdeFactory.avroConsumer(EventPackageContainer.class))
.peek((key, value) -> log.debug("Processing [topic={}, key={} value={}]", EVENTS_AVRO, key, value));
}
@Bean
public KStream<UserId, Error> errors() {
return builder.stream(ERRORS, serdeFactory.avroConsumer(Error.class))
.peek((key, value) -> log.debug("Processing [topic={}, key={} value={}]", ERRORS, key, value));
}
}
{code}
New code (works well):
{code:java}
@Slf4j
@Component
public class Streams {
private final KStream<UserId, String> events;
private final KStream<UserId, EventPackageContainer> eventsAvro;
private final KStream<UserId, String> eventsRaw;
private final KStream<UserId, Error> errors;
public Streams(StreamsBuilder builder, EventsSerdeFactory serdeFactory) {
events = createStream(builder, EVENTS, serdeFactory.stringConsumer());
errors = createStream(builder, ERRORS, serdeFactory.avroConsumer(Error.class));
eventsRaw = createStream(builder, EVENTS_RAW, serdeFactory.stringConsumer());
eventsAvro = createStream(builder, EVENTS_AVRO, serdeFactory.avroConsumer(EventPackageContainer.class));
}
public KStream<UserId, String> eventsRaw() {
return eventsRaw;
}
public KStream<UserId, String> events() {
return events;
}
public KStream<UserId, EventPackageContainer> eventsAvro() {
return eventsAvro;
}
public KStream<UserId, Error> errors() {
return errors;
}
private <K, V> KStream<K, V> createStream(StreamsBuilder builder, String topic, Consumed<K, V> consumed) {
return builder.stream(topic, consumed)
.peek((key, value) -> log.debug("Processing [topic={}, key={} value={}]", topic, key, value));
}
}
{code}
was (Author: kowolm):
We had the same problem - we try 2.4.0 and 2.5.0 with the same result.
I followed [~vvcephei] advice and we changed our code to have deterministic streams creation order.
Old code (we experienced that issue):
{code:java}
@Slf4j
@RequiredArgsConstructor
@Configuration
public class Streams {
private final StreamsBuilder builder;
private final EventsSerdeFactory serdeFactory;
@Bean
public KStream<UserId, String> eventsRaw() {
return builder.stream(EVENTS_RAW, serdeFactory.stringConsumer())
.peek((key, value) -> log.debug("Processing [topic={}, key={} value={}]", EVENTS_RAW, key, value));
}
@Bean
public KStream<UserId, String> events() {
return builder.stream(EVENTS, serdeFactory.stringConsumer())
.peek((key, value) -> log.debug("Processing [topic={}, key={} value={}]", EVENTS, key, value));
}
@Bean
public KStream<UserId, EventPackageContainer> eventsAvro() {
return builder.stream(EVENTS_AVRO, serdeFactory.avroConsumer(EventPackageContainer.class))
.peek((key, value) -> log.debug("Processing [topic={}, key={} value={}]", EVENTS_AVRO, key, value));
}
@Bean
public KStream<UserId, Error> errors() {
return builder.stream(ERRORS, serdeFactory.avroConsumer(Error.class))
.peek((key, value) -> log.debug("Processing [topic={}, key={} value={}]", ERRORS, key, value));
}
}
{code}
New code (works well):
{code:java}
@Slf4j
@Component
public class Streams {
private final KStream<UserId, String> events;
private final KStream<UserId, EventPackageContainer> eventsAvro;
private final KStream<UserId, String> eventsRaw;
private final KStream<UserId, Error> errors;
public Streams(StreamsBuilder builder, EventsSerdeFactory serdeFactory) {
events = createStream(builder, EVENTS, serdeFactory.stringConsumer());
errors = createStream(builder, ERRORS, serdeFactory.avroConsumer(Error.class));
eventsRaw = createStream(builder, EVENTS_RAW, serdeFactory.stringConsumer());
eventsAvro = createStream(builder, EVENTS_AVRO, serdeFactory.avroConsumer(EventPackageContainer.class));
}
public KStream<UserId, String> eventsRaw() {
return eventsRaw;
}
public KStream<UserId, String> events() {
return events;
}
public KStream<UserId, EventPackageContainer> eventsAvro() {
return eventsAvro;
}
public KStream<UserId, Error> errors() {
return errors;
}
private <K, V> KStream<K, V> createStream(StreamsBuilder builder, String topic, Consumed<K, V> consumed) {
return builder.stream(topic, consumed)
.peek((key, value) -> log.debug("Processing [topic={}, key={} value={}]", topic, key, value));
}
}
{code}
> NullPointerException in StreamTask (Kafka Streams 2.5.0)
> --------------------------------------------------------
>
> Key: KAFKA-10205
> URL: https://issues.apache.org/jira/browse/KAFKA-10205
> Project: Kafka
> Issue Type: Bug
> Components: streams
> Affects Versions: 2.5.0
> Reporter: Brian Forkan
> Assignee: John Roesler
> Priority: Minor
>
> In our Kafka Streams application we have been experiencing a NullPointerException when deploying a new version of our application. This does not happen during a normal rolling restart.
> The exception is:
> {code:java}
> Error caught during partition assignment, will abort the current process and re-throw at the end of rebalance","stack_trace":"java.lang.NullPointerException: nullError caught during partition assignment, will abort the current process and re-throw at the end of rebalance","stack_trace":"java.lang.NullPointerException: null at org.apache.kafka.streams.processor.internals.StreamTask.<init>(StreamTask.java:186) at org.apache.kafka.streams.processor.internals.StreamTask.<init>(StreamTask.java:115) at org.apache.kafka.streams.processor.internals.StreamThread$TaskCreator.createTask(StreamThread.java:352) at org.apache.kafka.streams.processor.internals.StreamThread$TaskCreator.createTask(StreamThread.java:310) at org.apache.kafka.streams.processor.internals.StreamThread$AbstractTaskCreator.createTasks(StreamThread.java:295) at org.apache.kafka.streams.processor.internals.TaskManager.addNewActiveTasks(TaskManager.java:160) at org.apache.kafka.streams.processor.internals.TaskManager.createTasks(TaskManager.java:120) at org.apache.kafka.streams.processor.internals.StreamsRebalanceListener.onPartitionsAssigned(StreamsRebalanceListener.java:77) at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.invokePartitionsAssigned(ConsumerCoordinator.java:278) at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.onJoinComplete(ConsumerCoordinator.java:419) at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.joinGroupIfNeeded(AbstractCoordinator.java:439) at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureActiveGroup(AbstractCoordinator.java:358) at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.poll(ConsumerCoordinator.java:490) at org.apache.kafka.clients.consumer.KafkaConsumer.updateAssignmentMetadataIfNeeded(KafkaConsumer.java:1275) at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1243) at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1173) at brave.kafka.clients.TracingConsumer.poll(TracingConsumer.java:86) at brave.kafka.clients.TracingConsumer.poll(TracingConsumer.java:80) at org.apache.kafka.streams.processor.internals.StreamThread.pollRequests(StreamThread.java:853) at org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:753) at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:697) at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:670)
> {code}
> And the relevant lines of code - [https://github.com/apache/kafka/blob/2.5/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java#L184-L196]
> I suspect "topology.source(partition.topic());" is returning null.
> Has anyone experienced this issue before? I suspect there is a problem with our topology but I can't replicate this on my machine so I can't tell.
>
--
This message was sent by Atlassian Jira
(v8.3.4#803005)