Flink是按值还是按引用处理状态？ValueState操作疑问

我见过的所有Flink处理ValueState的示例都类似下面这种写法：

private ValueState<Long> myState;

@Override
public void open(Configuration parameters) {
  myState = getRuntimeContext().getState(new ValueStateDescriptor<>(
      "myState", Types.LONG));
}

@Override
public void processElement(Integer value, Context ctx, Collector<Integer> out) {
  // 读取状态
  Long stateValue = myState.value();
  if (stateValue == null) {
    // 首次初始化
    stateValue = 0L;
  }

  // 业务逻辑处理
  
  // 将修改后的值写回状态
  myState.update(stateValue);
  out.collect(value);
}

我不清楚value()和update()的内部逻辑：这两个方法执行时是否会做序列化/反序列化？状态里存储的是对象的副本还是引用？

当使用POJO而非Long这类基本类型包装类时，这个差异就变得至关重要了：

private static class MyPojo {
  public String name;
  public long birthday;
}

private ValueState<MyPojo> myState;

@Override
public void open(Configuration parameters) {
  myState = getRuntimeContext().getState(new ValueStateDescriptor<>(
      "myState", Types.POJO(MyPojo.class)));
}

@Override
public void processElement(Integer value, Context ctx, Collector<Integer> out) {
  MyPojo stateValue = myState.value();
  if (stateValue == null) {
    stateValue = new MyPojo();
    myState.update(stateValue);
  }

  // 业务逻辑处理

  stateValue.name = String.valueOf(value);
  stateValue.birthday = System.currentTimeMillis();
  // 这里需要调用update()吗？还是修改会通过引用直接生效？

  out.collect(value);
}

核心问题：修改从状态中取出的POJO对象后，是否必须调用update()？还是Flink会自动持久化我对该对象的修改？

问题解答 #

1. 基本类型包装类（如Long）的情况

Long这类基本类型包装类是不可变对象，调用value()时，Flink会从状态后端读取并反序列化出一个新的对象副本返回给你。由于对象不可变，你修改时本质是创建了一个新的Long对象，因此必须调用update()将新对象写入状态，否则修改不会被持久化。

2. POJO等可变对象的情况

调用value()时，Flink返回的是对象的引用，而非副本，但这并不意味着你修改引用指向的对象后，Flink会自动持久化修改：

• Flink的状态持久化（如Checkpoint、快照）依赖于显式的更新通知——只有调用update()时，状态后端才会知道需要重新序列化这个对象并保存最新状态。
• 如果仅修改POJO的属性但不调用update()，当前处理流程中你能看到修改后的对象，但一旦触发Checkpoint或故障恢复，之前的修改会丢失：因为状态后端没有收到更新指令，仍然会使用旧的序列化版本来恢复状态。

因此，修改POJO对象的属性后，必须调用update()，明确告知Flink状态已更新，这样修改才会被持久化到状态后端。

内容的提问来源于stack exchange，提问作者Kim Gräsman