Michael Rylander created KAFKA-5980:
---------------------------------------
Summary: FailOnInvalidTimestamp does not log error
Key: KAFKA-5980
URL: https://issues.apache.org/jira/browse/KAFKA-5980
Project: Kafka
Issue Type: Bug
Affects Versions: 0.10.2.1
Reporter: Michael Rylander
In this block of code from the StreamThread class, an error caused by an
invalid timestamp is re-thrown and the comment says that it should already be
logged, however this particular error is NOT logged anywhere and only shows up
in standard error.
{code}
/**
* Execute the stream processors
*
* @throws KafkaException for any Kafka-related exceptions
* @throws Exception for any other non-Kafka exceptions
*/
@Override
public void run() {
log.info("{} Starting", logPrefix);
try {
runLoop();
cleanRun = true;
} catch (KafkaException e) {
// just re-throw the exception as it should be logged already
throw e;
} catch (Exception e) {
// we have caught all Kafka related exceptions, and other runtime
exceptions
// should be due to user application errors
log.error("{} Streams application error during processing: ",
logPrefix, e);
throw e;
} finally {
shutdown();
}
}
{code}
Message from Standard Error.
{code}
Exception in thread "StreamThread-1"
org.apache.kafka.streams.errors.StreamsException: Input record
ConsumerRecord(...) has invalid (negative) timestamp. Possibly because a
pre-0.10 producer client was used to write this record to Kafka without
embedding a timestamp, or because the input topic was created before upgrading
the Kafka cluster to 0.10+. Use a different TimestampExtractor to process this
data.
at
org.apache.kafka.streams.processor.FailOnInvalidTimestamp.onInvalidTimestamp(FailOnInvalidTimestamp.java:62)
at
org.apache.kafka.streams.processor.ExtractRecordMetadataTimestamp.extract(ExtractRecordMetadataTimestamp.java:60)
at
org.apache.kafka.streams.processor.FailOnInvalidTimestamp.extract(FailOnInvalidTimestamp.java:46)
at
org.apache.kafka.streams.processor.internals.RecordQueue.addRawRecords(RecordQueue.java:86)
at
org.apache.kafka.streams.processor.internals.PartitionGroup.addRawRecords(PartitionGroup.java:117)
at
org.apache.kafka.streams.processor.internals.StreamTask.addRecords(StreamTask.java:158)
at
org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:605)
at
org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:361)
{code}
--
This message was sent by Atlassian JIRA
(v6.4.14#64029)
