{"action":"created","issue":{"url":"https://api.github.com/repos/SOHU-Co/kafka-node/issues/761","repository_url":"https://api.github.com/repos/SOHU-Co/kafka-node","labels_url":"https://api.github.com/repos/SOHU-Co/kafka-node/issues/761/labels{/name}","comments_url":"https://api.github.com/repos/SOHU-Co/kafka-node/issues/761/comments","events_url":"https://api.github.com/repos/SOHU-Co/kafka-node/issues/761/events","html_url":"https://github.com/SOHU-Co/kafka-node/issues/761","id":256185998,"number":761,"title":"HighLevelProducer does not work embedding a timestamp.","user":{"login":"kty1965","id":1715578,"avatar_url":"https://avatars3.githubusercontent.com/u/1715578?v=4","gravatar_id":"","url":"https://api.github.com/users/kty1965","html_url":"https://github.com/kty1965","followers_url":"https://api.github.com/users/kty1965/followers","following_url":"https://api.github.com/users/kty1965/following{/other_user}","gists_url":"https://api.github.com/users/kty1965/gists{/gist_id}","starred_url":"https://api.github.com/users/kty1965/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/kty1965/subscriptions","organizations_url":"https://api.github.com/users/kty1965/orgs","repos_url":"https://api.github.com/users/kty1965/repos","events_url":"https://api.github.com/users/kty1965/events{/privacy}","received_events_url":"https://api.github.com/users/kty1965/received_events","type":"User","site_admin":false},"labels":[],"state":"open","locked":false,"assignee":null,"assignees":[],"milestone":null,"comments":1,"created_at":"2017-09-08T08:58:47Z","updated_at":"2018-02-13T08:00:42Z","closed_at":null,"author_association":"NONE","body":"## Bug Report\r\nHighLevelProducer does not work embedding a timestamp.\r\n\r\n### Environment\r\n\r\n* Node version: v6.1.0\r\n* Kafka-node version: v2.2.2\r\n* Kafka version: 0.10.0.2\r\n\r\n### For specific cases also provide\r\n* Number of Brokers: 1\r\n* Number partitions for topic: 10\r\n\r\n### Include Sample Code to produce behavior\r\n\r\n```js\r\nproducer.send([\r\n  {\r\n    topic: topic,\r\n    messages: messages,\r\n    timestamp: Date.now()\r\n  }\r\n], cb))\r\n```\r\n\r\n### Include Sample Code to consumer behavior\r\n\r\n```java\r\nProperties config = new Properties();\r\n    config.put(StreamsConfig.APPLICATION_ID_CONFIG, \"test\");\r\n    config.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, \"borker1:9092\");\r\n    config.put(StreamsConfig.KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());\r\n    config.put(StreamsConfig.VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());\r\n    StringSerializer stringSerializer = new StringSerializer();\r\n    StringDeserializer stringDeserializer = new StringDeserializer();\r\n\r\n    WindowedSerializer<String> windowedSerializer = new WindowedSerializer<>(stringSerializer);\r\n    WindowedDeserializer<String> windowedDeserializer = new WindowedDeserializer<>(stringDeserializer);\r\n    Serde<Windowed<String>> windowedSerde = Serdes.serdeFrom(windowedSerializer,windowedDeserializer);\r\n\r\n    KStreamBuilder builder = new KStreamBuilder();\r\n    KStream<String, String> textLines = builder.stream(\"newSignals\");\r\n\r\n    KStream<Windowed<String>, Long> wordCounts = textLines\r\n        .mapValues(textLine -> {\r\n          System.out.println(textLine);\r\n          return textLine.toLowerCase();\r\n        })\r\n        .groupBy((key, word) -> {\r\n          System.out.println(word.split(\"\\\\W+\")[2]);\r\n          return word.split(\"\\\\W+\")[2];\r\n        })\r\n        .count(TimeWindows.of(60000), \"Counts\")\r\n        .toStream();\r\n    wordCounts.to(windowedSerde, Serdes.Long(), \"test\");\r\n\r\n    KafkaStreams streams = new KafkaStreams(builder, config);\r\n    streams.start();\r\n```\r\n\r\n### Include output with [Debug](https://github.com/SOHU-Co/kafka-node#how-do-i-debug-an-issue) turned on\r\n\r\n```\r\n\r\n```\r\n\r\n### Include output Java concumer log\r\n\r\n```\r\nException in thread \"StreamThread-1\" org.apache.kafka.streams.errors.StreamsException: Input record ConsumerRecord(topic = newSignals, partition = 0, offset = 0, CreateTime = -1, checksum = 2486091252, serialized key size = -1, serialized value size = 126, key = null, value = {\"device_id\":\"12345\",\"rssi\":-21,\"ts\":1504859056569692,\"square\":\"1234\",\"shop\":\"1234\"}) 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.\r\n\tat org.apache.kafka.streams.processor.FailOnInvalidTimestamp.onInvalidTimestamp(FailOnInvalidTimestamp.java:62)\r\n\tat org.apache.kafka.streams.processor.ExtractRecordMetadataTimestamp.extract(ExtractRecordMetadataTimestamp.java:60)\r\n\tat org.apache.kafka.streams.processor.FailOnInvalidTimestamp.extract(FailOnInvalidTimestamp.java:46)\r\n\tat org.apache.kafka.streams.processor.internals.RecordQueue.addRawRecords(RecordQueue.java:86)\r\n\tat org.apache.kafka.streams.processor.internals.PartitionGroup.addRawRecords(PartitionGroup.java:117)\r\n\tat org.apache.kafka.streams.processor.internals.StreamTask.addRecords(StreamTask.java:158)\r\n\tat org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:619)\r\n\tat org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:368)\r\n```\r\n\r\nThanks for your contribution!"},"comment":{"url":"https://api.github.com/repos/SOHU-Co/kafka-node/issues/comments/365179931","html_url":"https://github.com/SOHU-Co/kafka-node/issues/761#issuecomment-365179931","issue_url":"https://api.github.com/repos/SOHU-Co/kafka-node/issues/761","id":365179931,"user":{"login":"sreenivasulun","id":12527084,"avatar_url":"https://avatars0.githubusercontent.com/u/12527084?v=4","gravatar_id":"","url":"https://api.github.com/users/sreenivasulun","html_url":"https://github.com/sreenivasulun","followers_url":"https://api.github.com/users/sreenivasulun/followers","following_url":"https://api.github.com/users/sreenivasulun/following{/other_user}","gists_url":"https://api.github.com/users/sreenivasulun/gists{/gist_id}","starred_url":"https://api.github.com/users/sreenivasulun/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/sreenivasulun/subscriptions","organizations_url":"https://api.github.com/users/sreenivasulun/orgs","repos_url":"https://api.github.com/users/sreenivasulun/repos","events_url":"https://api.github.com/users/sreenivasulun/events{/privacy}","received_events_url":"https://api.github.com/users/sreenivasulun/received_events","type":"User","site_admin":false},"created_at":"2018-02-13T08:00:42Z","updated_at":"2018-02-13T08:00:42Z","author_association":"NONE","body":"I am also getting same issue now. Java streams expecting timestamp, but this node module is putting timestamp as -1. Please let me anybody has solution for this"}}
{"actor":{"display_login":"sreenivasulun"}}