From b31c5a6a325728b5dc5faebd1a33386952d733d5 Mon Sep 17 00:00:00 2001
From: Cong Zhao <zhaocong@apache.org>
Date: Wed, 9 Nov 2022 17:02:07 +0800
Subject: [PATCH] [fix][client] Support LocalDateTime Conversion (#18334)

* Support LocalDateTime Conversion

* move `TimestampMicrosConversion` to correct line
---
 .../pulsar/client/impl/schema/AvroSchema.java |  4 +++-
 .../client/impl/schema/AvroSchemaTest.java    | 21 +++++++++++++++++++
 2 files changed, 24 insertions(+), 1 deletion(-)

diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/AvroSchema.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/AvroSchema.java
index d1e9acf7413a3..d2e994efc53f4 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/AvroSchema.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/AvroSchema.java
@@ -118,6 +118,8 @@ public static void addLogicalTypeConversions(ReflectData reflectData, boolean js
         reflectData.addLogicalTypeConversion(new TimeConversions.DateConversion());
         reflectData.addLogicalTypeConversion(new TimeConversions.TimeMillisConversion());
         reflectData.addLogicalTypeConversion(new TimeConversions.TimeMicrosConversion());
+        reflectData.addLogicalTypeConversion(new TimeConversions.LocalTimestampMillisConversion());
+        reflectData.addLogicalTypeConversion(new TimeConversions.LocalTimestampMicrosConversion());
         if (jsr310ConversionEnabled) {
             // The conversion that is registered first is higher priority than the registered later.
             reflectData.addLogicalTypeConversion(new TimeConversions.TimestampMillisConversion());
@@ -128,8 +130,8 @@ public static void addLogicalTypeConversions(ReflectData reflectData, boolean js
             } catch (ClassNotFoundException e) {
                 // Skip if have not provide joda-time dependency.
             }
-            reflectData.addLogicalTypeConversion(new TimeConversions.TimestampMicrosConversion());
         }
+        reflectData.addLogicalTypeConversion(new TimeConversions.TimestampMicrosConversion());
         reflectData.addLogicalTypeConversion(new Conversions.UUIDConversion());
     }
 
diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/AvroSchemaTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/AvroSchemaTest.java
index 9de37ea35b8aa..2b183edae8e25 100644
--- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/AvroSchemaTest.java
+++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/AvroSchemaTest.java
@@ -32,6 +32,7 @@
 import java.math.BigDecimal;
 import java.time.Instant;
 import java.time.LocalDate;
+import java.time.LocalDateTime;
 import java.time.LocalTime;
 import java.time.temporal.ChronoUnit;
 import java.util.Arrays;
@@ -543,4 +544,24 @@ public void testTimestampWithJsonDefAndJSR310ConversionEnabled(){
         Assert.assertNotEquals(Instant.class, decodeWithJsonNoClassLoader.getValue().getClass());
     }
 
+    @Data
+    @AllArgsConstructor
+    @NoArgsConstructor
+    private static class LocalDateTimePojo {
+        LocalDateTime value;
+    }
+
+    @Test
+    public void testLocalDateTime() {
+        SchemaDefinition<LocalDateTimePojo> schemaDefinition =
+                SchemaDefinition.<LocalDateTimePojo>builder().withPojo(LocalDateTimePojo.class)
+                        .withJSR310ConversionEnabled(true).build();
+
+        AvroSchema<LocalDateTimePojo> avroSchema = AvroSchema.of(schemaDefinition);
+        LocalDateTime now = LocalDateTime.now();
+        byte[] bytes = avroSchema.encode(new LocalDateTimePojo(now));
+
+        LocalDateTimePojo pojo = avroSchema.decode(bytes);
+        assertEquals(pojo.getValue().truncatedTo(ChronoUnit.MILLIS), now.truncatedTo(ChronoUnit.MILLIS));
+    }
 }