update doc

reswqa · reswqa · commit e590b3baefb4 · 2024-09-26T22:33:36.000+08:00
diff --git a/docs/content.zh/docs/dev/datastream/fault-tolerance/state.md b/docs/content.zh/docs/dev/datastream/fault-tolerance/state.md
@@ -160,7 +160,7 @@ public class CountWindowAverage extends RichFlatMapFunction<Tuple2<Long, Long>,
     }
 
     @Override
-    public void open(Configuration config) {
+    public void open(OpenContext ctx) {
         ValueStateDescriptor<Tuple2<Long, Long>> descriptor =
                 new ValueStateDescriptor<>(
                         "average", // the state name
diff --git a/docs/content.zh/docs/learn-flink/etl.md b/docs/content.zh/docs/learn-flink/etl.md
@@ -236,7 +236,7 @@ minutesByStartCell
 
 对其中的每一个接口，Flink 同样提供了一个所谓 "rich" 的变体，如 `RichFlatMapFunction`，其中增加了以下方法，包括：
 
-- `open(Configuration c)`
+- `open(OpenContext context)`
 - `close()`
 - `getRuntimeContext()`
 
@@ -280,7 +280,7 @@ public static class Deduplicator extends RichFlatMapFunction<Event, Event> {
     ValueState<Boolean> keyHasBeenSeen;
 
     @Override
-    public void open(Configuration conf) {
+    public void open(OpenContext ctx) {
         ValueStateDescriptor<Boolean> desc = new ValueStateDescriptor<>("keyHasBeenSeen", Types.BOOLEAN);
         keyHasBeenSeen = getRuntimeContext().getState(desc);
     }
@@ -373,7 +373,7 @@ public static class ControlFunction extends RichCoFlatMapFunction<String, String
     private ValueState<Boolean> blocked;
       
     @Override
-    public void open(Configuration config) {
+    public void open(OpenContext ctx) {
         blocked = getRuntimeContext()
             .getState(new ValueStateDescriptor<>("blocked", Boolean.class));
     }
diff --git a/docs/content.zh/docs/learn-flink/event_driven.md b/docs/content.zh/docs/learn-flink/event_driven.md
@@ -79,7 +79,7 @@ public static class PseudoWindow extends
 
     @Override
     // 在初始化期间调用一次。
-    public void open(Configuration conf) {
+    public void open(OpenContext ctx) {
         . . .
     }
 
@@ -126,7 +126,7 @@ public static class PseudoWindow extends
 private transient MapState<Long, Float> sumOfTips;
 
 @Override
-public void open(Configuration conf) {
+public void open(OpenContext ctx) {
 
     MapStateDescriptor<Long, Float> sumDesc =
             new MapStateDescriptor<>("sumOfTips", Long.class, Float.class);
diff --git a/docs/content.zh/docs/ops/metrics.md b/docs/content.zh/docs/ops/metrics.md
@@ -52,7 +52,7 @@ public class MyMapper extends RichMapFunction<String, String> {
   private transient Counter counter;
 
   @Override
-  public void open(Configuration config) {
+  public void open(OpenContext ctx) {
     this.counter = getRuntimeContext()
       .getMetricGroup()
       .counter("myCounter");
@@ -116,7 +116,7 @@ public class MyMapper extends RichMapFunction<String, String> {
   private transient Counter counter;
 
   @Override
-  public void open(Configuration config) {
+  public void open(OpenContext ctx) {
     this.counter = getRuntimeContext()
       .getMetricGroup()
       .counter("myCustomCounter", new CustomCounter());
@@ -173,7 +173,7 @@ public class MyMapper extends RichMapFunction<String, String> {
   private transient int valueToExpose = 0;
 
   @Override
-  public void open(Configuration config) {
+  public void open(OpenContext ctx) {
     getRuntimeContext()
       .getMetricGroup()
       .gauge("MyGauge", new Gauge<Integer>() {
@@ -247,7 +247,7 @@ public class MyMapper extends RichMapFunction<Long, Long> {
   private transient Histogram histogram;
 
   @Override
-  public void open(Configuration config) {
+  public void open(OpenContext ctx) {
     this.histogram = getRuntimeContext()
       .getMetricGroup()
       .histogram("myHistogram", new MyHistogram());
@@ -307,7 +307,7 @@ public class MyMapper extends RichMapFunction<Long, Long> {
   private transient Histogram histogram;
 
   @Override
-  public void open(Configuration config) {
+  public void open(OpenContext ctx) {
     com.codahale.metrics.Histogram dropwizardHistogram =
       new com.codahale.metrics.Histogram(new SlidingWindowReservoir(500));
 
@@ -366,7 +366,7 @@ public class MyMapper extends RichMapFunction<Long, Long> {
   private transient Meter meter;
 
   @Override
-  public void open(Configuration config) {
+  public void open(OpenContext ctx) {
     this.meter = getRuntimeContext()
       .getMetricGroup()
       .meter("myMeter", new MyMeter());
@@ -440,7 +440,7 @@ public class MyMapper extends RichMapFunction<Long, Long> {
   private transient Meter meter;
 
   @Override
-  public void open(Configuration config) {
+  public void open(OpenContext ctx) {
     com.codahale.metrics.Meter dropwizardMeter = new com.codahale.metrics.Meter();
 
     this.meter = getRuntimeContext()
diff --git a/docs/content/docs/dev/datastream/fault-tolerance/state.md b/docs/content/docs/dev/datastream/fault-tolerance/state.md
@@ -192,7 +192,7 @@ public class CountWindowAverage extends RichFlatMapFunction<Tuple2<Long, Long>,
     }
 
     @Override
-    public void open(Configuration config) {
+    public void open(OpenContext ctx) {
         ValueStateDescriptor<Tuple2<Long, Long>> descriptor =
                 new ValueStateDescriptor<>(
                         "average", // the state name
diff --git a/docs/content/docs/learn-flink/etl.md b/docs/content/docs/learn-flink/etl.md
@@ -274,7 +274,7 @@ Abstract Method pattern.
 For each of these interfaces, Flink also provides a so-called "rich" variant, e.g.,
 `RichFlatMapFunction`, which has some additional methods, including:
 
-- `open(Configuration c)`
+- `open(OpenContext context)`
 - `close()`
 - `getRuntimeContext()`
 
@@ -329,7 +329,7 @@ public static class Deduplicator extends RichFlatMapFunction<Event, Event> {
     ValueState<Boolean> keyHasBeenSeen;
 
     @Override
-    public void open(Configuration conf) {
+    public void open(OpenContext ctx) {
         ValueStateDescriptor<Boolean> desc = new ValueStateDescriptor<>("keyHasBeenSeen", Types.BOOLEAN);
         keyHasBeenSeen = getRuntimeContext().getState(desc);
     }
@@ -447,7 +447,7 @@ public static class ControlFunction extends RichCoFlatMapFunction<String, String
     private ValueState<Boolean> blocked;
       
     @Override
-    public void open(Configuration config) {
+    public void open(OpenContext ctx) {
         blocked = getRuntimeContext()
             .getState(new ValueStateDescriptor<>("blocked", Boolean.class));
     }
diff --git a/docs/content/docs/learn-flink/event_driven.md b/docs/content/docs/learn-flink/event_driven.md
@@ -75,7 +75,7 @@ public static class PseudoWindow extends
 
     @Override
     // Called once during initialization.
-    public void open(Configuration conf) {
+    public void open(OpenContext ctx) {
         . . .
     }
 
@@ -116,7 +116,7 @@ Things to be aware of:
 private transient MapState<Long, Float> sumOfTips;
 
 @Override
-public void open(Configuration conf) {
+public void open(OpenContext ctx) {
 
     MapStateDescriptor<Long, Float> sumDesc =
             new MapStateDescriptor<>("sumOfTips", Long.class, Float.class);
diff --git a/docs/content/docs/ops/metrics.md b/docs/content/docs/ops/metrics.md
@@ -52,7 +52,7 @@ public class MyMapper extends RichMapFunction<String, String> {
   private transient Counter counter;
 
   @Override
-  public void open(Configuration config) {
+  public void open(OpenContext ctx) {
     this.counter = getRuntimeContext()
       .getMetricGroup()
       .counter("myCounter");
@@ -116,7 +116,7 @@ public class MyMapper extends RichMapFunction<String, String> {
   private transient Counter counter;
 
   @Override
-  public void open(Configuration config) {
+  public void open(OpenContext ctx) {
     this.counter = getRuntimeContext()
       .getMetricGroup()
       .counter("myCustomCounter", new CustomCounter());
@@ -173,7 +173,7 @@ public class MyMapper extends RichMapFunction<String, String> {
   private transient int valueToExpose = 0;
 
   @Override
-  public void open(Configuration config) {
+  public void open(OpenContext ctx) {
     getRuntimeContext()
       .getMetricGroup()
       .gauge("MyGauge", new Gauge<Integer>() {
@@ -247,7 +247,7 @@ public class MyMapper extends RichMapFunction<Long, Long> {
   private transient Histogram histogram;
 
   @Override
-  public void open(Configuration config) {
+  public void open(OpenContext ctx) {
     this.histogram = getRuntimeContext()
       .getMetricGroup()
       .histogram("myHistogram", new MyHistogram());
@@ -307,7 +307,7 @@ public class MyMapper extends RichMapFunction<Long, Long> {
   private transient Histogram histogram;
 
   @Override
-  public void open(Configuration config) {
+  public void open(OpenContext ctx) {
     com.codahale.metrics.Histogram dropwizardHistogram =
       new com.codahale.metrics.Histogram(new SlidingWindowReservoir(500));
 
@@ -366,7 +366,7 @@ public class MyMapper extends RichMapFunction<Long, Long> {
   private transient Meter meter;
 
   @Override
-  public void open(Configuration config) {
+  public void open(OpenContext ctx) {
     this.meter = getRuntimeContext()
       .getMetricGroup()
       .meter("myMeter", new MyMeter());
@@ -440,7 +440,7 @@ public class MyMapper extends RichMapFunction<Long, Long> {
   private transient Meter meter;
 
   @Override
-  public void open(Configuration config) {
+  public void open(OpenContext ctx) {
     com.codahale.metrics.Meter dropwizardMeter = new com.codahale.metrics.Meter();
 
     this.meter = getRuntimeContext()
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/functions/BroadcastVariableInitializer.java b/flink-core/src/main/java/org/apache/flink/api/common/functions/BroadcastVariableInitializer.java
@@ -41,7 +41,7 @@
  *
  *     private Map<Long, String> map;
  *
- *     public void open(Configuration cfg) throws Exception {
+ *     public void open(OpenContext ctx) throws Exception {
  *         getRuntimeContext().getBroadcastVariableWithInitializer("mapvar",
  *             new BroadcastVariableInitializer<Tuple2<Long, String>, Map<Long, String>>() {
  *
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/functions/RichFunction.java b/flink-core/src/main/java/org/apache/flink/api/common/functions/RichFunction.java
@@ -54,14 +54,6 @@ public interface RichFunction extends Function {
      * }
      * }</pre>
      *
-     * <p>By default, this method does nothing.
-     *
-     * <p>1. If you implement {@code open(OpenContext openContext)}, the {@code open(OpenContext
-     * openContext)} will be invoked and the {@code open(Configuration parameters)} won't be
-     * invoked. 2. If you don't implement {@code open(OpenContext openContext)}, the {@code
-     * open(Configuration parameters)} will be invoked in the default implementation of the {@code
-     * open(OpenContext openContext)}.
-     *
      * @param openContext The context containing information about the context in which the function
      *     is opened.
      * @throws Exception Implementations may forward exceptions, which are caught by the runtime.
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/functions/RuntimeContext.java b/flink-core/src/main/java/org/apache/flink/api/common/functions/RuntimeContext.java
@@ -232,7 +232,7 @@ <T, C> C getBroadcastVariableWithInitializer(
      *
      *     private ValueState<Long> state;
      *
-     *     public void open(Configuration cfg) {
+     *     public void open(OpenContext ctx) {
      *         state = getRuntimeContext().getState(
      *                 new ValueStateDescriptor<Long>("count", LongSerializer.INSTANCE, 0L));
      *     }
@@ -269,7 +269,7 @@ <T, C> C getBroadcastVariableWithInitializer(
      *
      *     private ListState<MyType> state;
      *
-     *     public void open(Configuration cfg) {
+     *     public void open(OpenContext ctx) {
      *         state = getRuntimeContext().getListState(
      *                 new ListStateDescriptor<>("myState", MyType.class));
      *     }
@@ -310,7 +310,7 @@ <T, C> C getBroadcastVariableWithInitializer(
      *
      *     private ReducingState<Long> state;
      *
-     *     public void open(Configuration cfg) {
+     *     public void open(OpenContext ctx) {
      *         state = getRuntimeContext().getReducingState(
      *                 new ReducingStateDescriptor<>("sum", (a, b) -> a + b, Long.class));
      *     }
@@ -348,7 +348,7 @@ <T, C> C getBroadcastVariableWithInitializer(
      *
      *     private AggregatingState<MyType, Long> state;
      *
-     *     public void open(Configuration cfg) {
+     *     public void open(OpenContext ctx) {
      *         state = getRuntimeContext().getAggregatingState(
      *                 new AggregatingStateDescriptor<>("sum", aggregateFunction, Long.class));
      *     }
@@ -388,7 +388,7 @@ <IN, ACC, OUT> AggregatingState<IN, OUT> getAggregatingState(
      *
      *     private MapState<MyType, Long> state;
      *
-     *     public void open(Configuration cfg) {
+     *     public void open(OpenContext ctx) {
      *         state = getRuntimeContext().getMapState(
      *                 new MapStateDescriptor<>("sum", MyType.class, Long.class));
      *     }
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/state/KeyedStateStore.java b/flink-core/src/main/java/org/apache/flink/api/common/state/KeyedStateStore.java
@@ -46,7 +46,7 @@ public interface KeyedStateStore {
      *
      *     private ValueState<Long> state;
      *
-     *     public void open(Configuration cfg) {
+     *     public void open(OpenContext ctx) {
      *         state = getRuntimeContext().getState(
      *                 new ValueStateDescriptor<Long>("count", LongSerializer.INSTANCE, 0L));
      *     }
@@ -83,7 +83,7 @@ public interface KeyedStateStore {
      *
      *     private ListState<MyType> state;
      *
-     *     public void open(Configuration cfg) {
+     *     public void open(OpenContext ctx) {
      *         state = getRuntimeContext().getListState(
      *                 new ListStateDescriptor<>("myState", MyType.class));
      *     }
@@ -124,7 +124,7 @@ public interface KeyedStateStore {
      *
      *     private ReducingState<Long> state;
      *
-     *     public void open(Configuration cfg) {
+     *     public void open(OpenContext ctx) {
      *         state = getRuntimeContext().getReducingState(
      *                 new ReducingStateDescriptor<>("sum", (a, b) -> a + b, Long.class));
      *     }
@@ -162,7 +162,7 @@ public interface KeyedStateStore {
      *
      *     private AggregatingState<MyType, Long> state;
      *
-     *     public void open(Configuration cfg) {
+     *     public void open(OpenContext ctx) {
      *         state = getRuntimeContext().getAggregatingState(
      *                 new AggregatingStateDescriptor<>("sum", aggregateFunction, Long.class));
      *     }
@@ -202,7 +202,7 @@ <IN, ACC, OUT> AggregatingState<IN, OUT> getAggregatingState(
      *
      *     private MapState<MyType, Long> state;
      *
-     *     public void open(Configuration cfg) {
+     *     public void open(OpenContext ctx) {
      *         state = getRuntimeContext().getMapState(
      *                 new MapStateDescriptor<>("sum", MyType.class, Long.class));
      *     }
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/BatchTask.java b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/BatchTask.java
@@ -1496,7 +1496,7 @@ public static void openUserCode(Function stub, Configuration parameters) throws
             FunctionUtils.openFunction(stub, DefaultOpenContext.INSTANCE);
         } catch (Throwable t) {
             throw new Exception(
-                    "The user defined 'open(Configuration)' method in "
+                    "The user defined 'open(OpenContext)' method in "
                             + stub.getClass().toString()
                             + " caused an exception: "
                             + t.getMessage(),
diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/api/checkpoint/CheckpointedFunction.java b/flink-runtime/src/main/java/org/apache/flink/streaming/api/checkpoint/CheckpointedFunction.java
@@ -131,7 +131,7 @@
  *
  *     private ValueState<Long> count;
  *
- *     public void open(Configuration cfg) throws Exception {
+ *     public void open(OpenContext ctx) throws Exception {
  *         count = getRuntimeContext().getState(new ValueStateDescriptor<>("myCount", Long.class));
  *     }
  *

Original file line number	Diff line number	Diff line change
`@@ -160,7 +160,7 @@ public class CountWindowAverage extends RichFlatMapFunction<Tuple2<Long, Long>,`
`160`	`160`	`}`
`161`	`161`
`162`	`162`	`@Override`
`163`		`- public void open(Configuration config) {`
	`163`	`+ public void open(OpenContext ctx) {`
`164`	`164`	`ValueStateDescriptor<Tuple2<Long, Long>> descriptor =`
`165`	`165`	`new ValueStateDescriptor<>(`
`166`	`166`	`"average", // the state name`
Original file line number	Diff line number	Diff line change
`@@ -192,7 +192,7 @@ public class CountWindowAverage extends RichFlatMapFunction<Tuple2<Long, Long>,`
`192`	`192`	`}`
`193`	`193`
`194`	`194`	`@Override`
`195`		`- public void open(Configuration config) {`
	`195`	`+ public void open(OpenContext ctx) {`
`196`	`196`	`ValueStateDescriptor<Tuple2<Long, Long>> descriptor =`
`197`	`197`	`new ValueStateDescriptor<>(`
`198`	`198`	`"average", // the state name`
Original file line number	Diff line number	Diff line change
`@@ -41,7 +41,7 @@`
`41`	`41`	`*`
`42`	`42`	`* private Map<Long, String> map;`
`43`	`43`	`*`
`44`		`- * public void open(Configuration cfg) throws Exception {`
	`44`	`+ * public void open(OpenContext ctx) throws Exception {`
`45`	`45`	`* getRuntimeContext().getBroadcastVariableWithInitializer("mapvar",`
`46`	`46`	`* new BroadcastVariableInitializer<Tuple2<Long, String>, Map<Long, String>>() {`
`47`	`47`	`*`