Skip to content

Commit 270be74

Browse files
schonglooschongloo
committed
Fixing a race condition in the configMap writes
Replacing the hardcoded subtaskIndex guard for configMap writes with a timer based approach.
1 parent ea4c2e4 commit 270be74

4 files changed

Lines changed: 132 additions & 53 deletions

File tree

‎flink-kubernetes-operator-bluegreen-client/pom.xml‎

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -104,4 +104,22 @@ under the License.
104104
</dependency>
105105
</dependencies>
106106

107+
<build>
108+
<plugins>
109+
<plugin>
110+
<groupId>org.apache.maven.plugins</groupId>
111+
<artifactId>maven-compiler-plugin</artifactId>
112+
<configuration>
113+
<annotationProcessorPaths>
114+
<path>
115+
<groupId>org.projectlombok</groupId>
116+
<artifactId>lombok</artifactId>
117+
<version>${lombok.version}</version>
118+
</path>
119+
</annotationProcessorPaths>
120+
</configuration>
121+
</plugin>
122+
</plugins>
123+
</build>
124+
107125
</project>

‎flink-kubernetes-operator-bluegreen-client/src/main/java/org/apache/flink/kubernetes/operator/bluegreen/client/GateProcessFunction.java‎

Lines changed: 76 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,7 @@
2727
import org.apache.flink.util.Preconditions;
2828

2929
import io.fabric8.kubernetes.api.model.ConfigMap;
30+
import io.fabric8.kubernetes.client.KubernetesClientException;
3031
import io.fabric8.kubernetes.client.informers.ResourceEventHandler;
3132
import org.slf4j.Logger;
3233
import org.slf4j.LoggerFactory;
@@ -46,11 +47,18 @@ abstract class GateProcessFunction<I> extends ProcessFunction<I, I> implements S
4647

4748
protected final BlueGreenDeploymentType blueGreenDeploymentType;
4849

49-
// TODO: make this configurable? This cannot be a constant
50-
protected final int subtaskIndexGuide = 1;
50+
// How long (ms) to wait after a write condition is first detected before writing to the
51+
// ConfigMap. This window gives other subtasks a chance to observe the write via the informer,
52+
// reducing duplicate K8s API calls.
53+
private static final long WRITE_DEDUP_DELAY_MS = 500L;
54+
55+
// Processing time (ms) when a pending write was first requested; -1 means no write pending.
56+
private long pendingWriteSince = -1L;
57+
// Set by notifyClearToTeardown() so that maybePerformPendingWrite() performs the teardown
58+
// write even if the pending write was originally scheduled for a different operation.
59+
private boolean pendingClearToTeardown = false;
5160

5261
private GateKubernetesService gateKubernetesService;
53-
private Boolean clearToTeardown = false;
5462
protected GateContext baseContext;
5563
private String namespace;
5664
private String configMapName;
@@ -79,6 +87,17 @@ public void open(Configuration parameters) throws Exception {
7987
processConfigMap(gateKubernetesService.parseConfigMap());
8088
}
8189

90+
/**
91+
* Records the current processing time as the start of a pending write window. No-op if a write
92+
* is already pending. The actual write is performed by {@link #maybePerformPendingWrite} on the
93+
* next element processed after the dedup delay has elapsed.
94+
*/
95+
protected final void scheduleWriteTimer(Context ctx) {
96+
if (pendingWriteSince < 0) {
97+
pendingWriteSince = ctx.timerService().currentProcessingTime();
98+
}
99+
}
100+
82101
protected abstract void processElementActive(
83102
I value, ProcessFunction<I, I>.Context ctx, Collector<I> out)
84103
throws IllegalAccessException;
@@ -89,7 +108,8 @@ protected abstract void processElementStandby(
89108

90109
@Override
91110
public void processElement(I value, ProcessFunction<I, I>.Context ctx, Collector<I> out)
92-
throws IllegalAccessException {
111+
throws Exception {
112+
maybePerformPendingWrite(ctx);
93113
switch (baseContext.getOutputMode()) {
94114
case ACTIVE:
95115
processElementActive(value, ctx, out);
@@ -104,6 +124,29 @@ public void processElement(I value, ProcessFunction<I, I>.Context ctx, Collector
104124
}
105125
}
106126

127+
private void maybePerformPendingWrite(Context ctx) throws Exception {
128+
if (pendingWriteSince < 0) {
129+
return;
130+
}
131+
long now = ctx.timerService().currentProcessingTime();
132+
if (now - pendingWriteSince < WRITE_DEDUP_DELAY_MS) {
133+
return;
134+
}
135+
pendingWriteSince = -1L;
136+
boolean handled = handleScheduledWrite(ctx);
137+
if (!handled && pendingClearToTeardown) {
138+
pendingClearToTeardown = false;
139+
if (baseContext.getGateStage() != CLEAR_TO_TEARDOWN) {
140+
logInfo("Writing " + CLEAR_TO_TEARDOWN + " to ConfigMap");
141+
gateKubernetesService.updateConfigMapEntries(
142+
Map.of(TRANSITION_STAGE.getLabel(), CLEAR_TO_TEARDOWN.toString()));
143+
logInfo(CLEAR_TO_TEARDOWN + " set!");
144+
} else {
145+
logInfo(CLEAR_TO_TEARDOWN + " already set, skipping");
146+
}
147+
}
148+
}
149+
107150
private void setKubernetesEnvironment() {
108151
this.gateKubernetesService = new GateKubernetesService(namespace, configMapName);
109152

@@ -161,19 +204,22 @@ private void processConfigMap(ConfigMap configMap) {
161204
onContextUpdate(baseContext, filteredData);
162205
}
163206

164-
protected final void notifyClearToTeardown() {
165-
// Notify only once
166-
if (!clearToTeardown && getRuntimeContext().getIndexOfThisSubtask() == subtaskIndexGuide) {
167-
logInfo("Setting " + CLEAR_TO_TEARDOWN);
168-
gateKubernetesService.updateConfigMapEntries(
169-
Map.of(TRANSITION_STAGE.getLabel(), CLEAR_TO_TEARDOWN.toString()));
170-
logInfo(CLEAR_TO_TEARDOWN + " set!");
171-
this.clearToTeardown = true;
172-
}
207+
protected final void notifyClearToTeardown(Context ctx) {
208+
pendingClearToTeardown = true;
209+
scheduleWriteTimer(ctx);
210+
}
211+
212+
/**
213+
* Hook for subclasses to handle watermark-specific ConfigMap writes when the dedup window
214+
* elapses. Returns {@code true} if the write was handled (base-class CLEAR_TO_TEARDOWN logic is
215+
* skipped), {@code false} to fall through.
216+
*/
217+
protected boolean handleScheduledWrite(Context ctx) throws Exception {
218+
return false;
173219
}
174220

175221
protected final void updateConfigMapCustomEntries(Map<String, String> customEntries)
176-
throws IllegalAccessException {
222+
throws Exception {
177223
// Validating only "custom" entries/keys can be updated
178224
var keysToUpdate = customEntries.keySet();
179225
var baseContextKeys =
@@ -189,7 +235,22 @@ protected final void updateConfigMapCustomEntries(Map<String, String> customEntr
189235
throw new IllegalAccessException(error);
190236
}
191237
logInfo("Updating custom entries: " + customEntries);
192-
gateKubernetesService.updateConfigMapEntries(customEntries);
238+
int maxRetries = 3;
239+
for (int attempt = 0; attempt <= maxRetries; attempt++) {
240+
try {
241+
gateKubernetesService.updateConfigMapEntries(customEntries);
242+
return;
243+
} catch (KubernetesClientException e) {
244+
if (e.getCode() == 409 && attempt < maxRetries) {
245+
logInfo(
246+
"ConfigMap update conflict on attempt "
247+
+ (attempt + 1)
248+
+ ", retrying...");
249+
} else {
250+
throw e;
251+
}
252+
}
253+
}
193254
}
194255

195256
// Temporary "utility" function for development

‎flink-kubernetes-operator-bluegreen-client/src/main/java/org/apache/flink/kubernetes/operator/bluegreen/client/WatermarkGateProcessFunction.java‎

Lines changed: 34 additions & 26 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,7 @@
2020
import org.apache.flink.kubernetes.operator.api.bluegreen.BlueGreenDeploymentType;
2121
import org.apache.flink.kubernetes.operator.api.bluegreen.GateContext;
2222
import org.apache.flink.kubernetes.operator.api.bluegreen.TransitionStage;
23+
import org.apache.flink.streaming.api.TimerService;
2324
import org.apache.flink.streaming.api.functions.ProcessFunction;
2425
import org.apache.flink.util.Collector;
2526
import org.apache.flink.util.Preconditions;
@@ -36,8 +37,6 @@ public class WatermarkGateProcessFunction<I> extends GateProcessFunction<I>
3637

3738
private WatermarkGateContext currentWatermarkGateContext;
3839

39-
private boolean waitingForWatermark = false;
40-
4140
WatermarkGateProcessFunction(
4241
BlueGreenDeploymentType blueGreenDeploymentType,
4342
String namespace,
@@ -96,7 +95,7 @@ protected void processElementActive(
9695
var currentGateStage = currentWatermarkGateContext.getBaseContext().getGateStage();
9796
if (currentGateStage == TransitionStage.TRANSITIONING) {
9897
logInfo(" -- Waiting for WM to be set - ");
99-
notifyWaitingForWatermark();
98+
notifyWaitingForWatermark(ctx);
10099
} else {
101100
logInfo("Waiting for the TRANSITIONING state, current: " + currentGateStage);
102101
}
@@ -110,7 +109,7 @@ protected void processElementStandby(
110109
if (currentWatermarkGateContext.getWatermarkToggleValue() != null) {
111110
var watermarkToggleValue = currentWatermarkGateContext.getWatermarkToggleValue();
112111

113-
if (getWatermarkBoundary(ctx) <= watermarkToggleValue) {
112+
if (getWatermarkBoundary(ctx.timerService()) <= watermarkToggleValue) {
114113
if (watermarkToggleValue > watermarkExtractor.apply(value)) {
115114
// Should still output the element
116115
out.collect(value);
@@ -121,7 +120,7 @@ protected void processElementStandby(
121120
} else {
122121
// Went past the Watermark Boundary: BLOCK ELEMENT
123122
logInfo(" -- Past WM Boundary -- ");
124-
notifyClearToTeardown();
123+
notifyClearToTeardown(ctx);
125124
}
126125
} else {
127126
// This ACTIVE job is transitioning to STANDBY, output elements
@@ -131,27 +130,33 @@ protected void processElementStandby(
131130
}
132131
}
133132

134-
private long getWatermarkBoundary(ProcessFunction<I, I>.Context ctx) {
135-
return ctx.timerService().currentWatermark() > 0
136-
? ctx.timerService().currentWatermark()
137-
: ctx.timerService().currentProcessingTime();
133+
private long getWatermarkBoundary(TimerService timerService) {
134+
return timerService.currentWatermark() > 0
135+
? timerService.currentWatermark()
136+
: timerService.currentProcessingTime();
138137
}
139138

140-
protected void updateWatermarkInConfigMap(ProcessFunction<I, I>.Context ctx)
141-
throws IllegalAccessException {
142-
var curWmCtx = currentWatermarkGateContext;
139+
protected void updateWatermarkInConfigMap(Context ctx) {
140+
scheduleWriteTimer(ctx);
141+
}
143142

144-
if (getRuntimeContext().getIndexOfThisSubtask() == subtaskIndexGuide
145-
&& curWmCtx.getWatermarkGateStage() == WatermarkGateStage.WAITING_FOR_WATERMARK) {
146-
var nextWatermarkToggleValue =
147-
getWatermarkBoundary(ctx)
148-
+ curWmCtx.getBaseContext().getDeploymentTeardownDelayMs();
143+
protected void notifyWaitingForWatermark(Context ctx) {
144+
scheduleWriteTimer(ctx);
145+
}
149146

150-
// Setting the value in advance to avoid subsequent elements in this subtask from
151-
// setting it
152-
// while the changes get reflected in Kubernetes
153-
currentWatermarkGateContext.setWatermarkToggleValue(nextWatermarkToggleValue);
147+
@Override
148+
protected boolean handleScheduledWrite(Context ctx) throws Exception {
149+
var wmCtx = currentWatermarkGateContext;
154150

151+
// Standby job: Active job has signalled it's waiting — compute and write the WM toggle.
152+
if (wmCtx.getWatermarkGateStage() == WatermarkGateStage.WAITING_FOR_WATERMARK
153+
&& wmCtx.getWatermarkToggleValue() == null) {
154+
var nextWatermarkToggleValue =
155+
getWatermarkBoundary(ctx.timerService())
156+
+ wmCtx.getBaseContext().getDeploymentTeardownDelayMs();
157+
// Set optimistically so subsequent elements on this subtask don't reschedule
158+
// before the ConfigMap informer propagates.
159+
currentWatermarkGateContext.setWatermarkToggleValue(nextWatermarkToggleValue);
155160
logInfo("Updating the ConfigMap Watermark value to: " + nextWatermarkToggleValue);
156161
updateConfigMapCustomEntries(
157162
Map.of(
@@ -160,19 +165,22 @@ protected void updateWatermarkInConfigMap(ProcessFunction<I, I>.Context ctx)
160165
WatermarkGateContext.WATERMARK_STAGE,
161166
WatermarkGateStage.WATERMARK_SET.toString()));
162167
logInfo("Watermark updated!");
168+
return true;
163169
}
164-
}
165170

166-
protected void notifyWaitingForWatermark() throws IllegalAccessException {
167-
if (!waitingForWatermark
168-
&& getRuntimeContext().getIndexOfThisSubtask() == subtaskIndexGuide) {
171+
// Active job: signal that it is waiting for the Standby job to provide the WM toggle.
172+
if (wmCtx.getBaseContext().getGateStage() == TransitionStage.TRANSITIONING
173+
&& wmCtx.getWatermarkToggleValue() == null
174+
&& wmCtx.getWatermarkGateStage() != WatermarkGateStage.WAITING_FOR_WATERMARK) {
169175
logInfo("Setting " + WatermarkGateStage.WAITING_FOR_WATERMARK);
170176
updateConfigMapCustomEntries(
171177
Map.of(
172178
WatermarkGateContext.WATERMARK_STAGE,
173179
WatermarkGateStage.WAITING_FOR_WATERMARK.toString()));
174180
logInfo(WatermarkGateStage.WAITING_FOR_WATERMARK + " set!");
175-
waitingForWatermark = true;
181+
return true;
176182
}
183+
184+
return false; // fall through to base-class CLEAR_TO_TEARDOWN handling
177185
}
178186
}

‎flink-kubernetes-operator-bluegreen-client/src/test/java/org/apache/flink/kubernetes/operator/bluegreen/client/WatermarkGateProcessFunctionTest.java‎

Lines changed: 4 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -21,7 +21,6 @@
2121
import org.apache.flink.kubernetes.operator.api.bluegreen.GateContext;
2222
import org.apache.flink.kubernetes.operator.api.bluegreen.GateContextOptions;
2323
import org.apache.flink.kubernetes.operator.api.bluegreen.TransitionStage;
24-
import org.apache.flink.streaming.api.functions.ProcessFunction;
2524
import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness;
2625
import org.apache.flink.streaming.util.ProcessFunctionTestHarnesses;
2726

@@ -356,7 +355,6 @@ private static class TestWatermarkGateProcessFunction
356355
private GateContext mockBaseContext;
357356

358357
public boolean notifyWaitingForWatermarkCalled = false;
359-
public boolean notifyClearToTeardownCalled = false;
360358
public boolean updateWatermarkInConfigMapCalled = false;
361359

362360
TestWatermarkGateProcessFunction(
@@ -401,21 +399,15 @@ protected void logInfo(String message) {
401399
logMessages.add(message);
402400
}
403401

404-
/**
405-
* Override to capture the call without invoking Kubernetes or the subtask-index guard. In
406-
* production, only one subtask writes to the ConfigMap; in tests, all subtask indexes are
407-
* 0, so the guard would always suppress the call.
408-
*/
402+
/** Override to capture the call without invoking Kubernetes. */
409403
@Override
410-
protected void notifyWaitingForWatermark() throws IllegalAccessException {
404+
protected void notifyWaitingForWatermark(Context ctx) {
411405
notifyWaitingForWatermarkCalled = true;
412406
}
413407

414-
/** Override to capture the call without invoking Kubernetes or the subtask-index guard. */
408+
/** Override to capture the call without invoking Kubernetes. */
415409
@Override
416-
protected void updateWatermarkInConfigMap(
417-
ProcessFunction<TestMessage, TestMessage>.Context ctx)
418-
throws IllegalAccessException {
410+
protected void updateWatermarkInConfigMap(Context ctx) {
419411
updateWatermarkInConfigMapCalled = true;
420412
}
421413

0 commit comments

Comments
 (0)