[fix][client] Support LocalDateTime Conversion (apache#18334)

* Support LocalDateTime Conversion * move `TimestampMicrosConversion` to correct line
streamnative · Nov 9, 2022 · b31c5a6 · b31c5a6
1 parent ef5346b
commit b31c5a6
Show file tree

Hide file tree

Showing 2 changed files with 24 additions and 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
@@ -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
@@ -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));
+    }
 }