Fix source record timestamp -- seconds to milliseconds (#26)

* Update redis source timestamp from seconds to milliseconds (#1) * updated source record converter timestamp * removed logging * updated version and changelog * fixed changelog message improvement Co-authored-by: Jared Petersen <[email protected]>
jaredpetersen · Mar 16, 2022 · 5341700 · 5341700
1 parent 69baf71
commit 5341700
Show file tree

Hide file tree

Showing 4 changed files with 7 additions and 3 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -4,6 +4,10 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [1.2.3] - 2022-03-15
+### Fixed
+- Fixed a bug with record timestamps for the source connector -- switched from epoch seconds (incorrect) to epoch milliseconds
+
 ## [1.2.2] - 2021-07-22
 ### Changed
 - Use capitalization in log messages

diff --git a/pom.xml b/pom.xml
@@ -5,7 +5,7 @@
 
   <groupId>io.github.jaredpetersen</groupId>
   <artifactId>kafka-connect-redis</artifactId>
-  <version>1.2.2</version>
+  <version>1.2.3</version>
   <packaging>jar</packaging>
 
   <name>Kafka Redis Connector (Sink and Source)</name>

diff --git a/src/main/java/io/github/jaredpetersen/kafkaconnectredis/source/listener/RecordConverter.java b/src/main/java/io/github/jaredpetersen/kafkaconnectredis/source/listener/RecordConverter.java
@@ -55,7 +55,7 @@ public SourceRecord convert(RedisMessage redisMessage) {
       key,
       VALUE_SCHEMA,
       value,
-      Instant.now().getEpochSecond()
+      Instant.now().toEpochMilli()
     );
   }
 }
diff --git a/...t/java/io/github/jaredpetersen/kafkaconnectredis/source/listener/RecordConverterTest.java b/...t/java/io/github/jaredpetersen/kafkaconnectredis/source/listener/RecordConverterTest.java
@@ -32,6 +32,6 @@ void convertTransformsRedisMessageToSourceRecord() {
     assertEquals(redisMessage.getPattern(), ((Struct) sourceRecord.key()).getString("pattern"));
     assertEquals(Schema.Type.STRUCT, sourceRecord.valueSchema().type());
     assertEquals(redisMessage.getMessage(), ((Struct) sourceRecord.value()).getString("message"));
-    assertTrue(sourceRecord.timestamp() <= Instant.now().getEpochSecond());
+    assertTrue(sourceRecord.timestamp() <= Instant.now().toEpochMilli());
   }
 }