zoukankan      html  css  js  c++  java
  • KafkaStream时间戳问题CreateTime = -1引起的程序中断

    Exception in thread "app-8835188a-e0a0-46da-ac2a-6820ec197628-StreamThread-1" org.apache.kafka.streams.errors.StreamsException: Input record ConsumerRecord(topic = raw_103, partition = 1, offset = 7032668, CreateTime = -1, serialized key size = -1, serialized value size = 111, headers = RecordHeaders(headers = [], isReadOnly = false), key = null, value = { "key1": [ 103, "4113471085724846255", "--", "2018-04-17 21:33:53" ], "key2": [ [ 213309, "--", 20128, 1 ] ] }) 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:73)
            at org.apache.kafka.streams.processor.ExtractRecordMetadataTimestamp.extract(ExtractRecordMetadataTimestamp.java:61)
            at org.apache.kafka.streams.processor.FailOnInvalidTimestamp.extract(FailOnInvalidTimestamp.java:48)
            at org.apache.kafka.streams.processor.internals.RecordQueue.addRawRecords(RecordQueue.java:98)
            at org.apache.kafka.streams.processor.internals.PartitionGroup.addRawRecords(PartitionGroup.java:117)
            at org.apache.kafka.streams.processor.internals.StreamTask.addRecords(StreamTask.java:560)
            at org.apache.kafka.streams.processor.internals.StreamThread.addRecordsToTasks(StreamThread.java:896)
            at org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:797)
            at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:750)
            at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:720)
    
    

    之前直接改了源码。后来从度娘中找到解决方法:
    新增时间异常捕获类MyEventTimeExtractor.class, 直接返回0

    import org.apache.kafka.clients.consumer.ConsumerRecord;
    import org.apache.kafka.streams.processor.TimestampExtractor;
    
    public class MyEventTimeExtractor implements TimestampExtractor{
    	@Override
    	public long extract(ConsumerRecord<Object, Object> record,
    			long previousTimestamp) {
    		return 0;
    	}
    }
    

    然后在属性添加下面配置:

    props.put(StreamsConfig.DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG, MyEventTimeExtractor.class);
    

    编译执行,ok

  • 相关阅读:
    racle wm_concat(column)函数的使用
    位运算应用口诀和实例
    关于Oracle中查询的数字值的显示格式需要保留小数点后两位(或者三位,及其他位数)
    SQL触发器实例讲解
    编写更好的jQuery代码的建议
    Jquery插件开发学习
    Java中super的几种用法并与this的区别
    全面总结Java泛型
    Oracle统计函数之Lead
    VBscript读取excel表,中数据导入到SQLserver数据库里面
  • 原文地址:https://www.cnblogs.com/30go/p/8876877.html
Copyright © 2011-2022 走看看