cover empty eventime case in record

Author: zzzming

README.md

@@ -12,7 +12,7 @@ GET `admin/v2/functions/connectors` displays the nar is loaded successfully as
 ```
 
 ```
-$ bin/pulsar-admin sinks create --archive ./connectors/pulsar-io-s3-1.0.nar --inputs aws-s3-input-topic --name aws-s3-test --sink-config-file ./connectors/pulsar-postgres-jdbc-sink.yaml
+$ bin/pulsar-admin sinks create --archive ./connectors/pulsar-io-s3-1.0.nar --inputs aws-s3-input-topic --name aws-s3-test --sink-config-file ./connectors/pulsar-s3-io.yaml --processing-guarantees EFFECTIVELY_ONCE --subs-position Earliest
 "Created successfully"
 
 $ bin/pulsar-admin sinks list

s3/src/main/java/com/kesque/pulsar/sink/s3/AWSS3Sink.java

@@ -84,9 +84,10 @@ public class AWSS3Sink implements Sink<byte[]> {
     @Override
     public void write(Record<byte[]> record) throws Exception {
         synchronized (this) {
-            //int len = record.getValue().length;
-
-            this.lastRecordEpoch = record.getEventTime().get();
+            Optional<Long> eventTimeOptional = record.getEventTime();
+            if (eventTimeOptional.isPresent()) {
+                this.lastRecordEpoch = eventTimeOptional.get();
+            }
             Long ledgerId = getLedgerId(record.getRecordSequence().get());
             LOG.info("ledgerID {} event time {}", ledgerId, this.lastRecordEpoch);
             // Optional<Message<byte[]>> msgOption = record.getMessage(); //.get();