fix pom version

googleapis · Jul 16, 2021 · e13c4a3 · e13c4a3
1 parent a9eb732
commit e13c4a3
Show file tree

Hide file tree

Showing 3 changed files with 5 additions and 4 deletions.
diff --git a/pom.xml b/pom.xml
@@ -114,7 +114,7 @@
         <dependency>
             <groupId>org.apache.flink</groupId>
             <artifactId>flink-connector-test-utils</artifactId>
-            <version>1.13-SNAPSHOT</version>
+            <version>1.13.0</version>
             <scope>test</scope>
         </dependency>
     </dependencies>

diff --git a/src/main/java/com/google/cloud/pubsublite/flink/reader/PubsubLiteRecordEmitter.java b/src/main/java/com/google/cloud/pubsublite/flink/reader/PubsubLiteRecordEmitter.java
@@ -27,7 +27,7 @@ public void emitRecord(
       SourceOutput<T> sourceOutput,
       SubscriptionPartitionSplitState subscriptionPartitionSplitState) {
     if (record.value().isPresent()) {
-      sourceOutput.collect(record.value().get(), record.timestamp());
+      sourceOutput.collect(record.value().get(), record.timestamp().toEpochMilli());
     }
     subscriptionPartitionSplitState.setCurrent(record.offset());
   }

diff --git a/src/test/java/com/google/cloud/pubsublite/flink/reader/PubsubLiteRecordEmitterTest.java b/src/test/java/com/google/cloud/pubsublite/flink/reader/PubsubLiteRecordEmitterTest.java
@@ -22,6 +22,7 @@
 import com.google.cloud.pubsublite.Offset;
 import com.google.cloud.pubsublite.flink.split.SubscriptionPartitionSplit;
 import com.google.cloud.pubsublite.flink.split.SubscriptionPartitionSplitState;
+import java.time.Instant;
 import java.util.Optional;
 import org.apache.flink.api.connector.source.SourceOutput;
 import org.junit.Test;
@@ -34,7 +35,7 @@ public class PubsubLiteRecordEmitterTest {
   static final SubscriptionPartitionSplit split =
       SubscriptionPartitionSplit.create(
           exampleSubscriptionPath(), examplePartition(), Offset.of(0));
-  static final long exampleTime = 1000;
+  static final Instant exampleTime = Instant.ofEpochMilli(1000);
 
   @Mock SourceOutput<Long> mockOutput;
 
@@ -47,7 +48,7 @@ public void testRecordEmitted() {
         Record.create(Optional.of(100L), exampleOffset(), exampleTime), mockOutput, state);
 
     assertThat(state.toSplit().start()).isEqualTo(exampleOffset());
-    verify(mockOutput).collect(100L, exampleTime);
+    verify(mockOutput).collect(100L, exampleTime.toEpochMilli());
   }
 
   @Test