From b58f30d71a1ce8eaa0a33002b94e019f0aa72fac Mon Sep 17 00:00:00 2001
From: Sagar Sumit <sagarsumit09@gmail.com>
Date: Mon, 12 Sep 2022 20:25:54 +0530
Subject: [PATCH 1/5] [HUDI-4687] Avoid setAccessible which breaks strong
 encapsulation

---
 hudi-common/pom.xml                           |  5 +
 .../common/util/ObjectSizeCalculator.java     | 47 ++++-----
 .../common/util/TestObjectSizeCalculator.java | 95 +++++++++++++++++++
 pom.xml                                       |  7 ++
 4 files changed, 132 insertions(+), 22 deletions(-)
 create mode 100644 hudi-common/src/test/java/org/apache/hudi/common/util/TestObjectSizeCalculator.java

diff --git a/hudi-common/pom.xml b/hudi-common/pom.xml
index d3f37864e3ee..ae5303cc7072 100644
--- a/hudi-common/pom.xml
+++ b/hudi-common/pom.xml
@@ -101,6 +101,11 @@
   </build>
 
   <dependencies>
+    <dependency>
+      <groupId>org.openjdk.jol</groupId>
+      <artifactId>jol-core</artifactId>
+    </dependency>
+
     <!-- Logging -->
     <dependency>
       <groupId>org.apache.logging.log4j</groupId>
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/util/ObjectSizeCalculator.java b/hudi-common/src/main/java/org/apache/hudi/common/util/ObjectSizeCalculator.java
index 7e625e8eb478..5d16950abfb8 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/util/ObjectSizeCalculator.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/util/ObjectSizeCalculator.java
@@ -18,14 +18,16 @@
 
 package org.apache.hudi.common.util;
 
-import org.apache.hudi.common.util.jvm.MemoryLayoutSpecification;
 import org.apache.hudi.common.util.jvm.HotSpotMemoryLayoutSpecification32bit;
 import org.apache.hudi.common.util.jvm.HotSpotMemoryLayoutSpecification64bit;
 import org.apache.hudi.common.util.jvm.HotSpotMemoryLayoutSpecification64bitCompressed;
+import org.apache.hudi.common.util.jvm.MemoryLayoutSpecification;
 import org.apache.hudi.common.util.jvm.OpenJ9MemoryLayoutSpecification32bit;
 import org.apache.hudi.common.util.jvm.OpenJ9MemoryLayoutSpecification64bit;
 import org.apache.hudi.common.util.jvm.OpenJ9MemoryLayoutSpecification64bitCompressed;
 
+import org.openjdk.jol.info.ClassLayout;
+
 import java.lang.management.ManagementFactory;
 import java.lang.management.MemoryPoolMXBean;
 import java.lang.reflect.Array;
@@ -87,7 +89,7 @@ public static long getObjectSize(Object obj) throws UnsupportedOperationExceptio
   // added.
   private final int superclassFieldPadding;
 
-  private final Map<Class<?>, ClassSizeInfo> classSizeInfos = new IdentityHashMap<>();
+  private static final Map<Class<?>, ClassSizeInfo> CLASS_SIZE_INFO_MAP = new IdentityHashMap<>();
 
   private final Set<Object> alreadyVisited = Collections.newSetFromMap(new IdentityHashMap<>());
   private final Deque<Object> pending = new ArrayDeque<>(64);
@@ -127,6 +129,8 @@ public synchronized long calculateObjectSize(Object obj) {
         }
         obj = pending.removeFirst();
       }
+    } catch (IllegalAccessException | SecurityException e) {
+      return ClassLayout.parseClass(obj.getClass()).instanceSize();
     } finally {
       alreadyVisited.clear();
       pending.clear();
@@ -134,16 +138,20 @@ public synchronized long calculateObjectSize(Object obj) {
     }
   }
 
-  private ClassSizeInfo getClassSizeInfo(final Class<?> clazz) {
-    ClassSizeInfo csi = classSizeInfos.get(clazz);
+  private static ClassSizeInfo getClassSizeInfo(final Class<?> clazz,
+                                                int referenceSize,
+                                                int superclassFieldPadding,
+                                                int objectHeaderSize,
+                                                int objectPadding) {
+    ClassSizeInfo csi = CLASS_SIZE_INFO_MAP.get(clazz);
     if (csi == null) {
-      csi = new ClassSizeInfo(clazz);
-      classSizeInfos.put(clazz, csi);
+      csi = new ClassSizeInfo(clazz, referenceSize, superclassFieldPadding, objectHeaderSize, objectPadding);
+      CLASS_SIZE_INFO_MAP.put(clazz, csi);
     }
     return csi;
   }
 
-  private void visit(Object obj) {
+  private void visit(Object obj) throws IllegalAccessException {
     if (alreadyVisited.contains(obj)) {
       return;
     }
@@ -155,7 +163,7 @@ private void visit(Object obj) {
       if (clazz.isArray()) {
         visitArray(obj);
       } else {
-        getClassSizeInfo(clazz).visit(obj, this);
+        getClassSizeInfo(clazz, referenceSize, superclassFieldPadding, objectHeaderSize, objectPadding).visit(obj, this);
       }
     }
   }
@@ -200,7 +208,7 @@ private static class ArrayElementsVisitor {
       this.array = array;
     }
 
-    public void visit(ObjectSizeCalculator calc) {
+    public void visit(ObjectSizeCalculator calc) throws IllegalAccessException {
       for (Object elem : array) {
         if (elem != null) {
           calc.visit(elem);
@@ -223,7 +231,7 @@ static long roundTo(long x, int multiple) {
     return ((x + multiple - 1) / multiple) * multiple;
   }
 
-  private class ClassSizeInfo {
+  private static class ClassSizeInfo {
 
     // Padded fields + header size
     private final long objectSize;
@@ -232,7 +240,7 @@ private class ClassSizeInfo {
     private final long fieldsSize;
     private final Field[] referenceFields;
 
-    public ClassSizeInfo(Class<?> clazz) {
+    public ClassSizeInfo(Class<?> clazz, int referenceSize, int superclassFieldPadding, int objectHeaderSize, int objectPadding) {
       long fieldsSize = 0;
       final List<Field> referenceFields = new LinkedList<>();
       for (Field f : clazz.getDeclaredFields()) {
@@ -243,14 +251,13 @@ public ClassSizeInfo(Class<?> clazz) {
         if (type.isPrimitive()) {
           fieldsSize += getPrimitiveFieldSize(type);
         } else {
-          f.setAccessible(true);
           referenceFields.add(f);
           fieldsSize += referenceSize;
         }
       }
       final Class<?> superClass = clazz.getSuperclass();
       if (superClass != null) {
-        final ClassSizeInfo superClassInfo = getClassSizeInfo(superClass);
+        final ClassSizeInfo superClassInfo = getClassSizeInfo(superClass, referenceSize, superclassFieldPadding, objectHeaderSize, objectPadding);
         fieldsSize += roundTo(superClassInfo.fieldsSize, superclassFieldPadding);
         referenceFields.addAll(Arrays.asList(superClassInfo.referenceFields));
       }
@@ -259,18 +266,14 @@ public ClassSizeInfo(Class<?> clazz) {
       this.referenceFields = referenceFields.toArray(new Field[referenceFields.size()]);
     }
 
-    void visit(Object obj, ObjectSizeCalculator calc) {
+    void visit(Object obj, ObjectSizeCalculator calc) throws IllegalAccessException {
       calc.increaseSize(objectSize);
       enqueueReferencedObjects(obj, calc);
     }
 
-    public void enqueueReferencedObjects(Object obj, ObjectSizeCalculator calc) {
+    public void enqueueReferencedObjects(Object obj, ObjectSizeCalculator calc) throws IllegalAccessException {
       for (Field f : referenceFields) {
-        try {
-          calc.enqueue(f.get(obj));
-        } catch (IllegalAccessException e) {
-          throw new AssertionError("Unexpected denial of access to " + f, e);
-        }
+        calc.enqueue(f.get(obj));
       }
     }
   }
@@ -307,7 +310,7 @@ static MemoryLayoutSpecification getEffectiveMemoryLayoutSpecification() {
         return new OpenJ9MemoryLayoutSpecification32bit();
       } else if (!"64".equals(dataModel)) {
         throw new UnsupportedOperationException(
-                "Unrecognized value '" + dataModel + "' of sun.arch.data.model system property");
+            "Unrecognized value '" + dataModel + "' of sun.arch.data.model system property");
       }
 
       long maxMemory = 0;
@@ -329,7 +332,7 @@ static MemoryLayoutSpecification getEffectiveMemoryLayoutSpecification() {
         return new HotSpotMemoryLayoutSpecification32bit();
       } else if (!"64".equals(dataModel)) {
         throw new UnsupportedOperationException(
-                "Unrecognized value '" + dataModel + "' of sun.arch.data.model system property");
+            "Unrecognized value '" + dataModel + "' of sun.arch.data.model system property");
       }
 
       final int vmVersion = Integer.parseInt(strVmVersion.substring(0, strVmVersion.indexOf('.')));
diff --git a/hudi-common/src/test/java/org/apache/hudi/common/util/TestObjectSizeCalculator.java b/hudi-common/src/test/java/org/apache/hudi/common/util/TestObjectSizeCalculator.java
new file mode 100644
index 000000000000..3d1ecfab9607
--- /dev/null
+++ b/hudi-common/src/test/java/org/apache/hudi/common/util/TestObjectSizeCalculator.java
@@ -0,0 +1,95 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.hudi.common.util;
+
+import org.apache.hudi.common.model.HoodieRecord;
+
+import org.apache.avro.Schema;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.List;
+
+public class TestObjectSizeCalculator {
+
+  @Test
+  public void testGetObjectSize() {
+    EmptyClass emptyClass = new EmptyClass();
+    StringClass stringClass = new StringClass();
+    PayloadClass payloadClass = new PayloadClass();
+    String emptyString = "";
+    String string = "hello";
+    String[] stringArray = {emptyString, string, " world"};
+    String[] anotherStringArray = new String[100];
+    List<String> stringList = new ArrayList<>();
+    StringBuilder stringBuilder = new StringBuilder(100);
+    int maxIntPrimitive = Integer.MAX_VALUE;
+    int minIntPrimitive = Integer.MIN_VALUE;
+    Integer maxInteger = Integer.MAX_VALUE;
+    Integer minInteger = Integer.MIN_VALUE;
+    long zeroLong = 0L;
+    double zeroDouble = 0.0;
+    boolean booleanField = true;
+    Object object = new Object();
+
+    Assertions.assertDoesNotThrow(() -> {
+      printObjectSize(emptyString);
+      printObjectSize(string);
+      printObjectSize(stringArray);
+      printObjectSize(anotherStringArray);
+      printObjectSize(stringList);
+      printObjectSize(stringBuilder);
+      printObjectSize(maxIntPrimitive);
+      printObjectSize(minIntPrimitive);
+      printObjectSize(maxInteger);
+      printObjectSize(minInteger);
+      printObjectSize(zeroLong);
+      printObjectSize(zeroDouble);
+      printObjectSize(booleanField);
+      printObjectSize(DayOfWeek.TUESDAY);
+      printObjectSize(object);
+      printObjectSize(emptyClass);
+      printObjectSize(stringClass);
+      printObjectSize(payloadClass);
+      printObjectSize(Schema.create(Schema.Type.STRING));
+    });
+  }
+
+  public static void printObjectSize(Object object) {
+    System.out.println("Object type: " + object.getClass() + ", size: " + ObjectSizeCalculator.getObjectSize(object) + " bytes");
+  }
+
+  class EmptyClass {
+  }
+
+  class StringClass {
+    private String s;
+  }
+
+  class PayloadClass implements Serializable {
+    private HoodieRecord record;
+  }
+
+  public enum DayOfWeek {
+    MONDAY, TUESDAY, WEDNESDAY, THURSDAY, FRIDAY, SATURDAY, SUNDAY
+  }
+}
diff --git a/pom.xml b/pom.xml
index 767d046ee5b4..a9cc0920f621 100644
--- a/pom.xml
+++ b/pom.xml
@@ -200,6 +200,7 @@
     <protoc.version>3.21.5</protoc.version>
     <dynamodb.lockclient.version>1.1.0</dynamodb.lockclient.version>
     <zookeeper.version>3.5.7</zookeeper.version>
+    <openjdk.jol.version>0.2</openjdk.jol.version>
     <dynamodb-local.port>8000</dynamodb-local.port>
     <dynamodb-local.endpoint>http://localhost:${dynamodb-local.port}</dynamodb-local.endpoint>
     <springboot.version>2.7.3</springboot.version>
@@ -635,6 +636,12 @@
         <version>${scala.collection-compat.version}</version>
       </dependency>
 
+      <dependency>
+        <groupId>org.openjdk.jol</groupId>
+        <artifactId>jol-core</artifactId>
+        <version>${openjdk.jol.version}</version>
+      </dependency>
+
       <!-- Logging -->
       <!-- NOTE: All the following deps have to have "provided" scope to make sure these are not conflicting
            w/ implementations that are using Hudi as a library. For ex, all Spark < 3.3 are still relying on Log4j1

From b3e80a32511aa7a35aa696f5d9eb30a63b349504 Mon Sep 17 00:00:00 2001
From: Sagar Sumit <sagarsumit09@gmail.com>
Date: Fri, 23 Sep 2022 19:08:26 +0530
Subject: [PATCH 2/5] Address feedback and fix tests

---
 .../common/util/ObjectSizeCalculator.java     | 30 +++++++-----
 .../common/util/TestObjectSizeCalculator.java | 48 +++++++++----------
 .../hudi/sink/TestWriteCopyOnWrite.java       | 10 ++--
 3 files changed, 44 insertions(+), 44 deletions(-)

diff --git a/hudi-common/src/main/java/org/apache/hudi/common/util/ObjectSizeCalculator.java b/hudi-common/src/main/java/org/apache/hudi/common/util/ObjectSizeCalculator.java
index 5d16950abfb8..a285999ad623 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/util/ObjectSizeCalculator.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/util/ObjectSizeCalculator.java
@@ -89,7 +89,7 @@ public static long getObjectSize(Object obj) throws UnsupportedOperationExceptio
   // added.
   private final int superclassFieldPadding;
 
-  private static final Map<Class<?>, ClassSizeInfo> CLASS_SIZE_INFO_MAP = new IdentityHashMap<>();
+  private final Map<Class<?>, ClassSizeInfo> classSizeInfos = new IdentityHashMap<>();
 
   private final Set<Object> alreadyVisited = Collections.newSetFromMap(new IdentityHashMap<>());
   private final Deque<Object> pending = new ArrayDeque<>(64);
@@ -130,6 +130,14 @@ public synchronized long calculateObjectSize(Object obj) {
         obj = pending.removeFirst();
       }
     } catch (IllegalAccessException | SecurityException e) {
+      // JDK versions 16 or later enforce strong encapsulation and do now allow to invoke `setAccessible` on a field,
+      // especially when the `isAccessible` is false. More details in JEP 403. While integrating Hudi with other
+      // software packages that compile against JDK 16 or later (e.g. Trino), the IllegalAccessException will be thrown.
+      // In that case, we use Java Object Layout (JOL) to estimate the object size.
+      //
+      // NOTE: We cannot get the object size base on the amount of byte serialized because there is no guarantee
+      //       that the incoming object is serializable. We could have used Java's Instrumentation API but it
+      //       needs an instrumentation agent that can be hooked to the JVM. In lieu of that, we are using JOL.
       return ClassLayout.parseClass(obj.getClass()).instanceSize();
     } finally {
       alreadyVisited.clear();
@@ -138,15 +146,11 @@ public synchronized long calculateObjectSize(Object obj) {
     }
   }
 
-  private static ClassSizeInfo getClassSizeInfo(final Class<?> clazz,
-                                                int referenceSize,
-                                                int superclassFieldPadding,
-                                                int objectHeaderSize,
-                                                int objectPadding) {
-    ClassSizeInfo csi = CLASS_SIZE_INFO_MAP.get(clazz);
+  private ClassSizeInfo getClassSizeInfo(final Class<?> clazz) {
+    ClassSizeInfo csi = classSizeInfos.get(clazz);
     if (csi == null) {
-      csi = new ClassSizeInfo(clazz, referenceSize, superclassFieldPadding, objectHeaderSize, objectPadding);
-      CLASS_SIZE_INFO_MAP.put(clazz, csi);
+      csi = new ClassSizeInfo(clazz);
+      classSizeInfos.put(clazz, csi);
     }
     return csi;
   }
@@ -163,7 +167,7 @@ private void visit(Object obj) throws IllegalAccessException {
       if (clazz.isArray()) {
         visitArray(obj);
       } else {
-        getClassSizeInfo(clazz, referenceSize, superclassFieldPadding, objectHeaderSize, objectPadding).visit(obj, this);
+        getClassSizeInfo(clazz).visit(obj, this);
       }
     }
   }
@@ -231,7 +235,7 @@ static long roundTo(long x, int multiple) {
     return ((x + multiple - 1) / multiple) * multiple;
   }
 
-  private static class ClassSizeInfo {
+  private class ClassSizeInfo {
 
     // Padded fields + header size
     private final long objectSize;
@@ -240,7 +244,7 @@ private static class ClassSizeInfo {
     private final long fieldsSize;
     private final Field[] referenceFields;
 
-    public ClassSizeInfo(Class<?> clazz, int referenceSize, int superclassFieldPadding, int objectHeaderSize, int objectPadding) {
+    public ClassSizeInfo(Class<?> clazz) {
       long fieldsSize = 0;
       final List<Field> referenceFields = new LinkedList<>();
       for (Field f : clazz.getDeclaredFields()) {
@@ -257,7 +261,7 @@ public ClassSizeInfo(Class<?> clazz, int referenceSize, int superclassFieldPaddi
       }
       final Class<?> superClass = clazz.getSuperclass();
       if (superClass != null) {
-        final ClassSizeInfo superClassInfo = getClassSizeInfo(superClass, referenceSize, superclassFieldPadding, objectHeaderSize, objectPadding);
+        final ClassSizeInfo superClassInfo = getClassSizeInfo(superClass);
         fieldsSize += roundTo(superClassInfo.fieldsSize, superclassFieldPadding);
         referenceFields.addAll(Arrays.asList(superClassInfo.referenceFields));
       }
diff --git a/hudi-common/src/test/java/org/apache/hudi/common/util/TestObjectSizeCalculator.java b/hudi-common/src/test/java/org/apache/hudi/common/util/TestObjectSizeCalculator.java
index 3d1ecfab9607..ca5de749a6ab 100644
--- a/hudi-common/src/test/java/org/apache/hudi/common/util/TestObjectSizeCalculator.java
+++ b/hudi-common/src/test/java/org/apache/hudi/common/util/TestObjectSizeCalculator.java
@@ -22,13 +22,15 @@
 import org.apache.hudi.common.model.HoodieRecord;
 
 import org.apache.avro.Schema;
-import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 
 import java.io.Serializable;
 import java.util.ArrayList;
 import java.util.List;
 
+import static org.apache.hudi.common.util.ObjectSizeCalculator.getObjectSize;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
 public class TestObjectSizeCalculator {
 
   @Test
@@ -51,31 +53,25 @@ public void testGetObjectSize() {
     boolean booleanField = true;
     Object object = new Object();
 
-    Assertions.assertDoesNotThrow(() -> {
-      printObjectSize(emptyString);
-      printObjectSize(string);
-      printObjectSize(stringArray);
-      printObjectSize(anotherStringArray);
-      printObjectSize(stringList);
-      printObjectSize(stringBuilder);
-      printObjectSize(maxIntPrimitive);
-      printObjectSize(minIntPrimitive);
-      printObjectSize(maxInteger);
-      printObjectSize(minInteger);
-      printObjectSize(zeroLong);
-      printObjectSize(zeroDouble);
-      printObjectSize(booleanField);
-      printObjectSize(DayOfWeek.TUESDAY);
-      printObjectSize(object);
-      printObjectSize(emptyClass);
-      printObjectSize(stringClass);
-      printObjectSize(payloadClass);
-      printObjectSize(Schema.create(Schema.Type.STRING));
-    });
-  }
-
-  public static void printObjectSize(Object object) {
-    System.out.println("Object type: " + object.getClass() + ", size: " + ObjectSizeCalculator.getObjectSize(object) + " bytes");
+    assertEquals(24, getObjectSize(emptyString));
+    assertEquals(24, getObjectSize(string));
+    assertEquals(16, getObjectSize(stringArray));
+    assertEquals(416, getObjectSize(anotherStringArray));
+    assertEquals(24, getObjectSize(stringList));
+    assertEquals(24, getObjectSize(stringBuilder));
+    assertEquals(16, getObjectSize(maxIntPrimitive));
+    assertEquals(16, getObjectSize(minIntPrimitive));
+    assertEquals(16, getObjectSize(maxInteger));
+    assertEquals(16, getObjectSize(minInteger));
+    assertEquals(24, getObjectSize(zeroLong));
+    assertEquals(24, getObjectSize(zeroDouble));
+    assertEquals(16, getObjectSize(booleanField));
+    assertEquals(24, getObjectSize(DayOfWeek.TUESDAY));
+    assertEquals(16, getObjectSize(object));
+    assertEquals(32, getObjectSize(emptyClass));
+    assertEquals(24, getObjectSize(stringClass));
+    assertEquals(24, getObjectSize(payloadClass));
+    assertEquals(32, getObjectSize(Schema.create(Schema.Type.STRING)));
   }
 
   class EmptyClass {
diff --git a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/TestWriteCopyOnWrite.java b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/TestWriteCopyOnWrite.java
index 9a6aaf01e669..9f53f874d708 100644
--- a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/TestWriteCopyOnWrite.java
+++ b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/TestWriteCopyOnWrite.java
@@ -188,7 +188,7 @@ public void testUpsertWithDelete() throws Exception {
   @Test
   public void testInsertWithMiniBatches() throws Exception {
     // reset the config option
-    conf.setDouble(FlinkOptions.WRITE_BATCH_SIZE, 0.0008); // 839 bytes batch size
+    conf.setDouble(FlinkOptions.WRITE_BATCH_SIZE, 0.00008); // 839 bytes batch size
 
     Map<String, String> expected = getMiniBatchExpected();
 
@@ -213,7 +213,7 @@ public void testInsertWithMiniBatches() throws Exception {
   @Test
   public void testInsertWithDeduplication() throws Exception {
     // reset the config option
-    conf.setDouble(FlinkOptions.WRITE_BATCH_SIZE, 0.0008); // 839 bytes batch size
+    conf.setDouble(FlinkOptions.WRITE_BATCH_SIZE, 0.00008); // 839 bytes batch size
     conf.setBoolean(FlinkOptions.PRE_COMBINE, true);
 
     Map<String, String> expected = new HashMap<>();
@@ -263,7 +263,7 @@ public void testInsertClustering() throws Exception {
     // reset the config option
     conf.setString(FlinkOptions.OPERATION, "insert");
     conf.setBoolean(FlinkOptions.INSERT_CLUSTER, true);
-    conf.setDouble(FlinkOptions.WRITE_TASK_MAX_SIZE, 200.0008); // 839 bytes buffer size
+    conf.setDouble(FlinkOptions.WRITE_TASK_MAX_SIZE, 200.00008); // 839 bytes buffer size
 
     TestWriteMergeOnRead.TestHarness.instance()
         // record (operation: 'I') is 304 bytes and record (operation: 'U') is 352 bytes.
@@ -312,7 +312,7 @@ public void testInsertAsyncClustering() throws Exception {
   @Test
   public void testInsertWithSmallBufferSize() throws Exception {
     // reset the config option
-    conf.setDouble(FlinkOptions.WRITE_TASK_MAX_SIZE, 200.0008); // 839 bytes buffer size
+    conf.setDouble(FlinkOptions.WRITE_TASK_MAX_SIZE, 200.00008); // 839 bytes buffer size
 
     Map<String, String> expected = getMiniBatchExpected();
 
@@ -406,7 +406,7 @@ protected void validateIndexLoaded() throws Exception {
   public void testWriteExactlyOnce() throws Exception {
     // reset the config option
     conf.setLong(FlinkOptions.WRITE_COMMIT_ACK_TIMEOUT, 1L);
-    conf.setDouble(FlinkOptions.WRITE_TASK_MAX_SIZE, 200.0006); // 630 bytes buffer size
+    conf.setDouble(FlinkOptions.WRITE_TASK_MAX_SIZE, 200.00006); // 630 bytes buffer size
     preparePipeline(conf)
         .consume(TestData.DATA_SET_INSERT)
         .emptyEventBuffer()

From 0f723714f4d30d0da445a5c03be8c0c8b16e25e1 Mon Sep 17 00:00:00 2001
From: Sagar Sumit <sagarsumit09@gmail.com>
Date: Tue, 27 Sep 2022 18:22:51 +0530
Subject: [PATCH 3/5] Include JOL library in bundles

---
 packaging/hudi-flink-bundle/pom.xml         | 5 +++++
 packaging/hudi-hadoop-mr-bundle/pom.xml     | 5 +++++
 packaging/hudi-hive-sync-bundle/pom.xml     | 5 +++++
 packaging/hudi-integ-test-bundle/pom.xml    | 5 +++++
 packaging/hudi-kafka-connect-bundle/pom.xml | 5 +++++
 packaging/hudi-presto-bundle/pom.xml        | 5 +++++
 packaging/hudi-spark-bundle/pom.xml         | 5 +++++
 packaging/hudi-trino-bundle/pom.xml         | 5 +++++
 packaging/hudi-utilities-bundle/pom.xml     | 5 +++++
 9 files changed, 45 insertions(+)

diff --git a/packaging/hudi-flink-bundle/pom.xml b/packaging/hudi-flink-bundle/pom.xml
index de0478ff0691..d51a9af45fd3 100644
--- a/packaging/hudi-flink-bundle/pom.xml
+++ b/packaging/hudi-flink-bundle/pom.xml
@@ -163,6 +163,7 @@
                   <include>org.apache.htrace:htrace-core4</include>
                   <include>commons-codec:commons-codec</include>
                   <include>commons-io:commons-io</include>
+                  <include>org.openjdk.jol:jol-core</include>
                 </includes>
               </artifactSet>
               <relocations>
@@ -222,6 +223,10 @@
                   <pattern>com.fasterxml.jackson.</pattern>
                   <shadedPattern>${flink.bundle.shade.prefix}com.fasterxml.jackson.</shadedPattern>
                 </relocation>
+                <relocation>
+                  <pattern>org.openjdk.jol.</pattern>
+                  <shadedPattern>org.apache.hudi.org.openjdk.jol.</shadedPattern>
+                </relocation>
                 <!-- The classes below in org.apache.hadoop.metrics2 package come from
                 hbase-hadoop-compat and hbase-hadoop2-compat, which have to be shaded one by one,
                 instead of shading all classes under org.apache.hadoop.metrics2 including ones
diff --git a/packaging/hudi-hadoop-mr-bundle/pom.xml b/packaging/hudi-hadoop-mr-bundle/pom.xml
index 1788ffb2edf4..e87a86e5a018 100644
--- a/packaging/hudi-hadoop-mr-bundle/pom.xml
+++ b/packaging/hudi-hadoop-mr-bundle/pom.xml
@@ -90,6 +90,7 @@
                   <include>org.apache.htrace:htrace-core4</include>
                   <include>com.yammer.metrics:metrics-core</include>
                   <include>commons-io:commons-io</include>
+                  <include>org.openjdk.jol:jol-core</include>
                 </includes>
               </artifactSet>
               <relocations>
@@ -144,6 +145,10 @@
                   <pattern>com.google.common.</pattern>
                   <shadedPattern>org.apache.hudi.com.google.common.</shadedPattern>
                 </relocation>
+                <relocation>
+                  <pattern>org.openjdk.jol.</pattern>
+                  <shadedPattern>org.apache.hudi.org.openjdk.jol.</shadedPattern>
+                </relocation>
                 <!-- The classes below in org.apache.hadoop.metrics2 package come from
                 hbase-hadoop-compat and hbase-hadoop2-compat, which have to be shaded one by one,
                 instead of shading all classes under org.apache.hadoop.metrics2 including ones
diff --git a/packaging/hudi-hive-sync-bundle/pom.xml b/packaging/hudi-hive-sync-bundle/pom.xml
index 7164af659b71..71334018a1c4 100644
--- a/packaging/hudi-hive-sync-bundle/pom.xml
+++ b/packaging/hudi-hive-sync-bundle/pom.xml
@@ -90,6 +90,7 @@
                   <include>org.objenesis:objenesis</include>
                   <include>com.esotericsoftware:minlog</include>
                   <include>commons-io:commons-io</include>
+                  <include>org.openjdk.jol:jol-core</include>
                 </includes>
               </artifactSet>
               <relocations>
@@ -124,6 +125,10 @@
                   <pattern>org.apache.htrace.</pattern>
                   <shadedPattern>org.apache.hudi.org.apache.htrace.</shadedPattern>
                 </relocation>
+                <relocation>
+                  <pattern>org.openjdk.jol.</pattern>
+                  <shadedPattern>org.apache.hudi.org.openjdk.jol.</shadedPattern>
+                </relocation>
                 <!-- The classes below in org.apache.hadoop.metrics2 package come from
                 hbase-hadoop-compat and hbase-hadoop2-compat, which have to be shaded one by one,
                 instead of shading all classes under org.apache.hadoop.metrics2 including ones
diff --git a/packaging/hudi-integ-test-bundle/pom.xml b/packaging/hudi-integ-test-bundle/pom.xml
index 776f9f73d1e4..b48e42ac0a06 100644
--- a/packaging/hudi-integ-test-bundle/pom.xml
+++ b/packaging/hudi-integ-test-bundle/pom.xml
@@ -184,6 +184,7 @@
                   <include>io.prometheus:simpleclient_dropwizard</include>
                   <include>io.prometheus:simpleclient_pushgateway</include>
                   <include>io.prometheus:simpleclient_common</include>
+                  <include>org.openjdk.jol:jol-core</include>
                 </includes>
               </artifactSet>
               <relocations>
@@ -309,6 +310,10 @@
                   <pattern>org.apache.parquet.avro.</pattern>
                   <shadedPattern>org.apache.hudi.org.apache.parquet.avro.</shadedPattern>
                 </relocation>
+                <relocation>
+                  <pattern>org.openjdk.jol.</pattern>
+                  <shadedPattern>org.apache.hudi.org.openjdk.jol.</shadedPattern>
+                </relocation>
                 <!-- The classes below in org.apache.hadoop.metrics2 package come from
                 hbase-hadoop-compat and hbase-hadoop2-compat, which have to be shaded one by one,
                 instead of shading all classes under org.apache.hadoop.metrics2 including ones
diff --git a/packaging/hudi-kafka-connect-bundle/pom.xml b/packaging/hudi-kafka-connect-bundle/pom.xml
index cb63e7a5f2e2..057752e14b81 100644
--- a/packaging/hudi-kafka-connect-bundle/pom.xml
+++ b/packaging/hudi-kafka-connect-bundle/pom.xml
@@ -133,6 +133,7 @@
                                     <include>org.apache.htrace:htrace-core4</include>
                                     <include>org.scala-lang:*</include>
                                     <include>commons-io:commons-io</include>
+                                    <include>org.openjdk.jol:jol-core</include>
                                 </includes>
                             </artifactSet>
                             <relocations>
@@ -174,6 +175,10 @@
                                     <pattern>org.apache.htrace.</pattern>
                                     <shadedPattern>org.apache.hudi.org.apache.htrace.</shadedPattern>
                                 </relocation>
+                                <relocation>
+                                    <pattern>org.openjdk.jol.</pattern>
+                                    <shadedPattern>org.apache.hudi.org.openjdk.jol.</shadedPattern>
+                                </relocation>
                                 <!-- The classes below in org.apache.hadoop.metrics2 package come from
                 hbase-hadoop-compat and hbase-hadoop2-compat, which have to be shaded one by one,
                 instead of shading all classes under org.apache.hadoop.metrics2 including ones
diff --git a/packaging/hudi-presto-bundle/pom.xml b/packaging/hudi-presto-bundle/pom.xml
index 9045671659a0..c9b18b8694df 100644
--- a/packaging/hudi-presto-bundle/pom.xml
+++ b/packaging/hudi-presto-bundle/pom.xml
@@ -94,6 +94,7 @@
                   <include>commons-io:commons-io</include>
                   <include>commons-lang:commons-lang</include>
                   <include>com.google.protobuf:protobuf-java</include>
+                  <include>org.openjdk.jol:jol-core</include>
                 </includes>
               </artifactSet>
               <relocations>
@@ -164,6 +165,10 @@
                   <pattern>org.apache.parquet.avro.</pattern>
                   <shadedPattern>${presto.bundle.bootstrap.shade.prefix}org.apache.parquet.avro.</shadedPattern>
                 </relocation>
+                <relocation>
+                  <pattern>org.openjdk.jol.</pattern>
+                  <shadedPattern>org.apache.hudi.org.openjdk.jol.</shadedPattern>
+                </relocation>
                 <!-- The classes below in org.apache.hadoop.metrics2 package come from
                 hbase-hadoop-compat and hbase-hadoop2-compat, which have to be shaded one by one,
                 instead of shading all classes under org.apache.hadoop.metrics2 including ones
diff --git a/packaging/hudi-spark-bundle/pom.xml b/packaging/hudi-spark-bundle/pom.xml
index 13cbb813d471..511f9d449354 100644
--- a/packaging/hudi-spark-bundle/pom.xml
+++ b/packaging/hudi-spark-bundle/pom.xml
@@ -135,6 +135,7 @@
                   <include>org.apache.curator:curator-recipes</include>
                   <include>commons-codec:commons-codec</include>
                   <include>commons-io:commons-io</include>
+                  <include>org.openjdk.jol:jol-core</include>
                 </includes>
               </artifactSet>
               <relocations>
@@ -220,6 +221,10 @@
                   <pattern>com.google.common.</pattern>
                   <shadedPattern>org.apache.hudi.com.google.common.</shadedPattern>
                 </relocation>
+                <relocation>
+                  <pattern>org.openjdk.jol.</pattern>
+                  <shadedPattern>org.apache.hudi.org.openjdk.jol.</shadedPattern>
+                </relocation>
                 <!-- TODO: Revisit GH ISSUE #533 & PR#633-->
                 <!-- The classes below in org.apache.hadoop.metrics2 package come from
                 hbase-hadoop-compat and hbase-hadoop2-compat, which have to be shaded one by one,
diff --git a/packaging/hudi-trino-bundle/pom.xml b/packaging/hudi-trino-bundle/pom.xml
index f6f487c01e1f..39c5ced1a910 100644
--- a/packaging/hudi-trino-bundle/pom.xml
+++ b/packaging/hudi-trino-bundle/pom.xml
@@ -94,6 +94,7 @@
                   <include>commons-lang:commons-lang</include>
                   <include>commons-io:commons-io</include>
                   <include>com.google.protobuf:protobuf-java</include>
+                  <include>org.openjdk.jol:jol-core</include>
                 </includes>
               </artifactSet>
               <relocations>
@@ -156,6 +157,10 @@
                   <pattern>com.google.protobuf.</pattern>
                   <shadedPattern>${trino.bundle.bootstrap.shade.prefix}com.google.protobuf.</shadedPattern>
                 </relocation>
+                <relocation>
+                  <pattern>org.openjdk.jol.</pattern>
+                  <shadedPattern>org.apache.hudi.org.openjdk.jol.</shadedPattern>
+                </relocation>
                 <!-- The classes below in org.apache.hadoop.metrics2 package come from
                 hbase-hadoop-compat and hbase-hadoop2-compat, which have to be shaded one by one,
                 instead of shading all classes under org.apache.hadoop.metrics2 including ones
diff --git a/packaging/hudi-utilities-bundle/pom.xml b/packaging/hudi-utilities-bundle/pom.xml
index 86b1eea2a132..1f740b276f44 100644
--- a/packaging/hudi-utilities-bundle/pom.xml
+++ b/packaging/hudi-utilities-bundle/pom.xml
@@ -167,6 +167,7 @@
                   <include>org.apache.curator:curator-recipes</include>
                   <include>commons-codec:commons-codec</include>
                   <include>commons-io:commons-io</include>
+                  <include>org.openjdk.jol:jol-core</include>
                 </includes>
               </artifactSet>
               <relocations>
@@ -244,6 +245,10 @@
                   <pattern>org.eclipse.jetty.</pattern>
                   <shadedPattern>org.apache.hudi.org.eclipse.jetty.</shadedPattern>
                 </relocation>
+                <relocation>
+                  <pattern>org.openjdk.jol.</pattern>
+                  <shadedPattern>org.apache.hudi.org.openjdk.jol.</shadedPattern>
+                </relocation>
                 <!-- The classes below in org.apache.hadoop.metrics2 package come from
                 hbase-hadoop-compat and hbase-hadoop2-compat, which have to be shaded one by one,
                 instead of shading all classes under org.apache.hadoop.metrics2 including ones

From 23928b7db6d79929b01df68d2b6e1d892263223b Mon Sep 17 00:00:00 2001
From: Sagar Sumit <sagarsumit09@gmail.com>
Date: Tue, 27 Sep 2022 20:02:25 +0530
Subject: [PATCH 4/5] Use GraphLayout for deep size

---
 .../common/util/ObjectSizeCalculator.java     | 328 +-----------------
 .../common/util/TestObjectSizeCalculator.java |  29 +-
 .../hudi/sink/TestWriteCopyOnWrite.java       |  10 +-
 pom.xml                                       |   2 +-
 4 files changed, 40 insertions(+), 329 deletions(-)

diff --git a/hudi-common/src/main/java/org/apache/hudi/common/util/ObjectSizeCalculator.java b/hudi-common/src/main/java/org/apache/hudi/common/util/ObjectSizeCalculator.java
index a285999ad623..86f1d9215e8c 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/util/ObjectSizeCalculator.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/util/ObjectSizeCalculator.java
@@ -18,35 +18,11 @@
 
 package org.apache.hudi.common.util;
 
-import org.apache.hudi.common.util.jvm.HotSpotMemoryLayoutSpecification32bit;
-import org.apache.hudi.common.util.jvm.HotSpotMemoryLayoutSpecification64bit;
-import org.apache.hudi.common.util.jvm.HotSpotMemoryLayoutSpecification64bitCompressed;
-import org.apache.hudi.common.util.jvm.MemoryLayoutSpecification;
-import org.apache.hudi.common.util.jvm.OpenJ9MemoryLayoutSpecification32bit;
-import org.apache.hudi.common.util.jvm.OpenJ9MemoryLayoutSpecification64bit;
-import org.apache.hudi.common.util.jvm.OpenJ9MemoryLayoutSpecification64bitCompressed;
-
-import org.openjdk.jol.info.ClassLayout;
-
-import java.lang.management.ManagementFactory;
-import java.lang.management.MemoryPoolMXBean;
-import java.lang.reflect.Array;
-import java.lang.reflect.Field;
-import java.lang.reflect.Modifier;
-import java.util.ArrayDeque;
-import java.util.Arrays;
-import java.util.Collections;
-import java.util.Deque;
-import java.util.IdentityHashMap;
-import java.util.LinkedList;
-import java.util.List;
-import java.util.Map;
-import java.util.Objects;
-import java.util.Set;
+import org.openjdk.jol.info.GraphLayout;
 
 /**
  * Contains utility methods for calculating the memory usage of objects. It only works on the HotSpot and OpenJ9 JVMs, and infers
- * the actual memory layout (32 bit vs. 64 bit word size, compressed object pointers vs. uncompressed) from best
+ * the actual memory layout (32 bit vs. 64 bit word size, compressed object pointers vs. uncompressed) from the best
  * available indicators. It can reliably detect a 32 bit vs. 64 bit JVM. It can only make an educated guess at whether
  * compressed OOPs are used, though; specifically, it knows what the JVM's default choice of OOP compression would be
  * based on HotSpot version and maximum heap sizes, but if the choice is explicitly overridden with the
@@ -56,14 +32,9 @@
  * @author Attila Szegedi
  */
 public class ObjectSizeCalculator {
-  private static class CurrentLayout {
-
-    private static final MemoryLayoutSpecification SPEC = getEffectiveMemoryLayoutSpecification();
-  }
-
   /**
    * Given an object, returns the total allocated size, in bytes, of the object and all other objects reachable from it.
-   * Attempts to to detect the current JVM memory layout, but may fail with {@link UnsupportedOperationException};
+   * Attempts to detect the current JVM memory layout, but may fail with {@link UnsupportedOperationException};
    *
    * @param obj the object; can be null. Passing in a {@link java.lang.Class} object doesn't do anything special, it
    *        measures the size of all objects reachable through it (which will include its class loader, and by
@@ -73,287 +44,16 @@ private static class CurrentLayout {
    * @throws UnsupportedOperationException if the current vm memory layout cannot be detected.
    */
   public static long getObjectSize(Object obj) throws UnsupportedOperationException {
-    return obj == null ? 0 : new ObjectSizeCalculator(CurrentLayout.SPEC).calculateObjectSize(obj);
-  }
-
-  // Fixed object header size for arrays.
-  private final int arrayHeaderSize;
-  // Fixed object header size for non-array objects.
-  private final int objectHeaderSize;
-  // Padding for the object size - if the object size is not an exact multiple
-  // of this, it is padded to the next multiple.
-  private final int objectPadding;
-  // Size of reference (pointer) fields.
-  private final int referenceSize;
-  // Padding for the fields of superclass before fields of subclasses are
-  // added.
-  private final int superclassFieldPadding;
-
-  private final Map<Class<?>, ClassSizeInfo> classSizeInfos = new IdentityHashMap<>();
-
-  private final Set<Object> alreadyVisited = Collections.newSetFromMap(new IdentityHashMap<>());
-  private final Deque<Object> pending = new ArrayDeque<>(64);
-  private long size;
-
-  /**
-   * Creates an object size calculator that can calculate object sizes for a given {@code memoryLayoutSpecification}.
-   *
-   * @param memoryLayoutSpecification a description of the JVM memory layout.
-   */
-  public ObjectSizeCalculator(MemoryLayoutSpecification memoryLayoutSpecification) {
-    Objects.requireNonNull(memoryLayoutSpecification);
-    arrayHeaderSize = memoryLayoutSpecification.getArrayHeaderSize();
-    objectHeaderSize = memoryLayoutSpecification.getObjectHeaderSize();
-    objectPadding = memoryLayoutSpecification.getObjectPadding();
-    referenceSize = memoryLayoutSpecification.getReferenceSize();
-    superclassFieldPadding = memoryLayoutSpecification.getSuperclassFieldPadding();
-  }
-
-  /**
-   * Given an object, returns the total allocated size, in bytes, of the object and all other objects reachable from it.
-   *
-   * @param obj the object; can be null. Passing in a {@link java.lang.Class} object doesn't do anything special, it
-   *        measures the size of all objects reachable through it (which will include its class loader, and by
-   *        extension, all other Class objects loaded by the same loader, and all the parent class loaders). It doesn't
-   *        provide the size of the static fields in the JVM class that the Class object represents.
-   * @return the total allocated size of the object and all other objects it retains.
-   */
-  public synchronized long calculateObjectSize(Object obj) {
-    // Breadth-first traversal instead of naive depth-first with recursive
-    // implementation, so we don't blow the stack traversing long linked lists.
-    try {
-      for (;;) {
-        visit(obj);
-        if (pending.isEmpty()) {
-          return size;
-        }
-        obj = pending.removeFirst();
-      }
-    } catch (IllegalAccessException | SecurityException e) {
-      // JDK versions 16 or later enforce strong encapsulation and do now allow to invoke `setAccessible` on a field,
-      // especially when the `isAccessible` is false. More details in JEP 403. While integrating Hudi with other
-      // software packages that compile against JDK 16 or later (e.g. Trino), the IllegalAccessException will be thrown.
-      // In that case, we use Java Object Layout (JOL) to estimate the object size.
-      //
-      // NOTE: We cannot get the object size base on the amount of byte serialized because there is no guarantee
-      //       that the incoming object is serializable. We could have used Java's Instrumentation API but it
-      //       needs an instrumentation agent that can be hooked to the JVM. In lieu of that, we are using JOL.
-      return ClassLayout.parseClass(obj.getClass()).instanceSize();
-    } finally {
-      alreadyVisited.clear();
-      pending.clear();
-      size = 0;
-    }
-  }
-
-  private ClassSizeInfo getClassSizeInfo(final Class<?> clazz) {
-    ClassSizeInfo csi = classSizeInfos.get(clazz);
-    if (csi == null) {
-      csi = new ClassSizeInfo(clazz);
-      classSizeInfos.put(clazz, csi);
-    }
-    return csi;
-  }
-
-  private void visit(Object obj) throws IllegalAccessException {
-    if (alreadyVisited.contains(obj)) {
-      return;
-    }
-    final Class<?> clazz = obj.getClass();
-    if (clazz == ArrayElementsVisitor.class) {
-      ((ArrayElementsVisitor) obj).visit(this);
-    } else {
-      alreadyVisited.add(obj);
-      if (clazz.isArray()) {
-        visitArray(obj);
-      } else {
-        getClassSizeInfo(clazz).visit(obj, this);
-      }
-    }
-  }
-
-  private void visitArray(Object array) {
-    final Class<?> componentType = array.getClass().getComponentType();
-    final int length = Array.getLength(array);
-    if (componentType.isPrimitive()) {
-      increaseByArraySize(length, getPrimitiveFieldSize(componentType));
-    } else {
-      increaseByArraySize(length, referenceSize);
-      // If we didn't use an ArrayElementsVisitor, we would be enqueueing every
-      // element of the array here instead. For large arrays, it would
-      // tremendously enlarge the queue. In essence, we're compressing it into
-      // a small command object instead. This is different than immediately
-      // visiting the elements, as their visiting is scheduled for the end of
-      // the current queue.
-      switch (length) {
-        case 0: {
-          break;
-        }
-        case 1: {
-          enqueue(Array.get(array, 0));
-          break;
-        }
-        default: {
-          enqueue(new ArrayElementsVisitor((Object[]) array));
-        }
-      }
-    }
-  }
-
-  private void increaseByArraySize(int length, long elementSize) {
-    increaseSize(roundTo(arrayHeaderSize + length * elementSize, objectPadding));
-  }
-
-  private static class ArrayElementsVisitor {
-
-    private final Object[] array;
-
-    ArrayElementsVisitor(Object[] array) {
-      this.array = array;
-    }
-
-    public void visit(ObjectSizeCalculator calc) throws IllegalAccessException {
-      for (Object elem : array) {
-        if (elem != null) {
-          calc.visit(elem);
-        }
-      }
-    }
-  }
-
-  void enqueue(Object obj) {
-    if (obj != null) {
-      pending.addLast(obj);
-    }
-  }
-
-  void increaseSize(long objectSize) {
-    size += objectSize;
-  }
-
-  static long roundTo(long x, int multiple) {
-    return ((x + multiple - 1) / multiple) * multiple;
-  }
-
-  private class ClassSizeInfo {
-
-    // Padded fields + header size
-    private final long objectSize;
-    // Only the fields size - used to calculate the subclasses' memory
-    // footprint.
-    private final long fieldsSize;
-    private final Field[] referenceFields;
-
-    public ClassSizeInfo(Class<?> clazz) {
-      long fieldsSize = 0;
-      final List<Field> referenceFields = new LinkedList<>();
-      for (Field f : clazz.getDeclaredFields()) {
-        if (Modifier.isStatic(f.getModifiers())) {
-          continue;
-        }
-        final Class<?> type = f.getType();
-        if (type.isPrimitive()) {
-          fieldsSize += getPrimitiveFieldSize(type);
-        } else {
-          referenceFields.add(f);
-          fieldsSize += referenceSize;
-        }
-      }
-      final Class<?> superClass = clazz.getSuperclass();
-      if (superClass != null) {
-        final ClassSizeInfo superClassInfo = getClassSizeInfo(superClass);
-        fieldsSize += roundTo(superClassInfo.fieldsSize, superclassFieldPadding);
-        referenceFields.addAll(Arrays.asList(superClassInfo.referenceFields));
-      }
-      this.fieldsSize = fieldsSize;
-      this.objectSize = roundTo(objectHeaderSize + fieldsSize, objectPadding);
-      this.referenceFields = referenceFields.toArray(new Field[referenceFields.size()]);
-    }
-
-    void visit(Object obj, ObjectSizeCalculator calc) throws IllegalAccessException {
-      calc.increaseSize(objectSize);
-      enqueueReferencedObjects(obj, calc);
-    }
-
-    public void enqueueReferencedObjects(Object obj, ObjectSizeCalculator calc) throws IllegalAccessException {
-      for (Field f : referenceFields) {
-        calc.enqueue(f.get(obj));
-      }
-    }
-  }
-
-  private static long getPrimitiveFieldSize(Class<?> type) {
-    if (type == boolean.class || type == byte.class) {
-      return 1;
-    }
-    if (type == char.class || type == short.class) {
-      return 2;
-    }
-    if (type == int.class || type == float.class) {
-      return 4;
-    }
-    if (type == long.class || type == double.class) {
-      return 8;
-    }
-    throw new AssertionError("Encountered unexpected primitive type " + type.getName());
-  }
-
-  static MemoryLayoutSpecification getEffectiveMemoryLayoutSpecification() {
-    final String vmName = System.getProperty("java.vm.name");
-    if (vmName == null || !(vmName.startsWith("Java HotSpot(TM) ") || vmName.startsWith("OpenJDK")
-        || vmName.startsWith("TwitterJDK") || vmName.startsWith("Eclipse OpenJ9"))) {
-      throw new UnsupportedOperationException("ObjectSizeCalculator only supported on HotSpot or Eclipse OpenJ9 VMs");
-    }
-
-    final String strVmVersion = System.getProperty("java.vm.version");
-    // Support for OpenJ9 JVM
-    if (strVmVersion.startsWith("openj9")) {
-      final String dataModel = System.getProperty("sun.arch.data.model");
-      if ("32".equals(dataModel)) {
-        // Running with 32-bit data model
-        return new OpenJ9MemoryLayoutSpecification32bit();
-      } else if (!"64".equals(dataModel)) {
-        throw new UnsupportedOperationException(
-            "Unrecognized value '" + dataModel + "' of sun.arch.data.model system property");
-      }
-
-      long maxMemory = 0;
-      for (MemoryPoolMXBean mp : ManagementFactory.getMemoryPoolMXBeans()) {
-        maxMemory += mp.getUsage().getMax();
-      }
-      if (maxMemory < 57L * 1024 * 1024 * 1024) {
-        // OpenJ9 use compressed references below 57GB of RAM total
-        return new OpenJ9MemoryLayoutSpecification64bitCompressed();
-      } else {
-        // it's a 64-bit uncompressed references object model
-        return new OpenJ9MemoryLayoutSpecification64bit();
-      }
-    } else {
-      // Support for HotSpot JVM
-      final String dataModel = System.getProperty("sun.arch.data.model");
-      if ("32".equals(dataModel)) {
-        // Running with 32-bit data model
-        return new HotSpotMemoryLayoutSpecification32bit();
-      } else if (!"64".equals(dataModel)) {
-        throw new UnsupportedOperationException(
-            "Unrecognized value '" + dataModel + "' of sun.arch.data.model system property");
-      }
-
-      final int vmVersion = Integer.parseInt(strVmVersion.substring(0, strVmVersion.indexOf('.')));
-      if (vmVersion >= 17) {
-        long maxMemory = 0;
-        for (MemoryPoolMXBean mp : ManagementFactory.getMemoryPoolMXBeans()) {
-          maxMemory += mp.getUsage().getMax();
-        }
-        if (maxMemory < 30L * 1024 * 1024 * 1024) {
-          // HotSpot 17.0 and above use compressed OOPs below 30GB of RAM total
-          // for all memory pools (yes, including code cache).
-          return new HotSpotMemoryLayoutSpecification64bitCompressed();
-        }
-      }
-
-      // In other cases, it's a 64-bit uncompressed OOPs object model
-      return new HotSpotMemoryLayoutSpecification64bit();
-    }
+    // JDK versions 16 or later enforce strong encapsulation and block illegal reflective access.
+    // In effect, we cannot calculate object size by deep reflection and invoking `setAccessible` on a field,
+    // especially when the `isAccessible` is false. More details in JEP 403. While integrating Hudi with other
+    // software packages that compile against JDK 16 or later (e.g. Trino), the IllegalAccessException will be thrown.
+    // In that case, we use Java Object Layout (JOL) to estimate the object size.
+    //
+    // NOTE: We cannot get the object size base on the amount of byte serialized because there is no guarantee
+    //       that the incoming object is serializable. We could have used Java's Instrumentation API, but it
+    //       needs an instrumentation agent that can be hooked to the JVM. In lieu of that, we are using JOL.
+    //       GraphLayout gives the deep size of an object, including the size of objects that are referenced from the given object.
+    return obj == null ? 0 : GraphLayout.parseInstance(obj).totalSize();
   }
 }
diff --git a/hudi-common/src/test/java/org/apache/hudi/common/util/TestObjectSizeCalculator.java b/hudi-common/src/test/java/org/apache/hudi/common/util/TestObjectSizeCalculator.java
index ca5de749a6ab..712f4b85f84a 100644
--- a/hudi-common/src/test/java/org/apache/hudi/common/util/TestObjectSizeCalculator.java
+++ b/hudi-common/src/test/java/org/apache/hudi/common/util/TestObjectSizeCalculator.java
@@ -52,13 +52,15 @@ public void testGetObjectSize() {
     double zeroDouble = 0.0;
     boolean booleanField = true;
     Object object = new Object();
+    String name = "Alice Bob";
+    Person person = new Person(name);
 
-    assertEquals(24, getObjectSize(emptyString));
-    assertEquals(24, getObjectSize(string));
-    assertEquals(16, getObjectSize(stringArray));
+    assertEquals(40, getObjectSize(emptyString));
+    assertEquals(56, getObjectSize(string));
+    assertEquals(184, getObjectSize(stringArray));
     assertEquals(416, getObjectSize(anotherStringArray));
-    assertEquals(24, getObjectSize(stringList));
-    assertEquals(24, getObjectSize(stringBuilder));
+    assertEquals(40, getObjectSize(stringList));
+    assertEquals(240, getObjectSize(stringBuilder));
     assertEquals(16, getObjectSize(maxIntPrimitive));
     assertEquals(16, getObjectSize(minIntPrimitive));
     assertEquals(16, getObjectSize(maxInteger));
@@ -66,12 +68,13 @@ public void testGetObjectSize() {
     assertEquals(24, getObjectSize(zeroLong));
     assertEquals(24, getObjectSize(zeroDouble));
     assertEquals(16, getObjectSize(booleanField));
-    assertEquals(24, getObjectSize(DayOfWeek.TUESDAY));
+    assertEquals(80, getObjectSize(DayOfWeek.TUESDAY));
     assertEquals(16, getObjectSize(object));
     assertEquals(32, getObjectSize(emptyClass));
-    assertEquals(24, getObjectSize(stringClass));
-    assertEquals(24, getObjectSize(payloadClass));
-    assertEquals(32, getObjectSize(Schema.create(Schema.Type.STRING)));
+    assertEquals(40, getObjectSize(stringClass));
+    assertEquals(40, getObjectSize(payloadClass));
+    assertEquals(1240, getObjectSize(Schema.create(Schema.Type.STRING)));
+    assertEquals(104, getObjectSize(person));
   }
 
   class EmptyClass {
@@ -85,6 +88,14 @@ class PayloadClass implements Serializable {
     private HoodieRecord record;
   }
 
+  class Person {
+    private String name;
+
+    public Person(String name) {
+      this.name = name;
+    }
+  }
+
   public enum DayOfWeek {
     MONDAY, TUESDAY, WEDNESDAY, THURSDAY, FRIDAY, SATURDAY, SUNDAY
   }
diff --git a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/TestWriteCopyOnWrite.java b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/TestWriteCopyOnWrite.java
index 9f53f874d708..9a6aaf01e669 100644
--- a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/TestWriteCopyOnWrite.java
+++ b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/TestWriteCopyOnWrite.java
@@ -188,7 +188,7 @@ public void testUpsertWithDelete() throws Exception {
   @Test
   public void testInsertWithMiniBatches() throws Exception {
     // reset the config option
-    conf.setDouble(FlinkOptions.WRITE_BATCH_SIZE, 0.00008); // 839 bytes batch size
+    conf.setDouble(FlinkOptions.WRITE_BATCH_SIZE, 0.0008); // 839 bytes batch size
 
     Map<String, String> expected = getMiniBatchExpected();
 
@@ -213,7 +213,7 @@ public void testInsertWithMiniBatches() throws Exception {
   @Test
   public void testInsertWithDeduplication() throws Exception {
     // reset the config option
-    conf.setDouble(FlinkOptions.WRITE_BATCH_SIZE, 0.00008); // 839 bytes batch size
+    conf.setDouble(FlinkOptions.WRITE_BATCH_SIZE, 0.0008); // 839 bytes batch size
     conf.setBoolean(FlinkOptions.PRE_COMBINE, true);
 
     Map<String, String> expected = new HashMap<>();
@@ -263,7 +263,7 @@ public void testInsertClustering() throws Exception {
     // reset the config option
     conf.setString(FlinkOptions.OPERATION, "insert");
     conf.setBoolean(FlinkOptions.INSERT_CLUSTER, true);
-    conf.setDouble(FlinkOptions.WRITE_TASK_MAX_SIZE, 200.00008); // 839 bytes buffer size
+    conf.setDouble(FlinkOptions.WRITE_TASK_MAX_SIZE, 200.0008); // 839 bytes buffer size
 
     TestWriteMergeOnRead.TestHarness.instance()
         // record (operation: 'I') is 304 bytes and record (operation: 'U') is 352 bytes.
@@ -312,7 +312,7 @@ public void testInsertAsyncClustering() throws Exception {
   @Test
   public void testInsertWithSmallBufferSize() throws Exception {
     // reset the config option
-    conf.setDouble(FlinkOptions.WRITE_TASK_MAX_SIZE, 200.00008); // 839 bytes buffer size
+    conf.setDouble(FlinkOptions.WRITE_TASK_MAX_SIZE, 200.0008); // 839 bytes buffer size
 
     Map<String, String> expected = getMiniBatchExpected();
 
@@ -406,7 +406,7 @@ protected void validateIndexLoaded() throws Exception {
   public void testWriteExactlyOnce() throws Exception {
     // reset the config option
     conf.setLong(FlinkOptions.WRITE_COMMIT_ACK_TIMEOUT, 1L);
-    conf.setDouble(FlinkOptions.WRITE_TASK_MAX_SIZE, 200.00006); // 630 bytes buffer size
+    conf.setDouble(FlinkOptions.WRITE_TASK_MAX_SIZE, 200.0006); // 630 bytes buffer size
     preparePipeline(conf)
         .consume(TestData.DATA_SET_INSERT)
         .emptyEventBuffer()
diff --git a/pom.xml b/pom.xml
index a9cc0920f621..9357efff67b2 100644
--- a/pom.xml
+++ b/pom.xml
@@ -200,7 +200,7 @@
     <protoc.version>3.21.5</protoc.version>
     <dynamodb.lockclient.version>1.1.0</dynamodb.lockclient.version>
     <zookeeper.version>3.5.7</zookeeper.version>
-    <openjdk.jol.version>0.2</openjdk.jol.version>
+    <openjdk.jol.version>0.16</openjdk.jol.version>
     <dynamodb-local.port>8000</dynamodb-local.port>
     <dynamodb-local.endpoint>http://localhost:${dynamodb-local.port}</dynamodb-local.endpoint>
     <springboot.version>2.7.3</springboot.version>

From 1751ff401d075667a7acedde954a9b9c459ce7ef Mon Sep 17 00:00:00 2001
From: Sagar Sumit <sagarsumit09@gmail.com>
Date: Wed, 28 Sep 2022 09:30:44 +0530
Subject: [PATCH 5/5] Ignore warning in stdout

---
 .../test/java/org/apache/hudi/integ/ITTestHoodieSanity.java   | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/hudi-integ-test/src/test/java/org/apache/hudi/integ/ITTestHoodieSanity.java b/hudi-integ-test/src/test/java/org/apache/hudi/integ/ITTestHoodieSanity.java
index e432f9dc423f..40827c650a93 100644
--- a/hudi-integ-test/src/test/java/org/apache/hudi/integ/ITTestHoodieSanity.java
+++ b/hudi-integ-test/src/test/java/org/apache/hudi/integ/ITTestHoodieSanity.java
@@ -185,12 +185,12 @@ public void testRunHoodieJavaApp(String command, String hiveTableName, String ta
 
     // Ensure row count is 80 (without duplicates) (100 - 20 deleted)
     stdOutErr = executeHiveCommand("select count(1) from " + snapshotTableName);
-    assertEquals(80, Integer.parseInt(stdOutErr.getLeft().trim()),
+    assertEquals(80, Integer.parseInt(stdOutErr.getLeft().substring(stdOutErr.getLeft().lastIndexOf("\n")).trim()),
         "Expecting 80 rows to be present in the snapshot table");
 
     if (roTableName.isPresent()) {
       stdOutErr = executeHiveCommand("select count(1) from " + roTableName.get());
-      assertEquals(80, Integer.parseInt(stdOutErr.getLeft().trim()),
+      assertEquals(80, Integer.parseInt(stdOutErr.getLeft().substring(stdOutErr.getLeft().lastIndexOf("\n")).trim()),
           "Expecting 80 rows to be present in the snapshot table");
     }