keep the last record for cumulative ack

Author: zzzming

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

@@ -65,6 +65,15 @@ public void setSecretAccessKey(String secretKey) {
         this.secretAccessKey = secretKey;
     }
 
+    // support trigger types are ledger, time based, size based
+    private String triggerType = "ledger";
+    public String getTriggerType() {
+        return this.triggerType;
+    }
+    public void setTriggerType(String triggerType) {
+        this.triggerType = triggerType;
+    }
+
     @FieldDoc(
         required = false,
         defaultValue = "",

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

@@ -2,6 +2,9 @@
 
 import java.io.IOException;
 import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import java.time.Instant;
+
 import static java.nio.charset.StandardCharsets.UTF_8;
 
 import java.util.HashMap;
@@ -66,12 +69,11 @@ public class AWSS3Sink implements Sink<byte[]> {
 
     private ParquetRecordWriter recordWriter;
 
-    private String filename;
+    private volatile String filename;
+
+    private long lastRecordEpoch = 0;
 
-    // debug
-    private int groupCounter = 0;
-    private int fileSuffix = 0;
-    // end of debug
+    private Duration timeTriggerDuration = Duration.ofHours(1);
 
     /**
     * Write a message to Sink
@@ -82,18 +84,18 @@ public class AWSS3Sink implements Sink<byte[]> {
     @Override
     public void write(Record<byte[]> record) throws Exception {
         synchronized (this) {
-            int len = record.getValue().length;
+            //int len = record.getValue().length;
 
+            this.lastRecordEpoch = record.getEventTime().get();
             Long ledgerId = getLedgerId(record.getRecordSequence().get());
-            LOG.info("ledgerID {} and value's length {}", ledgerId, len);
+            LOG.info("ledgerID {} event time {}", ledgerId, this.lastRecordEpoch);
             // Optional<Message<byte[]>> msgOption = record.getMessage(); //.get();
             // LOG.error("message option isPresent {}", msgOption.isPresent());
 
             this.filename = getFilename(this.filePrefix, ledgerId);
             this.recordWriter.write(record, this.filename);
         }
 
-        //bytes to generic data ??
     }
 
     @Override
@@ -117,7 +119,6 @@ public void open(Map<String, Object> config, SinkContext sinkContext) throws Exc
         for (String topicName : sinkContext.getInputTopics()){
             filePrefix = topicName + "-" + filePrefix;
         }
-        System.out.println("filePrefix " + this.filePrefix);
         LOG.info("filePrefix is " + this.filePrefix);
 
         incomingList = Lists.newArrayList();
@@ -146,4 +147,8 @@ private boolean isAvroSchema(SchemaData schemaData) {
     private static String getFilename(String prefix, Long ledger) {
         return prefix + Long.toString(ledger);
     }
+
+    private boolean isTimeTriggered() {
+        return Util.isOver(Instant.ofEpochMilli(lastRecordEpoch), timeTriggerDuration);
+    }
 }

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

@@ -1,5 +1,9 @@
 package com.kesque.pulsar.sink.s3;
 
+import java.text.SimpleDateFormat;
+import java.time.Duration;
+import java.time.Instant;
+
 public class Util {
 
     public static String ensureValidBucketName(String bucketName) {
@@ -15,4 +19,24 @@ public static String ensureValidBucketName(String bucketName) {
 
 		return formatted;
 	}
+
+	public static String getHourlyTimestamp(long epoch) {
+		SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd-HH");
+		return format.format(epoch);
+	}
+
+	public static String getMinuteTimestamp(long epoch) {
+		SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd-HH-mm");
+		return format.format(epoch);
+	}
+
+	/**
+	 * Check if the current time is over the duration limite since the start.
+	 * @param start
+	 * @param limit
+	 * @return
+	 */
+	public static boolean isOver(Instant start, Duration limit) {
+		return Duration.between(start, Instant.now()).compareTo(limit) > 0;
+	}
 }

s3/src/main/java/com/kesque/pulsar/sink/s3/format/parquet/ParquetRecordWriter.java

@@ -43,7 +43,7 @@ public class ParquetRecordWriter implements RecordWriter {
     private Configuration parquetWriterConfig;
     private Schema avroSchema;
     private volatile String currentFile = "";
-    private volatile Record<byte[]> lastRecord; // kept for batch ack
+    private volatile Record<byte[]> currentRecord; // kept for batch ack
 
     // parallel writer size
     int WRITER_LIMIT = 4;
@@ -77,12 +77,13 @@ public void write(Record<byte[]> record, String file) {
 
         GenericData.Record convertedRecord = (org.apache.avro.generic.GenericData.Record) JsonUtil.convertToAvro(GenericData.get(), datum, avroSchema);
         writeParquet(convertedRecord, file);
-        this.lastRecord = record;
+        this.currentRecord = record;
     }
 
     private synchronized void writeParquet(GenericData.Record record, String file) {
         log.info("currentFile is {} file name is {}", this.currentFile, file);
         String lastFile = this.currentFile; // save a copy because currentFile can be replace in the main thread
+        Record<byte[]> lastRecord = this.currentRecord; // ditto save a copy
         if (Strings.isNotBlank(lastFile) && !file.equals(lastFile)) {
             uploaderExecutor.execute(() -> {
                 ParquetWriter<GenericData.Record> writer = writerMap.get(lastFile);
@@ -108,9 +109,7 @@ private synchronized void writeParquet(GenericData.Record record, String file) {
                 s3ParquetOutputFileMap.remove(lastFile);
                 log.info("cumulative ack all pulsar messages and write to existing parquet writer, map size {}", writerMap.size());
                 lastRecord.ack(); // depends on cumulative ack
-
             });
-            
         }
         this.currentFile = file; // for the next write
 

s3/src/main/java/com/kesque/pulsar/sink/s3/storage/S3ParquetOutputFile.java

@@ -18,7 +18,6 @@ public S3ParquetOutputFile(S3Storage storage, String filename) {
 
     @Override
     public PositionOutputStream create(long blockSizeHint) throws IOException {
-        System.out.println("S3ParquetOutputStream create PositionOutputStream");
         s3out = (S3ParquetOutputStream) storage.create(filename, true);
         return s3out;
     }

s3/src/main/java/com/kesque/pulsar/sink/s3/storage/S3Storage.java

@@ -185,7 +185,6 @@ public S3OutputStream create(String path, boolean overwrite) {
           throw new IllegalArgumentException("Path can not be empty!");
         }
 
-        System.out.println("S3OutputStream . create() ... ");
         return new S3ParquetOutputStream(path, this.conf, s3);
     }
 }