From 67fa127463e58c166d6d55317eb249ebbdb5fa19 Mon Sep 17 00:00:00 2001
From: "hj.lee" <hj.lee@fullcontact.com>
Date: Thu, 18 Nov 2021 14:04:13 -0700
Subject: [PATCH] added int64 customer formater

---
 .../storage/partitioner/FieldPartitioner.java   | 17 ++++++++++++++++-
 1 file changed, 16 insertions(+), 1 deletion(-)

diff --git a/partitioner/src/main/java/io/confluent/connect/storage/partitioner/FieldPartitioner.java b/partitioner/src/main/java/io/confluent/connect/storage/partitioner/FieldPartitioner.java
index d1c192a24..5833dac88 100644
--- a/partitioner/src/main/java/io/confluent/connect/storage/partitioner/FieldPartitioner.java
+++ b/partitioner/src/main/java/io/confluent/connect/storage/partitioner/FieldPartitioner.java
@@ -23,6 +23,8 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.text.SimpleDateFormat;
+import java.util.Date;
 import java.util.List;
 import java.util.Map;
 
@@ -31,6 +33,8 @@
 
 public class FieldPartitioner<T> extends DefaultPartitioner<T> {
   private static final Logger log = LoggerFactory.getLogger(FieldPartitioner.class);
+  private static final SimpleDateFormat SIMPLE_FORMAT = new SimpleDateFormat("yyyy-MM-dd");
+
   private List<String> fieldNames;
 
 
@@ -60,10 +64,12 @@ public String encodePartition(SinkRecord sinkRecord) {
           case INT8:
           case INT16:
           case INT32:
-          case INT64:
             Number record = (Number) partitionKey;
             builder.append(fieldName + "=" + record.toString());
             break;
+          case INT64:
+            builder.append(fieldName + "=" + formatInt64(partitionKey));
+            break;
           case STRING:
             builder.append(fieldName + "=" + (String) partitionKey);
             break;
@@ -92,4 +98,13 @@ public List<T> partitionFields() {
     }
     return partitionFields;
   }
+
+  private String formatInt64(Object partitionKey) {
+    if (partitionKey instanceof Date) {
+      Date record = (Date) partitionKey;
+      return SIMPLE_FORMAT.format(record);
+    }
+    Number record = (Number) partitionKey;
+    return record.toString();
+  }
 }