[hotfix-#1216][kafka] Fix ObjectSizeCalculator only supported on HotS…

…pot VM (#1230) Co-authored-by: otxiyin <ptsxndg@gmail.com>
DTStack · Sep 7, 2022 · 5df8d09 · 5df8d09
1 parent 33efba2
commit 5df8d09
Showing 1 changed file with 11 additions and 2 deletions.
diff --git a/...c/main/java/com/dtstack/chunjun/connector/kafka/sink/DynamicKafkaSerializationSchema.java b/...c/main/java/com/dtstack/chunjun/connector/kafka/sink/DynamicKafkaSerializationSchema.java
@@ -24,6 +24,7 @@
 import com.dtstack.chunjun.dirty.utils.DirtyConfUtil;
 import com.dtstack.chunjun.metrics.AccumulatorCollector;
 import com.dtstack.chunjun.metrics.BaseMetric;
+import com.dtstack.chunjun.metrics.RowSizeCalculator;
 import com.dtstack.chunjun.restore.FormatState;
 import com.dtstack.chunjun.util.JsonUtil;
 
@@ -41,7 +42,6 @@
 import org.apache.flink.types.RowKind;
 import org.apache.flink.util.Preconditions;
 
-import jdk.nashorn.internal.ir.debug.ObjectSizeCalculator;
 import org.apache.kafka.clients.producer.ProducerRecord;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -98,6 +98,9 @@ public class DynamicKafkaSerializationSchema
  /** 累加器收集器 */
  protected AccumulatorCollector accumulatorCollector;
 
+ /** 对象大小计算器 */
+ protected RowSizeCalculator rowSizeCalculator;
+
  protected LongCounter bytesWriteCounter;
  protected LongCounter durationCounter;
  protected LongCounter numWriteCounter;
@@ -161,6 +164,7 @@ protected void beforeOpen() {
  initStatisticsAccumulator();
  initRestoreInfo();
  initAccumulatorCollector();
+ initRowSizeCalculator();
 
  checkpointMode =
  context.getCheckpointMode() == null
@@ -201,7 +205,7 @@ public void open(SerializationSchema.InitializationContext context) throws Excep
  protected void beforeSerialize(long size, RowData rowData) {
  updateDuration();
  numWriteCounter.add(size);
- bytesWriteCounter.add(ObjectSizeCalculator.getObjectSize(rowData));
+ bytesWriteCounter.add(rowSizeCalculator.getObjectSize(rowData));
  if (checkpointEnabled) {
  snapshotWriteCounter.add(size);
  }
@@ -380,6 +384,11 @@ private void initAccumulatorCollector() {
  accumulatorCollector.start();
  }
 
+ /** 初始化对象大小计算器 */
+ private void initRowSizeCalculator() {
+ rowSizeCalculator = RowSizeCalculator.getRowSizeCalculator();
+ }
+
  /** 从checkpoint状态缓存map中恢复上次任务的指标信息 */
  private void initRestoreInfo() {
  if (formatState == null) {