[jira] [Comment Edited] (FLINK-25691) ElasticsearchSinkITCase.testElasticsearchSink fails on AZP
[
https://issues.apache.org/jira/browse/FLINK-25691?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17526768#comment-17526768
]
macdoor615 edited comment on FLINK-25691 at 4/23/22 5:25 AM:
-
[~alexanderpreuss] [~martijnvisser]
We are facing the same problem with 1.14/1.15 and can not reproduce.
Maybe you can refer to this case.
He suspects that the ES cluster will cause the http long connection dead during
the OldGC, and will not create a new connection, causing the data refresh ES to
time out.
[https://blog.csdn.net/HugeBitter/article/details/119823989]
was (Author: macdoor615):
[~alexanderpreuss] [~martijnvisser]
We are facing the same problem and can not reproduce.
Maybe you can refer to this case.
He suspects that the ES cluster will cause the http long connection dead during
the OldGC, and will not create a new connection, causing the data refresh ES to
time out.
https://blog.csdn.net/HugeBitter/article/details/119823989
> ElasticsearchSinkITCase.testElasticsearchSink fails on AZP
> --
>
> Key: FLINK-25691
> URL: https://issues.apache.org/jira/browse/FLINK-25691
> Project: Flink
> Issue Type: Bug
> Components: Connectors / ElasticSearch
>Affects Versions: 1.15.0
>Reporter: Till Rohrmann
>Priority: Major
> Labels: auto-deprioritized-critical, stale-major, test-stability
>
> The test {{ElasticsearchSinkITCase.testElasticsearchSink}} fails on AZP with
> {code}
> 2022-01-18T08:10:11.9777311Z Jan 18 08:10:11 [ERROR]
> org.apache.flink.streaming.connectors.elasticsearch7.ElasticsearchSinkITCase.testElasticsearchSink
> Time elapsed: 31.816 s <<< ERROR!
> 2022-01-18T08:10:11.9778438Z Jan 18 08:10:11
> org.apache.flink.runtime.client.JobExecutionException: Job execution failed.
> 2022-01-18T08:10:11.9779184Z Jan 18 08:10:11 at
> org.apache.flink.runtime.jobmaster.JobResult.toJobExecutionResult(JobResult.java:144)
> 2022-01-18T08:10:11.9779993Z Jan 18 08:10:11 at
> org.apache.flink.runtime.minicluster.MiniClusterJobClient.lambda$getJobExecutionResult$3(MiniClusterJobClient.java:137)
> 2022-01-18T08:10:11.9780892Z Jan 18 08:10:11 at
> java.util.concurrent.CompletableFuture.uniApply(CompletableFuture.java:616)
> 2022-01-18T08:10:11.9781726Z Jan 18 08:10:11 at
> java.util.concurrent.CompletableFuture$UniApply.tryFire(CompletableFuture.java:591)
> 2022-01-18T08:10:11.9782380Z Jan 18 08:10:11 at
> java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:488)
> 2022-01-18T08:10:11.9783097Z Jan 18 08:10:11 at
> java.util.concurrent.CompletableFuture.complete(CompletableFuture.java:1975)
> 2022-01-18T08:10:11.9783866Z Jan 18 08:10:11 at
> org.apache.flink.runtime.rpc.akka.AkkaInvocationHandler.lambda$invokeRpc$1(AkkaInvocationHandler.java:258)
> 2022-01-18T08:10:11.9784615Z Jan 18 08:10:11 at
> java.util.concurrent.CompletableFuture.uniWhenComplete(CompletableFuture.java:774)
> 2022-01-18T08:10:11.9791362Z Jan 18 08:10:11 at
> java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(CompletableFuture.java:750)
> 2022-01-18T08:10:11.9792139Z Jan 18 08:10:11 at
> java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:488)
> 2022-01-18T08:10:11.9793011Z Jan 18 08:10:11 at
> java.util.concurrent.CompletableFuture.complete(CompletableFuture.java:1975)
> 2022-01-18T08:10:11.9793620Z Jan 18 08:10:11 at
> org.apache.flink.util.concurrent.FutureUtils.doForward(FutureUtils.java:1389)
> 2022-01-18T08:10:11.9794267Z Jan 18 08:10:11 at
> org.apache.flink.runtime.concurrent.akka.ClassLoadingUtils.lambda$null$1(ClassLoadingUtils.java:93)
> 2022-01-18T08:10:11.9795177Z Jan 18 08:10:11 at
> org.apache.flink.runtime.concurrent.akka.ClassLoadingUtils.runWithContextClassLoader(ClassLoadingUtils.java:68)
> 2022-01-18T08:10:11.9796451Z Jan 18 08:10:11 at
> org.apache.flink.runtime.concurrent.akka.ClassLoadingUtils.lambda$guardCompletionWithContextClassLoader$2(ClassLoadingUtils.java:92)
> 2022-01-18T08:10:11.9797325Z Jan 18 08:10:11 at
> java.util.concurrent.CompletableFuture.uniWhenComplete(CompletableFuture.java:774)
> 2022-01-18T08:10:11.9798108Z Jan 18 08:10:11 at
> java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(CompletableFuture.java:750)
> 2022-01-18T08:10:11.9798749Z Jan 18 08:10:11 at
> java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:488)
> 2022-01-18T08:10:11.9799364Z Jan 18 08:10:11 at
> java.util.concurrent.CompletableFuture.complete(CompletableFuture.java:1975)
> 2022-01-18T08:10:11.970Z Jan 18 08:10:11 at
> org.apache.flink.runtime.concurrent.akka.AkkaFutureUtils$1.onComplete(AkkaFutureUtils.java:47)
> 2022-01-18T08:10:11.9800561Z Jan 18 08:10:11 at
> akka.dispatch.OnComplete.internal(Future.scala:300)
>
[jira] [Commented] (FLINK-25691) ElasticsearchSinkITCase.testElasticsearchSink fails on AZP
[
https://issues.apache.org/jira/browse/FLINK-25691?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17526768#comment-17526768
]
macdoor615 commented on FLINK-25691:
[~alexanderpreuss] [~martijnvisser]
We are facing the same problem and can not reproduce.
Maybe you can refer to this case.
He suspects that the ES cluster will cause the http long connection dead during
the OldGC, and will not create a new connection, causing the data refresh ES to
time out.
https://blog.csdn.net/HugeBitter/article/details/119823989
> ElasticsearchSinkITCase.testElasticsearchSink fails on AZP
> --
>
> Key: FLINK-25691
> URL: https://issues.apache.org/jira/browse/FLINK-25691
> Project: Flink
> Issue Type: Bug
> Components: Connectors / ElasticSearch
>Affects Versions: 1.15.0
>Reporter: Till Rohrmann
>Priority: Major
> Labels: auto-deprioritized-critical, stale-major, test-stability
>
> The test {{ElasticsearchSinkITCase.testElasticsearchSink}} fails on AZP with
> {code}
> 2022-01-18T08:10:11.9777311Z Jan 18 08:10:11 [ERROR]
> org.apache.flink.streaming.connectors.elasticsearch7.ElasticsearchSinkITCase.testElasticsearchSink
> Time elapsed: 31.816 s <<< ERROR!
> 2022-01-18T08:10:11.9778438Z Jan 18 08:10:11
> org.apache.flink.runtime.client.JobExecutionException: Job execution failed.
> 2022-01-18T08:10:11.9779184Z Jan 18 08:10:11 at
> org.apache.flink.runtime.jobmaster.JobResult.toJobExecutionResult(JobResult.java:144)
> 2022-01-18T08:10:11.9779993Z Jan 18 08:10:11 at
> org.apache.flink.runtime.minicluster.MiniClusterJobClient.lambda$getJobExecutionResult$3(MiniClusterJobClient.java:137)
> 2022-01-18T08:10:11.9780892Z Jan 18 08:10:11 at
> java.util.concurrent.CompletableFuture.uniApply(CompletableFuture.java:616)
> 2022-01-18T08:10:11.9781726Z Jan 18 08:10:11 at
> java.util.concurrent.CompletableFuture$UniApply.tryFire(CompletableFuture.java:591)
> 2022-01-18T08:10:11.9782380Z Jan 18 08:10:11 at
> java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:488)
> 2022-01-18T08:10:11.9783097Z Jan 18 08:10:11 at
> java.util.concurrent.CompletableFuture.complete(CompletableFuture.java:1975)
> 2022-01-18T08:10:11.9783866Z Jan 18 08:10:11 at
> org.apache.flink.runtime.rpc.akka.AkkaInvocationHandler.lambda$invokeRpc$1(AkkaInvocationHandler.java:258)
> 2022-01-18T08:10:11.9784615Z Jan 18 08:10:11 at
> java.util.concurrent.CompletableFuture.uniWhenComplete(CompletableFuture.java:774)
> 2022-01-18T08:10:11.9791362Z Jan 18 08:10:11 at
> java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(CompletableFuture.java:750)
> 2022-01-18T08:10:11.9792139Z Jan 18 08:10:11 at
> java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:488)
> 2022-01-18T08:10:11.9793011Z Jan 18 08:10:11 at
> java.util.concurrent.CompletableFuture.complete(CompletableFuture.java:1975)
> 2022-01-18T08:10:11.9793620Z Jan 18 08:10:11 at
> org.apache.flink.util.concurrent.FutureUtils.doForward(FutureUtils.java:1389)
> 2022-01-18T08:10:11.9794267Z Jan 18 08:10:11 at
> org.apache.flink.runtime.concurrent.akka.ClassLoadingUtils.lambda$null$1(ClassLoadingUtils.java:93)
> 2022-01-18T08:10:11.9795177Z Jan 18 08:10:11 at
> org.apache.flink.runtime.concurrent.akka.ClassLoadingUtils.runWithContextClassLoader(ClassLoadingUtils.java:68)
> 2022-01-18T08:10:11.9796451Z Jan 18 08:10:11 at
> org.apache.flink.runtime.concurrent.akka.ClassLoadingUtils.lambda$guardCompletionWithContextClassLoader$2(ClassLoadingUtils.java:92)
> 2022-01-18T08:10:11.9797325Z Jan 18 08:10:11 at
> java.util.concurrent.CompletableFuture.uniWhenComplete(CompletableFuture.java:774)
> 2022-01-18T08:10:11.9798108Z Jan 18 08:10:11 at
> java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(CompletableFuture.java:750)
> 2022-01-18T08:10:11.9798749Z Jan 18 08:10:11 at
> java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:488)
> 2022-01-18T08:10:11.9799364Z Jan 18 08:10:11 at
> java.util.concurrent.CompletableFuture.complete(CompletableFuture.java:1975)
> 2022-01-18T08:10:11.970Z Jan 18 08:10:11 at
> org.apache.flink.runtime.concurrent.akka.AkkaFutureUtils$1.onComplete(AkkaFutureUtils.java:47)
> 2022-01-18T08:10:11.9800561Z Jan 18 08:10:11 at
> akka.dispatch.OnComplete.internal(Future.scala:300)
> 2022-01-18T08:10:11.9801061Z Jan 18 08:10:11 at
> akka.dispatch.OnComplete.internal(Future.scala:297)
> 2022-01-18T08:10:11.9801661Z Jan 18 08:10:11 at
> akka.dispatch.japi$CallbackBridge.apply(Future.scala:224)
> 2022-01-18T08:10:11.9802186Z Jan 18 08:10:11 at
> akka.dispatch.japi$CallbackBridge.apply(Future.scala:221)
> 2022-01-18T08:10:11.9802713Z Jan 18 08:10:11 at
> scala.concurrent.impl.CallbackRunnable.run(Promise.scala:60)
> 2022-01-18T08:10:11.9803348Z
[jira] [Commented] (FLINK-27359) Kubernetes operator throws NPE when testing with Flink 1.15
[
https://issues.apache.org/jira/browse/FLINK-27359?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17526760#comment-17526760
]
Yang Wang commented on FLINK-27359:
---
Duplicated due to bad network.
> Kubernetes operator throws NPE when testing with Flink 1.15
> ---
>
> Key: FLINK-27359
> URL: https://issues.apache.org/jira/browse/FLINK-27359
> Project: Flink
> Issue Type: Bug
> Components: Kubernetes Operator
>Reporter: Yang Wang
>Priority: Major
> Fix For: kubernetes-operator-1.0.0
>
>
> {code:java}
> 2022-04-22 10:19:18,307 o.a.f.k.o.c.FlinkDeploymentController [WARN
> ][default/flink-example-statemachine] Attempt count: 5, last attempt: true
> 2022-04-22 10:19:18,329 i.j.o.p.e.ReconciliationDispatcher
> [ERROR][default/flink-example-statemachine] Error during event processing
> ExecutionScope{ resource id:
> CustomResourceID{name='flink-example-statemachine', namespace='default'},
> version: 4979543} failed.
> org.apache.flink.kubernetes.operator.exception.ReconciliationException:
> java.lang.NullPointerException
> at
> org.apache.flink.kubernetes.operator.controller.FlinkDeploymentController.reconcile(FlinkDeploymentController.java:110)
> at
> org.apache.flink.kubernetes.operator.controller.FlinkDeploymentController.reconcile(FlinkDeploymentController.java:53)
> at
> io.javaoperatorsdk.operator.processing.Controller$2.execute(Controller.java:101)
> at
> io.javaoperatorsdk.operator.processing.Controller$2.execute(Controller.java:76)
> at
> io.javaoperatorsdk.operator.api.monitoring.Metrics.timeControllerExecution(Metrics.java:34)
> at
> io.javaoperatorsdk.operator.processing.Controller.reconcile(Controller.java:75)
> at
> io.javaoperatorsdk.operator.processing.event.ReconciliationDispatcher.reconcileExecution(ReconciliationDispatcher.java:143)
> at
> io.javaoperatorsdk.operator.processing.event.ReconciliationDispatcher.handleReconcile(ReconciliationDispatcher.java:109)
> at
> io.javaoperatorsdk.operator.processing.event.ReconciliationDispatcher.handleDispatch(ReconciliationDispatcher.java:74)
> at
> io.javaoperatorsdk.operator.processing.event.ReconciliationDispatcher.handleExecution(ReconciliationDispatcher.java:50)
> at
> io.javaoperatorsdk.operator.processing.event.EventProcessor$ControllerExecution.run(EventProcessor.java:349)
> at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown
> Source)
> at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown
> Source)
> at java.base/java.lang.Thread.run(Unknown Source)
> Caused by: java.lang.NullPointerException
> at
> org.apache.flink.kubernetes.operator.utils.FlinkUtils.lambda$deleteJobGraphInKubernetesHA$0(FlinkUtils.java:253)
> at java.base/java.util.ArrayList.forEach(Unknown Source)
> at
> org.apache.flink.kubernetes.operator.utils.FlinkUtils.deleteJobGraphInKubernetesHA(FlinkUtils.java:248)
> at
> org.apache.flink.kubernetes.operator.service.FlinkService.submitApplicationCluster(FlinkService.java:130)
> at
> org.apache.flink.kubernetes.operator.reconciler.deployment.ApplicationReconciler.deployFlinkJob(ApplicationReconciler.java:205)
> at
> org.apache.flink.kubernetes.operator.reconciler.deployment.ApplicationReconciler.restoreFromLastSavepoint(ApplicationReconciler.java:218)
> at
> org.apache.flink.kubernetes.operator.reconciler.deployment.ApplicationReconciler.reconcile(ApplicationReconciler.java:117)
> at
> org.apache.flink.kubernetes.operator.reconciler.deployment.ApplicationReconciler.reconcile(ApplicationReconciler.java:56)
> at
> org.apache.flink.kubernetes.operator.controller.FlinkDeploymentController.reconcile(FlinkDeploymentController.java:106)
> ... 13 more {code}
> The root cause is that the Kubernetes HA implementation has changed from
> 1.15. When the job is cancelled, the data of leader ConfigMap will be
> cleared.
--
This message was sent by Atlassian Jira
(v8.20.7#820007)
[jira] [Closed] (FLINK-27359) Kubernetes operator throws NPE when testing with Flink 1.15
[
https://issues.apache.org/jira/browse/FLINK-27359?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Yang Wang closed FLINK-27359.
-
Resolution: Duplicate
> Kubernetes operator throws NPE when testing with Flink 1.15
> ---
>
> Key: FLINK-27359
> URL: https://issues.apache.org/jira/browse/FLINK-27359
> Project: Flink
> Issue Type: Bug
> Components: Kubernetes Operator
>Reporter: Yang Wang
>Priority: Major
> Fix For: kubernetes-operator-1.0.0
>
>
> {code:java}
> 2022-04-22 10:19:18,307 o.a.f.k.o.c.FlinkDeploymentController [WARN
> ][default/flink-example-statemachine] Attempt count: 5, last attempt: true
> 2022-04-22 10:19:18,329 i.j.o.p.e.ReconciliationDispatcher
> [ERROR][default/flink-example-statemachine] Error during event processing
> ExecutionScope{ resource id:
> CustomResourceID{name='flink-example-statemachine', namespace='default'},
> version: 4979543} failed.
> org.apache.flink.kubernetes.operator.exception.ReconciliationException:
> java.lang.NullPointerException
> at
> org.apache.flink.kubernetes.operator.controller.FlinkDeploymentController.reconcile(FlinkDeploymentController.java:110)
> at
> org.apache.flink.kubernetes.operator.controller.FlinkDeploymentController.reconcile(FlinkDeploymentController.java:53)
> at
> io.javaoperatorsdk.operator.processing.Controller$2.execute(Controller.java:101)
> at
> io.javaoperatorsdk.operator.processing.Controller$2.execute(Controller.java:76)
> at
> io.javaoperatorsdk.operator.api.monitoring.Metrics.timeControllerExecution(Metrics.java:34)
> at
> io.javaoperatorsdk.operator.processing.Controller.reconcile(Controller.java:75)
> at
> io.javaoperatorsdk.operator.processing.event.ReconciliationDispatcher.reconcileExecution(ReconciliationDispatcher.java:143)
> at
> io.javaoperatorsdk.operator.processing.event.ReconciliationDispatcher.handleReconcile(ReconciliationDispatcher.java:109)
> at
> io.javaoperatorsdk.operator.processing.event.ReconciliationDispatcher.handleDispatch(ReconciliationDispatcher.java:74)
> at
> io.javaoperatorsdk.operator.processing.event.ReconciliationDispatcher.handleExecution(ReconciliationDispatcher.java:50)
> at
> io.javaoperatorsdk.operator.processing.event.EventProcessor$ControllerExecution.run(EventProcessor.java:349)
> at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown
> Source)
> at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown
> Source)
> at java.base/java.lang.Thread.run(Unknown Source)
> Caused by: java.lang.NullPointerException
> at
> org.apache.flink.kubernetes.operator.utils.FlinkUtils.lambda$deleteJobGraphInKubernetesHA$0(FlinkUtils.java:253)
> at java.base/java.util.ArrayList.forEach(Unknown Source)
> at
> org.apache.flink.kubernetes.operator.utils.FlinkUtils.deleteJobGraphInKubernetesHA(FlinkUtils.java:248)
> at
> org.apache.flink.kubernetes.operator.service.FlinkService.submitApplicationCluster(FlinkService.java:130)
> at
> org.apache.flink.kubernetes.operator.reconciler.deployment.ApplicationReconciler.deployFlinkJob(ApplicationReconciler.java:205)
> at
> org.apache.flink.kubernetes.operator.reconciler.deployment.ApplicationReconciler.restoreFromLastSavepoint(ApplicationReconciler.java:218)
> at
> org.apache.flink.kubernetes.operator.reconciler.deployment.ApplicationReconciler.reconcile(ApplicationReconciler.java:117)
> at
> org.apache.flink.kubernetes.operator.reconciler.deployment.ApplicationReconciler.reconcile(ApplicationReconciler.java:56)
> at
> org.apache.flink.kubernetes.operator.controller.FlinkDeploymentController.reconcile(FlinkDeploymentController.java:106)
> ... 13 more {code}
> The root cause is that the Kubernetes HA implementation has changed from
> 1.15. When the job is cancelled, the data of leader ConfigMap will be
> cleared.
--
This message was sent by Atlassian Jira
(v8.20.7#820007)
[jira] [Created] (FLINK-27359) Kubernetes operator throws NPE when testing with Flink 1.15
Yang Wang created FLINK-27359:
-
Summary: Kubernetes operator throws NPE when testing with Flink
1.15
Key: FLINK-27359
URL: https://issues.apache.org/jira/browse/FLINK-27359
Project: Flink
Issue Type: Bug
Components: Kubernetes Operator
Reporter: Yang Wang
Fix For: kubernetes-operator-1.0.0
{code:java}
2022-04-22 10:19:18,307 o.a.f.k.o.c.FlinkDeploymentController [WARN
][default/flink-example-statemachine] Attempt count: 5, last attempt: true
2022-04-22 10:19:18,329 i.j.o.p.e.ReconciliationDispatcher
[ERROR][default/flink-example-statemachine] Error during event processing
ExecutionScope{ resource id:
CustomResourceID{name='flink-example-statemachine', namespace='default'},
version: 4979543} failed.
org.apache.flink.kubernetes.operator.exception.ReconciliationException:
java.lang.NullPointerException
at
org.apache.flink.kubernetes.operator.controller.FlinkDeploymentController.reconcile(FlinkDeploymentController.java:110)
at
org.apache.flink.kubernetes.operator.controller.FlinkDeploymentController.reconcile(FlinkDeploymentController.java:53)
at
io.javaoperatorsdk.operator.processing.Controller$2.execute(Controller.java:101)
at
io.javaoperatorsdk.operator.processing.Controller$2.execute(Controller.java:76)
at
io.javaoperatorsdk.operator.api.monitoring.Metrics.timeControllerExecution(Metrics.java:34)
at
io.javaoperatorsdk.operator.processing.Controller.reconcile(Controller.java:75)
at
io.javaoperatorsdk.operator.processing.event.ReconciliationDispatcher.reconcileExecution(ReconciliationDispatcher.java:143)
at
io.javaoperatorsdk.operator.processing.event.ReconciliationDispatcher.handleReconcile(ReconciliationDispatcher.java:109)
at
io.javaoperatorsdk.operator.processing.event.ReconciliationDispatcher.handleDispatch(ReconciliationDispatcher.java:74)
at
io.javaoperatorsdk.operator.processing.event.ReconciliationDispatcher.handleExecution(ReconciliationDispatcher.java:50)
at
io.javaoperatorsdk.operator.processing.event.EventProcessor$ControllerExecution.run(EventProcessor.java:349)
at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown
Source)
at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown
Source)
at java.base/java.lang.Thread.run(Unknown Source)
Caused by: java.lang.NullPointerException
at
org.apache.flink.kubernetes.operator.utils.FlinkUtils.lambda$deleteJobGraphInKubernetesHA$0(FlinkUtils.java:253)
at java.base/java.util.ArrayList.forEach(Unknown Source)
at
org.apache.flink.kubernetes.operator.utils.FlinkUtils.deleteJobGraphInKubernetesHA(FlinkUtils.java:248)
at
org.apache.flink.kubernetes.operator.service.FlinkService.submitApplicationCluster(FlinkService.java:130)
at
org.apache.flink.kubernetes.operator.reconciler.deployment.ApplicationReconciler.deployFlinkJob(ApplicationReconciler.java:205)
at
org.apache.flink.kubernetes.operator.reconciler.deployment.ApplicationReconciler.restoreFromLastSavepoint(ApplicationReconciler.java:218)
at
org.apache.flink.kubernetes.operator.reconciler.deployment.ApplicationReconciler.reconcile(ApplicationReconciler.java:117)
at
org.apache.flink.kubernetes.operator.reconciler.deployment.ApplicationReconciler.reconcile(ApplicationReconciler.java:56)
at
org.apache.flink.kubernetes.operator.controller.FlinkDeploymentController.reconcile(FlinkDeploymentController.java:106)
... 13 more {code}
The root cause is that the Kubernetes HA implementation has changed from 1.15.
When the job is cancelled, the data of leader ConfigMap will be cleared.
--
This message was sent by Atlassian Jira
(v8.20.7#820007)
[jira] [Created] (FLINK-27358) Kubernetes operator throws NPE when testing with Flink 1.15
Yang Wang created FLINK-27358:
-
Summary: Kubernetes operator throws NPE when testing with Flink
1.15
Key: FLINK-27358
URL: https://issues.apache.org/jira/browse/FLINK-27358
Project: Flink
Issue Type: Bug
Components: Kubernetes Operator
Reporter: Yang Wang
Fix For: kubernetes-operator-1.0.0
{code:java}
2022-04-22 10:19:18,307 o.a.f.k.o.c.FlinkDeploymentController [WARN
][default/flink-example-statemachine] Attempt count: 5, last attempt: true
2022-04-22 10:19:18,329 i.j.o.p.e.ReconciliationDispatcher
[ERROR][default/flink-example-statemachine] Error during event processing
ExecutionScope{ resource id:
CustomResourceID{name='flink-example-statemachine', namespace='default'},
version: 4979543} failed.
org.apache.flink.kubernetes.operator.exception.ReconciliationException:
java.lang.NullPointerException
at
org.apache.flink.kubernetes.operator.controller.FlinkDeploymentController.reconcile(FlinkDeploymentController.java:110)
at
org.apache.flink.kubernetes.operator.controller.FlinkDeploymentController.reconcile(FlinkDeploymentController.java:53)
at
io.javaoperatorsdk.operator.processing.Controller$2.execute(Controller.java:101)
at
io.javaoperatorsdk.operator.processing.Controller$2.execute(Controller.java:76)
at
io.javaoperatorsdk.operator.api.monitoring.Metrics.timeControllerExecution(Metrics.java:34)
at
io.javaoperatorsdk.operator.processing.Controller.reconcile(Controller.java:75)
at
io.javaoperatorsdk.operator.processing.event.ReconciliationDispatcher.reconcileExecution(ReconciliationDispatcher.java:143)
at
io.javaoperatorsdk.operator.processing.event.ReconciliationDispatcher.handleReconcile(ReconciliationDispatcher.java:109)
at
io.javaoperatorsdk.operator.processing.event.ReconciliationDispatcher.handleDispatch(ReconciliationDispatcher.java:74)
at
io.javaoperatorsdk.operator.processing.event.ReconciliationDispatcher.handleExecution(ReconciliationDispatcher.java:50)
at
io.javaoperatorsdk.operator.processing.event.EventProcessor$ControllerExecution.run(EventProcessor.java:349)
at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown
Source)
at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown
Source)
at java.base/java.lang.Thread.run(Unknown Source)
Caused by: java.lang.NullPointerException
at
org.apache.flink.kubernetes.operator.utils.FlinkUtils.lambda$deleteJobGraphInKubernetesHA$0(FlinkUtils.java:253)
at java.base/java.util.ArrayList.forEach(Unknown Source)
at
org.apache.flink.kubernetes.operator.utils.FlinkUtils.deleteJobGraphInKubernetesHA(FlinkUtils.java:248)
at
org.apache.flink.kubernetes.operator.service.FlinkService.submitApplicationCluster(FlinkService.java:130)
at
org.apache.flink.kubernetes.operator.reconciler.deployment.ApplicationReconciler.deployFlinkJob(ApplicationReconciler.java:205)
at
org.apache.flink.kubernetes.operator.reconciler.deployment.ApplicationReconciler.restoreFromLastSavepoint(ApplicationReconciler.java:218)
at
org.apache.flink.kubernetes.operator.reconciler.deployment.ApplicationReconciler.reconcile(ApplicationReconciler.java:117)
at
org.apache.flink.kubernetes.operator.reconciler.deployment.ApplicationReconciler.reconcile(ApplicationReconciler.java:56)
at
org.apache.flink.kubernetes.operator.controller.FlinkDeploymentController.reconcile(FlinkDeploymentController.java:106)
... 13 more {code}
The root cause is that the Kubernetes HA implementation has changed from 1.15.
When the job is cancelled, the data of leader ConfigMap will be cleared.
--
This message was sent by Atlassian Jira
(v8.20.7#820007)
[GitHub] [flink] hehuiyuan commented on pull request #19297: [FLINK-26848][JDBC]Support to write data when disable flush-interval and max-rows
hehuiyuan commented on PR #19297: URL: https://github.com/apache/flink/pull/19297#issuecomment-1107312527 hi @fapaul @JingsongLi , have a look this pr. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[jira] [Updated] (FLINK-27357) In Flink HA Service on K8S, Web UI External Service should point to elected Job Manager leader's IP
[ https://issues.apache.org/jira/browse/FLINK-27357?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Jesus H Christ updated FLINK-27357: --- Description: Flink on Kubernetes has High Availability services which build an external service for the rest api access. [https://sourcegraph.com/github.com/apache/flink@master/-/blob/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/decorators/ExternalServiceDecorator.java] In the case of multiple job managers in reactive mode or to avoid 15s or so between job manager restarts in K8S High Availability services, a new job manager leader can be elected. In this case I think the external service we create for the web ui, and possibly even the REST API service, no longer points to the correct job manager as any of the job manager pods can have the jobmanager label. Is this correct? If so, it might help to use the endpoint containing the elected leader IP in the ConfigMap of one specific pod of the JobManager for the external service web ui, similar to how TaskManagers use JobManager IPs for High Availabiilty. It might also help to update the service or endpoint with the new IP to point to when updating the ConfigMap for JobManager leader election: [https://sourcegraph.com/github.com/apache/flink/-/blob/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/highavailability/KubernetesLeaderRetrievalDriverFactory.java] was: Flink on Kubernetes has High Availability services which build an external service for the rest api access. [https://sourcegraph.com/github.com/apache/flink@master/-/blob/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/decorators/ExternalServiceDecorator.java] In the case of multiple job managers in reactive mode or to avoid 15s or so between job manager restarts in K8S High Availability services, a new job manager leader can be elected. In this case I think the external service we create for the web ui, and possibly even the REST API service, no longer points to the correct job manager as any of the job manager pods can have the jobmanager label. It might help to use the endpoint containing the elected leader IP in the ConfigMap of one specific pod of the JobManager for the external service web ui, similar to how TaskManagers use JobManager IPs for High Availabiilty. It might also help to update the service or endpoint with the new IP to point to when updating the ConfigMap for JobManager leader election: [https://sourcegraph.com/github.com/apache/flink/-/blob/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/highavailability/KubernetesLeaderRetrievalDriverFactory.java] > In Flink HA Service on K8S, Web UI External Service should point to elected > Job Manager leader's IP > --- > > Key: FLINK-27357 > URL: https://issues.apache.org/jira/browse/FLINK-27357 > Project: Flink > Issue Type: Improvement >Reporter: Jesus H Christ >Priority: Minor > > Flink on Kubernetes has High Availability services which build an external > service for the rest api access. > [https://sourcegraph.com/github.com/apache/flink@master/-/blob/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/decorators/ExternalServiceDecorator.java] > In the case of multiple job managers in reactive mode or to avoid 15s or so > between job manager restarts in K8S High Availability services, a new job > manager leader can be elected. In this case I think the external service we > create for the web ui, and possibly even the REST API service, no longer > points to the correct job manager as any of the job manager pods can have the > jobmanager label. Is this correct? > If so, it might help to use the endpoint containing the elected leader IP in > the ConfigMap of one specific pod of the JobManager for the external service > web ui, similar to how TaskManagers use JobManager IPs for High Availabiilty. > > It might also help to update the service or endpoint with the new IP to > point to when updating the ConfigMap for JobManager leader election: > [https://sourcegraph.com/github.com/apache/flink/-/blob/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/highavailability/KubernetesLeaderRetrievalDriverFactory.java] -- This message was sent by Atlassian Jira (v8.20.7#820007)
[jira] [Updated] (FLINK-27357) In Flink HA Service on K8S, Web UI External Service should point to elected Job Manager leader's IP
[ https://issues.apache.org/jira/browse/FLINK-27357?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Jesus H Christ updated FLINK-27357: --- Description: Flink on Kubernetes has High Availability services which build an external service for the rest api access. [https://sourcegraph.com/github.com/apache/flink@master/-/blob/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/decorators/ExternalServiceDecorator.java] In the case of multiple job managers in reactive mode or to avoid 15s or so between job manager restarts in K8S High Availability services, a new job manager leader can be elected. In this case I think the external service we create for the web ui, and possibly even the REST API service, no longer points to the correct job manager as any of the job manager pods can have the jobmanager label. It might help to use the endpoint containing the elected leader IP in the ConfigMap of one specific pod of the JobManager for the external service web ui, similar to how TaskManagers use JobManager IPs for High Availabiilty. It might also help to update the service or endpoint with the new IP to point to when updating the ConfigMap for JobManager leader election: [https://sourcegraph.com/github.com/apache/flink/-/blob/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/highavailability/KubernetesLeaderRetrievalDriverFactory.java] was: Flink on Kubernetes has High Availability services which build an external service for the web ui access. [https://sourcegraph.com/github.com/apache/flink@master/-/blob/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/decorators/ExternalServiceDecorator.java] In the case of multiple job managers in reactive mode or to avoid 15s or so between job manager restarts in K8S High Availability services, a new job manager leader can be elected. In this case I think the service no longer points to the correct job manager as any of the job manager pods can have the jobmanager label. It might help to use the endpoint containing the elected leader IP in the ConfigMap of one specific pod of the JobManager for the external service web ui, similar to how TaskManagers use JobManager IPs for High Availabiilty. It might also help to update the service or endpoint with the new IP to point to when updating the ConfigMap for JobManager leader election: [https://sourcegraph.com/github.com/apache/flink/-/blob/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/highavailability/KubernetesLeaderRetrievalDriverFactory.java] > In Flink HA Service on K8S, Web UI External Service should point to elected > Job Manager leader's IP > --- > > Key: FLINK-27357 > URL: https://issues.apache.org/jira/browse/FLINK-27357 > Project: Flink > Issue Type: Improvement >Reporter: Jesus H Christ >Priority: Minor > > Flink on Kubernetes has High Availability services which build an external > service for the rest api access. > [https://sourcegraph.com/github.com/apache/flink@master/-/blob/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/decorators/ExternalServiceDecorator.java] > In the case of multiple job managers in reactive mode or to avoid 15s or so > between job manager restarts in K8S High Availability services, a new job > manager leader can be elected. In this case I think the external service we > create for the web ui, and possibly even the REST API service, no longer > points to the correct job manager as any of the job manager pods can have the > jobmanager label. > It might help to use the endpoint containing the elected leader IP in the > ConfigMap of one specific pod of the JobManager for the external service web > ui, similar to how TaskManagers use JobManager IPs for High Availabiilty. > > It might also help to update the service or endpoint with the new IP to > point to when updating the ConfigMap for JobManager leader election: > [https://sourcegraph.com/github.com/apache/flink/-/blob/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/highavailability/KubernetesLeaderRetrievalDriverFactory.java] -- This message was sent by Atlassian Jira (v8.20.7#820007)
[jira] [Updated] (FLINK-27357) In Flink HA Service on K8S, Web UI External Service should point to elected Job Manager leader's IP
[ https://issues.apache.org/jira/browse/FLINK-27357?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Jesus H Christ updated FLINK-27357: --- Description: Flink on Kubernetes has High Availability services which build an external service for the web ui access. [https://sourcegraph.com/github.com/apache/flink@master/-/blob/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/decorators/ExternalServiceDecorator.java] In the case of multiple job managers in reactive mode or to avoid 15s or so between job manager restarts in K8S High Availability services, a new job manager leader can be elected. In this case I think the service no longer points to the correct job manager as any of the job manager pods can have the jobmanager label. It might help to use the endpoint containing the elected leader IP in the ConfigMap of one specific pod of the JobManager for the external service web ui, similar to how TaskManagers use JobManager IPs for High Availabiilty. It might also help to update the service or endpoint with the new IP to point to when updating the ConfigMap for JobManager leader election: [https://sourcegraph.com/github.com/apache/flink/-/blob/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/highavailability/KubernetesLeaderRetrievalDriverFactory.java] was: Flink on Kubernetes has High Availability services which build an external service for the web ui access. [https://sourcegraph.com/github.com/apache/flink@master/-/blob/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/decorators/ExternalServiceDecorator.java] In the case of multiple job managers in reactive mode or to avoid 15s or so between job manager restarts in K8S High Availability services, a new job manager leader can be elected. In this case I think the service no longer points to the correct job manager as any of the job manager pods can have the jobmanager label. It might help to use the endpoint of one specific pod of the JobManager for the external service web ui, similar to how TaskManagers use JobManager IPs for High Availabiilty. It might also help to update the service with the new IP or endpoint somehow for the external service to point to when updating the ConfigMap for JobManager leader election: [https://sourcegraph.com/github.com/apache/flink/-/blob/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/highavailability/KubernetesLeaderRetrievalDriverFactory.java] > In Flink HA Service on K8S, Web UI External Service should point to elected > Job Manager leader's IP > --- > > Key: FLINK-27357 > URL: https://issues.apache.org/jira/browse/FLINK-27357 > Project: Flink > Issue Type: Improvement >Reporter: Jesus H Christ >Priority: Minor > > Flink on Kubernetes has High Availability services which build an external > service for the web ui access. > [https://sourcegraph.com/github.com/apache/flink@master/-/blob/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/decorators/ExternalServiceDecorator.java] > In the case of multiple job managers in reactive mode or to avoid 15s or so > between job manager restarts in K8S High Availability services, a new job > manager leader can be elected. In this case I think the service no longer > points to the correct job manager as any of the job manager pods can have the > jobmanager label. > It might help to use the endpoint containing the elected leader IP in the > ConfigMap of one specific pod of the JobManager for the external service web > ui, similar to how TaskManagers use JobManager IPs for High Availabiilty. > > It might also help to update the service or endpoint with the new IP to > point to when updating the ConfigMap for JobManager leader election: > [https://sourcegraph.com/github.com/apache/flink/-/blob/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/highavailability/KubernetesLeaderRetrievalDriverFactory.java] -- This message was sent by Atlassian Jira (v8.20.7#820007)
[jira] [Updated] (FLINK-27357) In Flink HA Service on K8S, Web UI External Service should point to elected Job Manager leader's IP
[ https://issues.apache.org/jira/browse/FLINK-27357?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Jesus H Christ updated FLINK-27357: --- Description: Flink on Kubernetes has High Availability services which build an external service for the web ui access. [https://sourcegraph.com/github.com/apache/flink@master/-/blob/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/decorators/ExternalServiceDecorator.java] In the case of multiple job managers in reactive mode or to avoid 15s or so between job manager restarts in K8S High Availability services, a new job manager leader can be elected. In this case I think the service no longer points to the correct job manager as any of the job manager pods can have the jobmanager label. It might help to use the endpoint of one specific pod of the JobManager for the external service web ui, similar to how TaskManagers use JobManager IPs for High Availabiilty. It might also help to update the service with the new IP or endpoint somehow for the external service to point to when updating the ConfigMap for JobManager leader election: [https://sourcegraph.com/github.com/apache/flink/-/blob/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/highavailability/KubernetesLeaderRetrievalDriverFactory.java] was: Flink on Kubernetes has High Availability services which build an external service for the web ui access. [https://sourcegraph.com/github.com/apache/flink@master/-/blob/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/decorators/ExternalServiceDecorator.java] In the case of multiple job managers in reactive mode or to avoid 15s or so between job manager restarts in K8S High Availability services, a new job manager leader can be elected. In this case I think the service no longer points to the correct job manager as any of the job manager pods can have the jobmanager label. It might help to use the endpoint of the JobManager for the external service web ui, similar to how Task managers do for High Availabiilty. It might also help to update the service with the new IP or endpoint somehow for the external service to point to when updating the ConfigMap for JobManager leader election: [https://sourcegraph.com/github.com/apache/flink/-/blob/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/highavailability/KubernetesLeaderRetrievalDriverFactory.java] > In Flink HA Service on K8S, Web UI External Service should point to elected > Job Manager leader's IP > --- > > Key: FLINK-27357 > URL: https://issues.apache.org/jira/browse/FLINK-27357 > Project: Flink > Issue Type: Improvement >Reporter: Jesus H Christ >Priority: Minor > > Flink on Kubernetes has High Availability services which build an external > service for the web ui access. > [https://sourcegraph.com/github.com/apache/flink@master/-/blob/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/decorators/ExternalServiceDecorator.java] > In the case of multiple job managers in reactive mode or to avoid 15s or so > between job manager restarts in K8S High Availability services, a new job > manager leader can be elected. In this case I think the service no longer > points to the correct job manager as any of the job manager pods can have the > jobmanager label. > It might help to use the endpoint of one specific pod of the JobManager for > the external service web ui, similar to how TaskManagers use JobManager IPs > for High Availabiilty. > > It might also help to update the service with the new IP or endpoint somehow > for the external service to point to when updating the ConfigMap for > JobManager leader election: > [https://sourcegraph.com/github.com/apache/flink/-/blob/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/highavailability/KubernetesLeaderRetrievalDriverFactory.java] -- This message was sent by Atlassian Jira (v8.20.7#820007)
[jira] [Updated] (FLINK-27357) In Flink HA Service on K8S, Web UI External Service should point to elected Job Manager leader's IP
[ https://issues.apache.org/jira/browse/FLINK-27357?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Jesus H Christ updated FLINK-27357: --- Description: Flink on Kubernetes has High Availability services which build an external service for the web ui access. [https://sourcegraph.com/github.com/apache/flink@master/-/blob/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/decorators/ExternalServiceDecorator.java] In the case of multiple job managers in reactive mode or to avoid 15s or so between job manager restarts in K8S High Availability services, a new job manager leader can be elected. In this case I think the service no longer points to the correct job manager as any of the job manager pods can have the jobmanager label. It might help to use the endpoint of the JobManager for the external service web ui, similar to how Task managers do for High Availabiilty. It might also help to update the service with the new IP or endpoint somehow for the external service to point to when updating the ConfigMap for JobManager leader election: [https://sourcegraph.com/github.com/apache/flink/-/blob/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/highavailability/KubernetesLeaderRetrievalDriverFactory.java] was: Flink on Kubernetes has High Availability services which build an external service for the web ui access. [https://sourcegraph.com/github.com/apache/flink@master/-/blob/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/decorators/ExternalServiceDecorator.java] In the case of multiple job managers to avoid 15s or so between job manager restarts, a new job manager leader can be elected. In this case I think the service no longer points to the correct job manager as any of the job manager pods can have the jobmanager label. It might help to use the endpoint of the JobManager for the external service web ui, similar to how Task managers do for High Availabiilty. It might also help to update the service with the new IP or endpoint somehow for the external service to point to when updating the ConfigMap for JobManager leader election: [https://sourcegraph.com/github.com/apache/flink/-/blob/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/highavailability/KubernetesLeaderRetrievalDriverFactory.java] > In Flink HA Service on K8S, Web UI External Service should point to elected > Job Manager leader's IP > --- > > Key: FLINK-27357 > URL: https://issues.apache.org/jira/browse/FLINK-27357 > Project: Flink > Issue Type: Improvement >Reporter: Jesus H Christ >Priority: Minor > > Flink on Kubernetes has High Availability services which build an external > service for the web ui access. > [https://sourcegraph.com/github.com/apache/flink@master/-/blob/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/decorators/ExternalServiceDecorator.java] > In the case of multiple job managers in reactive mode or to avoid 15s or so > between job manager restarts in K8S High Availability services, a new job > manager leader can be elected. In this case I think the service no longer > points to the correct job manager as any of the job manager pods can have the > jobmanager label. > It might help to use the endpoint of the JobManager for the external service > web ui, similar to how Task managers do for High Availabiilty. > > It might also help to update the service with the new IP or endpoint somehow > for the external service to point to when updating the ConfigMap for > JobManager leader election: > [https://sourcegraph.com/github.com/apache/flink/-/blob/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/highavailability/KubernetesLeaderRetrievalDriverFactory.java] -- This message was sent by Atlassian Jira (v8.20.7#820007)
[jira] [Updated] (FLINK-27357) In Flink HA Service on K8S, Web UI External Service should update with new elected Job Manager leader's IP
[ https://issues.apache.org/jira/browse/FLINK-27357?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Jesus H Christ updated FLINK-27357: --- Description: Flink on Kubernetes has High Availability services which build an external service for the web ui access. [https://sourcegraph.com/github.com/apache/flink@master/-/blob/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/decorators/ExternalServiceDecorator.java] In the case of multiple job managers to avoid 15s or so between job manager restarts, a new job manager leader can be elected. In this case I think the service no longer points to the correct job manager as any of the job manager pods can have the jobmanager label. It might help to use the endpoint of the JobManager for the external service web ui, similar to how Task managers do for High Availabiilty. It might also help to update the service with the new IP or endpoint somehow for the external service to point to when updating the ConfigMap for JobManager leader election: [https://sourcegraph.com/github.com/apache/flink/-/blob/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/highavailability/KubernetesLeaderRetrievalDriverFactory.java] was: Flink on Kubernetes has High Availability services which build an external service for the web ui access. [https://sourcegraph.com/github.com/apache/flink@master/-/blob/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/decorators/ExternalServiceDecorator.java] In the case of multiple job managers to avoid 15s or so between job manager restarts, a new job manager leader can be elected. In this case I think the service no longer points to the correct job manager as any of the job manager pods can have the jobmanager label. It might help to use the endpoint of the JobManager for the external service web ui, similar to how Task managers do for High Availabiilty. It might also help to update the service with the new IP or endpoint somehow for the external service to point to when updating the ConfigMap for JobManager leader election: [https://sourcegraph.com/github.com/apache/flink/-/blob/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/highavailability/KubernetesLeaderRetrievalDriverFactory.java] > In Flink HA Service on K8S, Web UI External Service should update with new > elected Job Manager leader's IP > -- > > Key: FLINK-27357 > URL: https://issues.apache.org/jira/browse/FLINK-27357 > Project: Flink > Issue Type: Improvement >Reporter: Jesus H Christ >Priority: Minor > > Flink on Kubernetes has High Availability services which build an external > service for the web ui access. > [https://sourcegraph.com/github.com/apache/flink@master/-/blob/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/decorators/ExternalServiceDecorator.java] > In the case of multiple job managers to avoid 15s or so between job manager > restarts, a new job manager leader can be elected. In this case I think the > service no longer points to the correct job manager as any of the job manager > pods can have the jobmanager label. > It might help to use the endpoint of the JobManager for the external service > web ui, similar to how Task managers do for High Availabiilty. > > It might also help to update the service with the new IP or endpoint somehow > for the external service to point to when updating the ConfigMap for > JobManager leader election: > [https://sourcegraph.com/github.com/apache/flink/-/blob/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/highavailability/KubernetesLeaderRetrievalDriverFactory.java] -- This message was sent by Atlassian Jira (v8.20.7#820007)
[jira] [Updated] (FLINK-27357) In Flink HA Service on K8S, Web UI External Service should point to elected Job Manager leader's IP
[ https://issues.apache.org/jira/browse/FLINK-27357?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Jesus H Christ updated FLINK-27357: --- Summary: In Flink HA Service on K8S, Web UI External Service should point to elected Job Manager leader's IP (was: In Flink HA Service on K8S, Web UI External Service should update with new elected Job Manager leader's IP) > In Flink HA Service on K8S, Web UI External Service should point to elected > Job Manager leader's IP > --- > > Key: FLINK-27357 > URL: https://issues.apache.org/jira/browse/FLINK-27357 > Project: Flink > Issue Type: Improvement >Reporter: Jesus H Christ >Priority: Minor > > Flink on Kubernetes has High Availability services which build an external > service for the web ui access. > [https://sourcegraph.com/github.com/apache/flink@master/-/blob/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/decorators/ExternalServiceDecorator.java] > In the case of multiple job managers to avoid 15s or so between job manager > restarts, a new job manager leader can be elected. In this case I think the > service no longer points to the correct job manager as any of the job manager > pods can have the jobmanager label. > It might help to use the endpoint of the JobManager for the external service > web ui, similar to how Task managers do for High Availabiilty. > > It might also help to update the service with the new IP or endpoint somehow > for the external service to point to when updating the ConfigMap for > JobManager leader election: > [https://sourcegraph.com/github.com/apache/flink/-/blob/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/highavailability/KubernetesLeaderRetrievalDriverFactory.java] -- This message was sent by Atlassian Jira (v8.20.7#820007)
[jira] [Updated] (FLINK-27357) In Flink HA Service on K8S, Web UI External Service should update with new elected Job Manager leader's IP
[ https://issues.apache.org/jira/browse/FLINK-27357?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Jesus H Christ updated FLINK-27357: --- Description: Flink on Kubernetes has High Availability services which build an external service for the web ui access. [https://sourcegraph.com/github.com/apache/flink@master/-/blob/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/decorators/ExternalServiceDecorator.java] In the case of multiple job managers to avoid 15s or so between job manager restarts, a new job manager leader can be elected. In this case I think the service no longer points to the correct job manager as any of the job manager pods can have the jobmanager label. It might help to use the endpoint of the JobManager for the external service web ui, similar to how Task managers do for High Availabiilty. It might also help to update the service with the new IP or endpoint somehow for the external service to point to when updating the ConfigMap for JobManager leader election: [https://sourcegraph.com/github.com/apache/flink/-/blob/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/highavailability/KubernetesLeaderRetrievalDriverFactory.java] was: Flink on Kubernetes has High Availability services which build an external service for the web ui access. [https://sourcegraph.com/github.com/apache/flink@master/-/blob/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/decorators/ExternalServiceDecorator.java] In the case of multiple job managers to avoid 15s or so between job manager restarts, a new job manager leader can be elected. In this case I think the service no longer points to the correct job manager as any of the job manager pods can have the jobmanager label. It might help to use IPs for JobManagers like Task managers do for High Availabiilty. It might also help to update the service with the new IP or endpoint somehow for the external service to point to when updating the ConfigMap for JobManager leader election: https://sourcegraph.com/github.com/apache/flink/-/blob/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/highavailability/KubernetesLeaderRetrievalDriverFactory.java > In Flink HA Service on K8S, Web UI External Service should update with new > elected Job Manager leader's IP > -- > > Key: FLINK-27357 > URL: https://issues.apache.org/jira/browse/FLINK-27357 > Project: Flink > Issue Type: Improvement >Reporter: Jesus H Christ >Priority: Minor > > Flink on Kubernetes has High Availability services which build an external > service for the web ui access. > [https://sourcegraph.com/github.com/apache/flink@master/-/blob/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/decorators/ExternalServiceDecorator.java] > In the case of multiple job managers to avoid 15s or so between job manager > restarts, a new job manager leader can be elected. In this case I think the > service no longer points to the correct job manager as any of the job manager > pods can have the jobmanager label. > It might help to use the endpoint of the JobManager for the external service > web ui, similar to how Task managers do for High Availabiilty. It might also > help to update the service with the new IP or endpoint somehow for the > external service to point to when updating the ConfigMap for JobManager > leader election: > [https://sourcegraph.com/github.com/apache/flink/-/blob/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/highavailability/KubernetesLeaderRetrievalDriverFactory.java] -- This message was sent by Atlassian Jira (v8.20.7#820007)
[jira] [Updated] (FLINK-27357) In Flink HA Service on K8S, Web UI External Service should update with new elected Job Manager leader's IP
[ https://issues.apache.org/jira/browse/FLINK-27357?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Jesus H Christ updated FLINK-27357: --- Description: Flink on Kubernetes has High Availability services which build an external service for the web ui access. [https://sourcegraph.com/github.com/apache/flink@master/-/blob/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/decorators/ExternalServiceDecorator.java] In the case of multiple job managers to avoid 15s or so between job manager restarts, a new job manager leader can be elected. In this case I think the service no longer points to the correct job manager as any of the job manager pods can have the jobmanager label. It might help to use IPs for JobManagers like Task managers do for High Availabiilty. It might also help to update the service with the new IP or endpoint somehow for the external service to point to when updating the ConfigMap for JobManager leader election: https://sourcegraph.com/github.com/apache/flink/-/blob/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/highavailability/KubernetesLeaderRetrievalDriverFactory.java was: Flink on Kubernetes has High Availability services which build an external service for the web ui access. [https://sourcegraph.com/github.com/apache/flink@master/-/blob/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/decorators/ExternalServiceDecorator.java] In the case of multiple job managers to avoid 15s or so between job manager restarts, a new job manager leader can be elected. In this case I think the service no longer points to the correct job manager as any of the job manager pods can have the jobmanager label. It might help to use IPs for JobManagers like Task managers do for HIgh Availabiilty. It might also help to update the service with the new IP or endpoint somehow for the external service to point to when updating the ConfigMap for JobManager leader election. > In Flink HA Service on K8S, Web UI External Service should update with new > elected Job Manager leader's IP > -- > > Key: FLINK-27357 > URL: https://issues.apache.org/jira/browse/FLINK-27357 > Project: Flink > Issue Type: Improvement >Reporter: Jesus H Christ >Priority: Minor > > Flink on Kubernetes has High Availability services which build an external > service for the web ui access. > [https://sourcegraph.com/github.com/apache/flink@master/-/blob/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/decorators/ExternalServiceDecorator.java] > In the case of multiple job managers to avoid 15s or so between job manager > restarts, a new job manager leader can be elected. In this case I think the > service no longer points to the correct job manager as any of the job manager > pods can have the jobmanager label. > It might help to use IPs for JobManagers like Task managers do for High > Availabiilty. It might also help to update the service with the new IP or > endpoint somehow for the external service to point to when updating the > ConfigMap for JobManager leader election: > https://sourcegraph.com/github.com/apache/flink/-/blob/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/highavailability/KubernetesLeaderRetrievalDriverFactory.java -- This message was sent by Atlassian Jira (v8.20.7#820007)
[jira] [Updated] (FLINK-27357) In Flink HA Service on K8S, Web UI External Service should update with new elected Job Manager leader's IP
[ https://issues.apache.org/jira/browse/FLINK-27357?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Jesus H Christ updated FLINK-27357: --- Description: Flink on Kubernetes has High Availability services which build an external service for the web ui access. [https://sourcegraph.com/github.com/apache/flink@master/-/blob/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/decorators/ExternalServiceDecorator.java] In the case of multiple job managers to avoid 15s or so between job manager restarts, a new job manager leader can be elected. In this case I think the service no longer points to the correct job manager as any of the job manager pods can have the jobmanager label. It might help to use IPs for JobManagers like Task managers do for HIgh Availabiilty. It might also help to update the service with the new IP or endpoint somehow for the external service to point to when updating the ConfigMap for JobManager leader election. was: Flink on Kubernetes has High Availability services which build an external service for the web ui access. [https://sourcegraph.com/github.com/apache/flink@master/-/blob/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/decorators/ExternalServiceDecorator.java] In the case of multiple job managers to avoid 15s or so between job manager restarts, a new job manager leader can be elected. In this case I think the service no longer points to the correct job manager as any of the job manager pods can have the jobmanager label. It might help to use IPs for JobManagers like Task managers do for HIgh Availabiilty. It might also help to update the service with the new IP or endpoint somehow for the external service to point to. > In Flink HA Service on K8S, Web UI External Service should update with new > elected Job Manager leader's IP > -- > > Key: FLINK-27357 > URL: https://issues.apache.org/jira/browse/FLINK-27357 > Project: Flink > Issue Type: Improvement >Reporter: Jesus H Christ >Priority: Minor > > Flink on Kubernetes has High Availability services which build an external > service for the web ui access. > [https://sourcegraph.com/github.com/apache/flink@master/-/blob/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/decorators/ExternalServiceDecorator.java] > In the case of multiple job managers to avoid 15s or so between job manager > restarts, a new job manager leader can be elected. In this case I think the > service no longer points to the correct job manager as any of the job manager > pods can have the jobmanager label. > It might help to use IPs for JobManagers like Task managers do for HIgh > Availabiilty. It might also help to update the service with the new IP or > endpoint somehow for the external service to point to when updating the > ConfigMap for JobManager leader election. -- This message was sent by Atlassian Jira (v8.20.7#820007)
[jira] [Updated] (FLINK-27357) In Flink HA Service on K8S, Web UI External Service should update with new elected Job Manager leader's IP
[ https://issues.apache.org/jira/browse/FLINK-27357?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Jesus H Christ updated FLINK-27357: --- Summary: In Flink HA Service on K8S, Web UI External Service should update with new elected Job Manager leader's IP (was: In Flink HA Service, Web UI External Service should update with new elected Job Manager leader's IP) > In Flink HA Service on K8S, Web UI External Service should update with new > elected Job Manager leader's IP > -- > > Key: FLINK-27357 > URL: https://issues.apache.org/jira/browse/FLINK-27357 > Project: Flink > Issue Type: Improvement >Reporter: Jesus H Christ >Priority: Minor > > Flink on Kubernetes has High Availability services which build an external > service for the web ui access. > [https://sourcegraph.com/github.com/apache/flink@master/-/blob/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/decorators/ExternalServiceDecorator.java] > In the case of multiple job managers to avoid 15s or so between job manager > restarts, a new job manager leader can be elected. In this case I think the > service no longer points to the correct job manager as any of the job manager > pods can have the jobmanager label. > It might help to use IPs for JobManagers like Task managers do for HIgh > Availabiilty. It might also help to update the service with the new IP or > endpoint somehow for the external service to point to. -- This message was sent by Atlassian Jira (v8.20.7#820007)
[jira] [Updated] (FLINK-27357) In Flink HA Service, Web UI External Service should update with new elected Job Manager leader's IP
[ https://issues.apache.org/jira/browse/FLINK-27357?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Jesus H Christ updated FLINK-27357: --- Description: Flink on Kubernetes has High Availability services which build an external service for the web ui access. [https://sourcegraph.com/github.com/apache/flink@master/-/blob/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/decorators/ExternalServiceDecorator.java] In the case of multiple job managers to avoid 15s or so between job manager restarts, a new job manager leader can be elected. In this case I think the service no longer points to the correct job manager as any of the job manager pods can have the jobmanager label. It might help to use IPs for JobManagers like Task managers do for HIgh Availabiilty. It might also help to update the service with the new IP or endpoint somehow for the external service to point to. was: Flink on Kubernetes has High Availability services which build an external service for the web ui access. [https://sourcegraph.com/github.com/apache/flink@master/-/blob/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/decorators/ExternalServiceDecorator.java] In the case of multiple job managers to avoid 15s or so between job manager restarts, a new job manager leader can be elected. In this case I think the service no longer points to the correct job manager as any of the job manager pods can have the jobmanager label > In Flink HA Service, Web UI External Service should update with new elected > Job Manager leader's IP > --- > > Key: FLINK-27357 > URL: https://issues.apache.org/jira/browse/FLINK-27357 > Project: Flink > Issue Type: Improvement >Reporter: Jesus H Christ >Priority: Minor > > Flink on Kubernetes has High Availability services which build an external > service for the web ui access. > [https://sourcegraph.com/github.com/apache/flink@master/-/blob/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/decorators/ExternalServiceDecorator.java] > In the case of multiple job managers to avoid 15s or so between job manager > restarts, a new job manager leader can be elected. In this case I think the > service no longer points to the correct job manager as any of the job manager > pods can have the jobmanager label. > It might help to use IPs for JobManagers like Task managers do for HIgh > Availabiilty. It might also help to update the service with the new IP or > endpoint somehow for the external service to point to. -- This message was sent by Atlassian Jira (v8.20.7#820007)
[jira] [Updated] (FLINK-27357) In Flink HA Service, Web UI External Service should update with new elected Job Manager leader's IP
[ https://issues.apache.org/jira/browse/FLINK-27357?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Jesus H Christ updated FLINK-27357: --- Description: Flink on Kubernetes has High Availability services which build an external service for the web ui access. [https://sourcegraph.com/github.com/apache/flink@master/-/blob/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/decorators/ExternalServiceDecorator.java] In the case of multiple job managers to avoid 15s or so between job manager restarts, a new job manager leader can be elected. In this case I think the service no longer points to the correct job manager as any of the job manager pods can have the jobmanager label was: Flink on Kubernetes has High Availability services which build an external service for the web ui access. [https://sourcegraph.com/github.com/apache/flink@master/-/blob/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/decorators/ExternalServiceDecorator.java] In the case of multiple job managers to avoid 15 or so between job manager restarts, a new job manager leader can be elected. In this case I think the service no longer points to the correct job manager as any of the job manager pods can have the jobmanager label Priority: Minor (was: Major) > In Flink HA Service, Web UI External Service should update with new elected > Job Manager leader's IP > --- > > Key: FLINK-27357 > URL: https://issues.apache.org/jira/browse/FLINK-27357 > Project: Flink > Issue Type: Improvement >Reporter: Jesus H Christ >Priority: Minor > > Flink on Kubernetes has High Availability services which build an external > service for the web ui access. > [https://sourcegraph.com/github.com/apache/flink@master/-/blob/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/decorators/ExternalServiceDecorator.java] > In the case of multiple job managers to avoid 15s or so between job manager > restarts, a new job manager leader can be elected. In this case I think the > service no longer points to the correct job manager as any of the job manager > pods can have the jobmanager label -- This message was sent by Atlassian Jira (v8.20.7#820007)
[jira] [Created] (FLINK-27357) In Flink HA Service, Web UI External Service should update with new elected Job Manager leader's IP
Jesus H Christ created FLINK-27357: -- Summary: In Flink HA Service, Web UI External Service should update with new elected Job Manager leader's IP Key: FLINK-27357 URL: https://issues.apache.org/jira/browse/FLINK-27357 Project: Flink Issue Type: Improvement Reporter: Jesus H Christ Flink on Kubernetes has High Availability services which build an external service for the web ui access. [https://sourcegraph.com/github.com/apache/flink@master/-/blob/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/decorators/ExternalServiceDecorator.java] In the case of multiple job managers to avoid 15 or so between job manager restarts, a new job manager leader can be elected. In this case I think the service no longer points to the correct job manager as any of the job manager pods can have the jobmanager label -- This message was sent by Atlassian Jira (v8.20.7#820007)
[jira] [Updated] (FLINK-26177) PulsarSourceITCase.testScaleDown fails with timeout
[
https://issues.apache.org/jira/browse/FLINK-26177?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Flink Jira Bot updated FLINK-26177:
---
Labels: auto-deprioritized-critical pull-request-available stale-blocker
test-stability (was: pull-request-available stale-blocker stale-critical
test-stability)
Priority: Major (was: Critical)
This issue was labeled "stale-critical" 7 days ago and has not received any
updates so it is being deprioritized. If this ticket is actually Critical,
please raise the priority and ask a committer to assign you the issue or revive
the public discussion.
> PulsarSourceITCase.testScaleDown fails with timeout
> ---
>
> Key: FLINK-26177
> URL: https://issues.apache.org/jira/browse/FLINK-26177
> Project: Flink
> Issue Type: Bug
> Components: Connectors / Pulsar
>Affects Versions: 1.15.0, 1.16.0
>Reporter: Matthias Pohl
>Priority: Major
> Labels: auto-deprioritized-critical, pull-request-available,
> stale-blocker, test-stability
>
> We observed a [build
> failure|https://dev.azure.com/mapohl/flink/_build/results?buildId=742=logs=f3dc9b18-b77a-55c1-591e-264c46fe44d1=2d3cd81e-1c37-5c31-0ee4-f5d5cdb9324d=26553]
> caused by {{PulsarSourceITCase.testScaleDown}}:
> {code}
> Feb 15 20:56:02 [ERROR] Tests run: 16, Failures: 1, Errors: 0, Skipped: 0,
> Time elapsed: 431.023 s <<< FAILURE! - in
> org.apache.flink.connector.pulsar.source.PulsarSourceITCase
> Feb 15 20:56:02 [ERROR]
> org.apache.flink.connector.pulsar.source.PulsarSourceITCase.testScaleDown(TestEnvironment,
> DataStreamSourceExternalContext, CheckpointingMode)[2] Time elapsed:
> 138.444 s <<< FAILURE!
> Feb 15 20:56:02 java.lang.AssertionError:
> Feb 15 20:56:02
> Feb 15 20:56:02 Expecting
> Feb 15 20:56:02
> Feb 15 20:56:02 to be completed within 2M.
> Feb 15 20:56:02
> Feb 15 20:56:02 exception caught while trying to get the future result:
> java.util.concurrent.TimeoutException
> Feb 15 20:56:02 at
> java.util.concurrent.CompletableFuture.timedGet(CompletableFuture.java:1784)
> [...]
> {code}
--
This message was sent by Atlassian Jira
(v8.20.7#820007)
[jira] [Updated] (FLINK-26238) PulsarSinkITCase.writeRecordsToPulsar failed on azure
[
https://issues.apache.org/jira/browse/FLINK-26238?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Flink Jira Bot updated FLINK-26238:
---
Labels: stale-major test-stability (was: test-stability)
I am the [Flink Jira Bot|https://github.com/apache/flink-jira-bot/] and I help
the community manage its development. I see this issues has been marked as
Major but is unassigned and neither itself nor its Sub-Tasks have been updated
for 60 days. I have gone ahead and added a "stale-major" to the issue". If this
ticket is a Major, please either assign yourself or give an update. Afterwards,
please remove the label or in 7 days the issue will be deprioritized.
> PulsarSinkITCase.writeRecordsToPulsar failed on azure
> -
>
> Key: FLINK-26238
> URL: https://issues.apache.org/jira/browse/FLINK-26238
> Project: Flink
> Issue Type: Bug
> Components: Connectors / Pulsar
>Affects Versions: 1.15.0
>Reporter: Yun Gao
>Priority: Major
> Labels: stale-major, test-stability
>
> {code:java}
> Feb 17 12:19:44 [ERROR] Tests run: 3, Failures: 1, Errors: 0, Skipped: 0,
> Time elapsed: 342.177 s <<< FAILURE! - in
> org.apache.flink.connector.pulsar.sink.PulsarSinkITCase
> Feb 17 12:19:44 [ERROR]
> org.apache.flink.connector.pulsar.sink.PulsarSinkITCase.writeRecordsToPulsar(DeliveryGuarantee)[3]
> Time elapsed: 302.4 s <<< FAILURE!
> Feb 17 12:19:44 java.lang.AssertionError:
> Feb 17 12:19:44
> Feb 17 12:19:44 Actual and expected should have same size but actual size is:
> Feb 17 12:19:44 0
> Feb 17 12:19:44 while expected size is:
> Feb 17 12:19:44 179
> Feb 17 12:19:44 Actual was:
> Feb 17 12:19:44 []
> Feb 17 12:19:44 Expected was:
> Feb 17 12:19:44 ["NONE-CzYEIFDd-0-eELBphcHiu",
> Feb 17 12:19:44 "NONE-CzYEIFDd-1-odr3NpH6pg",
> Feb 17 12:19:44 "NONE-CzYEIFDd-2-HfIphNFXoM",
> Feb 17 12:19:44 "NONE-CzYEIFDd-3-iaZ9v2HCnw",
> Feb 17 12:19:44 "NONE-CzYEIFDd-4-6KkXK34GZl",
> Feb 17 12:19:44 "NONE-CzYEIFDd-5-jK9UxXSQcX",
> Feb 17 12:19:44 "NONE-CzYEIFDd-6-HipVPVNqZA",
> Feb 17 12:19:44 "NONE-CzYEIFDd-7-lT4lVH3CzX",
> Feb 17 12:19:44 "NONE-CzYEIFDd-8-4jShEBuQaS",
> Feb 17 12:19:44 "NONE-CzYEIFDd-9-fInSd97msu",
> Feb 17 12:19:44 "NONE-CzYEIFDd-10-dGBm5e92os",
> Feb 17 12:19:44 "NONE-CzYEIFDd-11-GkINb6Dipx",
> Feb 17 12:19:44 "NONE-CzYEIFDd-12-M7Q8atHhNQ",
> Feb 17 12:19:44 "NONE-CzYEIFDd-13-EG2FpyziCL",
> Feb 17 12:19:44 "NONE-CzYEIFDd-14-4HwGJSOkTk",
> Feb 17 12:19:44 "NONE-CzYEIFDd-15-UC0IwwKN0O",
> Feb 17 12:19:44 "NONE-CzYEIFDd-16-D9FOV8hKBq",
> Feb 17 12:19:44 "NONE-CzYEIFDd-17-J2Zb6pNmOO",
> Feb 17 12:19:44 "NONE-CzYEIFDd-18-abo3YgkYKP",
> Feb 17 12:19:44 "NONE-CzYEIFDd-19-4Q5GbBRSc6",
> Feb 17 12:19:44 "NONE-CzYEIFDd-20-WxSP9oExJP",
> Feb 17 12:19:44 "NONE-CzYEIFDd-21-0wiqq21CY1",
> Feb 17 12:19:44 "NONE-CzYEIFDd-22-3iJQiFjgQu",
> Feb 17 12:19:44 "NONE-CzYEIFDd-23-78je74YwU6",
> Feb 17 12:19:44 "NONE-CzYEIFDd-24-tEkEaF9IuD",
> Feb 17 12:19:44 "NONE-CzYEIFDd-25-vDi5h44tjJ",
> Feb 17 12:19:44 "NONE-CzYEIFDd-26-GzIh4FLlvP",
> {code}
> https://dev.azure.com/apache-flink/apache-flink/_build/results?buildId=31739=logs=fc5181b0-e452-5c8f-68de-1097947f6483=995c650b-6573-581c-9ce6-7ad4cc038461=27095
--
This message was sent by Atlassian Jira
(v8.20.7#820007)
[jira] [Updated] (FLINK-26595) Improve the PostgresDialect method for getting upsert statements.
[
https://issues.apache.org/jira/browse/FLINK-26595?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Flink Jira Bot updated FLINK-26595:
---
Labels: pull-request-available stale-assigned (was: pull-request-available)
I am the [Flink Jira Bot|https://github.com/apache/flink-jira-bot/] and I help
the community manage its development. I see this issue is assigned but has not
received an update in 30 days, so it has been labeled "stale-assigned".
If you are still working on the issue, please remove the label and add a
comment updating the community on your progress. If this issue is waiting on
feedback, please consider this a reminder to the committer/reviewer. Flink is a
very active project, and so we appreciate your patience.
If you are no longer working on the issue, please unassign yourself so someone
else may work on it.
> Improve the PostgresDialect method for getting upsert statements.
> -
>
> Key: FLINK-26595
> URL: https://issues.apache.org/jira/browse/FLINK-26595
> Project: Flink
> Issue Type: Bug
> Components: Connectors / JDBC
>Affects Versions: 1.13.1
>Reporter: wuguihu
>Assignee: wuguihu
>Priority: Major
> Labels: pull-request-available, stale-assigned
> Attachments: image-20220311125613545.png,
> image-20220311130744606.png, image-20220311141815540.png,
> image-20220315001550269.png
>
>
> I'm trying to use Flink CDC to synchronize mysql data to matrixDB in real
> time.
> But I encountered an error.
> The error message is as follows:
> {quote}CIRCULAR REFERENCE:java.io.IOException: java.sql.BatchUpdateException:
> Batch entry 0 INSERT INTO user_1(id, name, address, phone_number, email)
> VALUES ('110'::numeric, 'user_110', 'Shanghai', '123567891234',
> 'user_...@foo.com') ON CONFLICT (id) DO UPDATE SET id=EXCLUDED.id,
> name=EXCLUDED.name, address=EXCLUDED.address,
> phone_number=EXCLUDED.phone_number, email=EXCLUDED.email was aborted: ERROR:
> modification of distribution columns in OnConflictUpdate is not supported
> Call getNextException to see other errors in the batch.
> {quote}
> This exception is caused by the getUpsertStatement method of PostgresDialect.
> There is something wrong with the upsert statement.
> In the Update statement, uniqueKey-related columns should be deleted;
>
> I did the following experiment to test my modifications.
> At the same time, I recompiled and packaged flink-connector-JDBC. Using the
> modified flink-connector-JDBC, my program no longer reported errors.
> {code:sql}
> -- 1、Create a table for maxtrixDB
> CREATE TABLE user_1 (
> id int,
> name VARCHAR(255) NOT NULL DEFAULT 'flink',
> address VARCHAR(1024),
> phone_number VARCHAR(512),
> email VARCHAR(255),
> UNIQUE(id)
> );
> -- 2、Insert a record.
> INSERT INTO user_1(id, name, address, phone_number, email)
> VALUES ('110'::numeric, 'user_110', 'Shanghai', '123567891234',
> 'user_...@foo.com')
> ON CONFLICT (id)
> DO UPDATE SET
> id=EXCLUDED.id,
> name=EXCLUDED.name,
> address=EXCLUDED.address,
> phone_number=EXCLUDED.phone_number,
> email=EXCLUDED.email;
> -- 3、Executing the above insert statement results in the following error.
> ERROR: modification of distribution columns in OnConflictUpdate is not
> supported
> -- 4、If the value is changed to the following statement, the command is
> executed successfully.
> INSERT INTO user_1(id, name, address, phone_number, email)
> VALUES ('110'::numeric, 'user_110', 'Shanghai', '123567891234',
> 'user_...@foo.com')
> ON CONFLICT (id)
> DO UPDATE SET
> name=EXCLUDED.name,
> address=EXCLUDED.address,
> phone_number=EXCLUDED.phone_number,
> email=EXCLUDED.email;
> {code}
>
>
> The PostgresDialect class handles upsert statements as follows：
> {code:java}
> // package org.apache.flink.connector.jdbc.dialect.psql
> public Optional getUpsertStatement(
> String tableName, String[] fieldNames, String[] uniqueKeyFields) {
> String uniqueColumns =
> Arrays.stream(uniqueKeyFields)
> .map(this::quoteIdentifier)
> .collect(Collectors.joining(", "));
> String updateClause =
> Arrays.stream(fieldNames)
> .map(f -> quoteIdentifier(f) + "=EXCLUDED." +
> quoteIdentifier(f))
> .collect(Collectors.joining(", "));
> return Optional.of(
> getInsertIntoStatement(tableName, fieldNames)
> + " ON CONFLICT ("
> + uniqueColumns
> + ")"
> + " DO UPDATE SET "
> + updateClause);
> }
> {code}
>
>
> To fix this problem, make the following changes to PostgresDialect：
> {code:java}
> //
[jira] [Updated] (FLINK-25451) KafkaEnumeratorTest.testDiscoverPartitionsPeriodically failed on azure
[
https://issues.apache.org/jira/browse/FLINK-25451?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Flink Jira Bot updated FLINK-25451:
---
Labels: stale-major test-stability (was: test-stability)
I am the [Flink Jira Bot|https://github.com/apache/flink-jira-bot/] and I help
the community manage its development. I see this issues has been marked as
Major but is unassigned and neither itself nor its Sub-Tasks have been updated
for 60 days. I have gone ahead and added a "stale-major" to the issue". If this
ticket is a Major, please either assign yourself or give an update. Afterwards,
please remove the label or in 7 days the issue will be deprioritized.
> KafkaEnumeratorTest.testDiscoverPartitionsPeriodically failed on azure
> --
>
> Key: FLINK-25451
> URL: https://issues.apache.org/jira/browse/FLINK-25451
> Project: Flink
> Issue Type: Bug
> Components: Connectors / Kafka
>Affects Versions: 1.15.0
>Reporter: Yun Gao
>Priority: Major
> Labels: stale-major, test-stability
>
> {code:java}
> Dec 25 04:38:34 [ERROR] Tests run: 10, Failures: 0, Errors: 1, Skipped: 0,
> Time elapsed: 58.393 s <<< FAILURE! - in
> org.apache.flink.connector.kafka.source.enumerator.KafkaEnumeratorTest
> Dec 25 04:38:34 [ERROR]
> org.apache.flink.connector.kafka.source.enumerator.KafkaEnumeratorTest.testDiscoverPartitionsPeriodically
> Time elapsed: 30.01 s <<< ERROR!
> Dec 25 04:38:34 org.junit.runners.model.TestTimedOutException: test timed out
> after 3 milliseconds
> Dec 25 04:38:34 at java.lang.Object.wait(Native Method)
> Dec 25 04:38:34 at java.lang.Object.wait(Object.java:502)
> Dec 25 04:38:34 at
> org.apache.kafka.common.internals.KafkaFutureImpl$SingleWaiter.await(KafkaFutureImpl.java:92)
> Dec 25 04:38:34 at
> org.apache.kafka.common.internals.KafkaFutureImpl.get(KafkaFutureImpl.java:260)
> Dec 25 04:38:34 at
> org.apache.flink.connector.kafka.source.enumerator.KafkaEnumeratorTest.testDiscoverPartitionsPeriodically(KafkaEnumeratorTest.java:221)
> Dec 25 04:38:34 at sun.reflect.NativeMethodAccessorImpl.invoke0(Native
> Method)
> Dec 25 04:38:34 at
> sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
> Dec 25 04:38:34 at
> sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
> Dec 25 04:38:34 at java.lang.reflect.Method.invoke(Method.java:498)
> Dec 25 04:38:34 at
> org.junit.runners.model.FrameworkMethod$1.runReflectiveCall(FrameworkMethod.java:59)
> Dec 25 04:38:34 at
> org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:12)
> Dec 25 04:38:34 at
> org.junit.runners.model.FrameworkMethod.invokeExplosively(FrameworkMethod.java:56)
> Dec 25 04:38:34 at
> org.junit.internal.runners.statements.InvokeMethod.evaluate(InvokeMethod.java:17)
> Dec 25 04:38:34 at
> org.junit.internal.runners.statements.FailOnTimeout$CallableStatement.call(FailOnTimeout.java:299)
> Dec 25 04:38:34 at
> org.junit.internal.runners.statements.FailOnTimeout$CallableStatement.call(FailOnTimeout.java:293)
> Dec 25 04:38:34 at
> java.util.concurrent.FutureTask.run(FutureTask.java:266)
> Dec 25 04:38:34 at java.lang.Thread.run(Thread.java:748)
> Dec 25 04:38:34
> {code}
> https://dev.azure.com/apache-flink/apache-flink/_build/results?buildId=28590=logs=b0097207-033c-5d9a-b48c-6d4796fbe60d=8338a7d2-16f7-52e5-f576-4b7b3071eb3d=6549
--
This message was sent by Atlassian Jira
(v8.20.7#820007)
[jira] [Updated] (FLINK-25747) UdfStreamOperatorCheckpointingITCase hangs on AZP
[
https://issues.apache.org/jira/browse/FLINK-25747?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Flink Jira Bot updated FLINK-25747:
---
Labels: auto-deprioritized-critical auto-deprioritized-major
test-stability (was: auto-deprioritized-critical stale-major test-stability)
Priority: Minor (was: Major)
This issue was labeled "stale-major" 7 days ago and has not received any
updates so it is being deprioritized. If this ticket is actually Major, please
raise the priority and ask a committer to assign you the issue or revive the
public discussion.
> UdfStreamOperatorCheckpointingITCase hangs on AZP
> -
>
> Key: FLINK-25747
> URL: https://issues.apache.org/jira/browse/FLINK-25747
> Project: Flink
> Issue Type: Bug
> Components: Runtime / Checkpointing
>Affects Versions: 1.15.0
>Reporter: Till Rohrmann
>Priority: Minor
> Labels: auto-deprioritized-critical, auto-deprioritized-major,
> test-stability
>
> The test {{UdfStreamOperatorCheckpointingITCase}} hangs on AZP.
> https://dev.azure.com/apache-flink/apache-flink/_build/results?buildId=29840=logs=b0a398c0-685b-599c-eb57-c8c2a771138e=d13f554f-d4b9-50f8-30ee-d49c6fb0b3cc=15424
--
This message was sent by Atlassian Jira
(v8.20.7#820007)
[GitHub] [flink] chinmayms commented on pull request #19514: [FLINK-27308][Filesystem][S3] Update the Hadoop implementation for filesystems to 3.3.2
chinmayms commented on PR #19514: URL: https://github.com/apache/flink/pull/19514#issuecomment-1106842571 Thanks @MartijnVisser ! -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [flink] tweise commented on pull request #19514: [FLINK-27308][Filesystem][S3] Update the Hadoop implementation for filesystems to 3.3.2
tweise commented on PR #19514: URL: https://github.com/apache/flink/pull/19514#issuecomment-1106834064 @MartijnVisser thanks for fixing the test. LGTM. From what I see it is the case, but would like to confirm nevertheless: All shaded dependencies are covered by the NOTICE changes? -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [flink] MartijnVisser commented on pull request #19514: [FLINK-27308][Filesystem][S3] Update the Hadoop implementation for filesystems to 3.3.2
MartijnVisser commented on PR #19514: URL: https://github.com/apache/flink/pull/19514#issuecomment-1106821488 @tweise The S3 tests have passed, see https://dev.azure.com/apache-flink/apache-flink/_build/results?buildId=35010=results - Can you review the PR? CC @chinmayms -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[jira] [Created] (FLINK-27356) Incorrect Number of Methods Listed for SplitReader
Austin Cawley-Edwards created FLINK-27356: - Summary: Incorrect Number of Methods Listed for SplitReader Key: FLINK-27356 URL: https://issues.apache.org/jira/browse/FLINK-27356 Project: Flink Issue Type: Improvement Components: Documentation Affects Versions: 1.14.0, 1.15.0 Reporter: Austin Cawley-Edwards The docs state that [`SplitReader`|https://github.com/apache/flink/blob/master/flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/reader/splitreader/SplitReader.java] only has 3 methods, but it has four. The `close()` method is missing from the docs. -- This message was sent by Atlassian Jira (v8.20.7#820007)
[jira] [Commented] (FLINK-15123) remove uniqueKeys from FlinkStatistic in blink planner
[
https://issues.apache.org/jira/browse/FLINK-15123?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17526503#comment-17526503
]
Ling Jin commented on FLINK-15123:
--
It seems the issue has no progress for a long time, I would have a try.
I am going to walking the code for now, and feedback if I find the clue for how
to solve this.
> remove uniqueKeys from FlinkStatistic in blink planner
> ---
>
> Key: FLINK-15123
> URL: https://issues.apache.org/jira/browse/FLINK-15123
> Project: Flink
> Issue Type: Technical Debt
> Components: Table SQL / Planner
>Reporter: godfrey he
>Priority: Major
> Labels: starter
> Fix For: 1.16.0
>
> Attachments: b_5.txt
>
>
> {{uniqueKeys}} is a kind of constraint, it's unreasonable that {{uniqueKeys}}
> is a kind of statistic. so we should remove uniqueKeys from
> {{FlinkStatistic}} in blink planner. Some temporary solutions (e.g.
> {{RichTableSourceQueryOperation}}) should also be resolved after primaryKey
> is introduced in {{TableSchema}}
--
This message was sent by Atlassian Jira
(v8.20.7#820007)
[jira] [Commented] (FLINK-27155) Reduce multiple reads to the same Changelog file in the same taskmanager during restore
[ https://issues.apache.org/jira/browse/FLINK-27155?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17526494#comment-17526494 ] Feifan Wang commented on FLINK-27155: - I think task in RUNNING state not mean we can clean up the cache file, because changelog download and applying to delegated backend is in RUNNING state. As for the local space to store changelog cache file, I think it should be fine in most scenarios, can you describe some scenarios in which the changelog might be too large to fit on the local disk ? Or should we provide an option to limit the total size of the cache files ? Add another point, I think we can save the decompressed content to local cache file, so that we can seek to change set start position efficiently (use file seek rather than read all previous bytes which cause IO amplification). > Reduce multiple reads to the same Changelog file in the same taskmanager > during restore > --- > > Key: FLINK-27155 > URL: https://issues.apache.org/jira/browse/FLINK-27155 > Project: Flink > Issue Type: Sub-task > Components: Runtime / State Backends >Reporter: Feifan Wang >Assignee: Feifan Wang >Priority: Major > Fix For: 1.16.0 > > > h3. Background > In the current implementation, State changes of different operators in the > same taskmanager may be written to the same changelog file, which effectively > reduces the number of files and requests to DFS. > But on the other hand, the current implementation also reads the same > changelog file multiple times on recovery. More specifically, the number of > times the same changelog file is accessed is related to the number of > ChangeSets contained in it. And since each read needs to skip the preceding > bytes, this network traffic is also wasted. > The result is a lot of unnecessary request to DFS when there are multiple > slots and keyed state in the same taskmanager. > h3. Proposal > We can reduce multiple reads to the same changelog file in the same > taskmanager during restore. > One possible approach is to read the changelog file all at once and cache it > in memory or local file for a period of time when reading the changelog file. > I think this could be a subtask of [v2 FLIP-158: Generalized incremental > checkpoints|https://issues.apache.org/jira/browse/FLINK-25842] . > Hi [~ym] , [~roman] how do you think about ? -- This message was sent by Atlassian Jira (v8.20.7#820007)
[jira] [Commented] (FLINK-26586) FileSystem uses unbuffered read I/O
[ https://issues.apache.org/jira/browse/FLINK-26586?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17526488#comment-17526488 ] Matthias Schwalbe commented on FLINK-26586: --- [~akalashnikov] , I just managed to build a recent version of flink on our premises, except for flink-runtime-web. Would you agree to coach me a little for the formal stuff (e.g. fork on GitHub, PR preparation etc.)? As to your question: * I would make buffering optional to isolate possible regressions, also for random-access of small reads filling a big buffer can be counter-productive, ... and similar reasons > FileSystem uses unbuffered read I/O > --- > > Key: FLINK-26586 > URL: https://issues.apache.org/jira/browse/FLINK-26586 > Project: Flink > Issue Type: Improvement > Components: API / State Processor, Connectors / FileSystem, Runtime > / Checkpointing >Affects Versions: 1.13.0, 1.14.0 >Reporter: Matthias Schwalbe >Priority: Major > Attachments: BufferedFSDataInputStreamWrapper.java, > BufferedLocalFileSystem.java > > > - I found out that, at least when using LocalFileSystem on a windows system, > read I/O to load a savepoint is unbuffered, > - See example stack [1] > - i.e. in order to load only a long in a serializer, it needs to go into > kernel mode 8 times and load the 8 bytes one by one > - I coded a BufferedFSDataInputStreamWrapper that allows to opt-in buffered > reads on any FileSystem implementation > - In our setting savepoint load is now 30 times faster > - I’ve once seen a Jira ticket as to improve savepoint load time in general > (lost the link unfortunately), maybe this approach can help with it > - not sure if HDFS has got the same problem > - I can contribute my implementation of a BufferedFSDataInputStreamWrapper > which can be integrated in any > [1] unbuffered reads stack: > read:207, FileInputStream (java.io) > read:68, LocalDataInputStream (org.apache.flink.core.fs.local) > read:50, FSDataInputStreamWrapper (org.apache.flink.core.fs) > read:42, ForwardingInputStream (org.apache.flink.runtime.util) > readInt:390, DataInputStream (java.io) > deserialize:80, BytePrimitiveArraySerializer > (org.apache.flink.api.common.typeutils.base.array) > next:298, FullSnapshotRestoreOperation$KeyGroupEntriesIterator > (org.apache.flink.runtime.state.restore) > next:273, FullSnapshotRestoreOperation$KeyGroupEntriesIterator > (org.apache.flink.runtime.state.restore) > restoreKVStateData:147, RocksDBFullRestoreOperation > (org.apache.flink.contrib.streaming.state.restore) -- This message was sent by Atlassian Jira (v8.20.7#820007)
[GitHub] [flink-kubernetes-operator] SteNicholas commented on pull request #178: [FLINK-27334] Support auto generate the doc for the `KubernetesOperatorConfigOptions`
SteNicholas commented on PR #178: URL: https://github.com/apache/flink-kubernetes-operator/pull/178#issuecomment-1106611797 @wangyang0918, thanks for your review. I have addressed above comments and verified the configuration html manually. PTAL. cc @gyfora . -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [flink] snuyanzin commented on a diff in pull request #19552: [FLINK-27185][connectors][formats] Convert formats and connectors modules to assertj
snuyanzin commented on code in PR #19552:
URL: https://github.com/apache/flink/pull/19552#discussion_r856240226
##
flink-connectors/flink-connector-base/src/test/java/org/apache/flink/connector/base/source/reader/fetcher/SplitFetcherManagerTest.java:
##
@@ -100,7 +93,7 @@ private final void testExceptionPropagation(
fetcher.checkErrors();
fail("expected exception");
} catch (Exception e) {
-assertSame(testingException, e.getCause().getCause());
+assertThat(e.getCause().getCause()).isSameAs(testingException);
Review Comment:
Shouldn't it be replaced with `assertThatThrownBy(() -> ... )` ?
##
flink-connectors/flink-connector-cassandra/src/test/java/org/apache/flink/streaming/connectors/cassandra/CassandraSinkBaseTest.java:
##
@@ -159,9 +156,9 @@ public void testThrowErrorOnSnapshot() throws Exception {
try {
testHarness.snapshot(123L, 123L);
-Assert.fail();
+fail("unknown failure");
} catch (Exception e) {
-Assert.assertTrue(e.getCause() instanceof IOException);
+assertThat(e.getCause()).isInstanceOf(IOException.class);
}
Review Comment:
Could it be replaced with
```java
assertThatThrownBy(() -> testHarness.snapshot(123L, 123L))
.hasCauseInstanceOf(IOException.class);
```
?
##
flink-connectors/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/internal/connection/SimpleJdbcConnectionProviderTest.java:
##
@@ -140,19 +130,17 @@ public void testInvalidDriverUrl() throws Exception {
provider.getOrEstablishConnection();
fail("expect exception");
} catch (SQLException ex) {
-assertThat(
Review Comment:
Probably `assertThatThrownBy` could be used like
```java
assertThatThrownBy(provider::getOrEstablishConnection)
.isInstanceOf(SQLException.class)
.hasMessageContaining("No suitable driver found
for " + FakeDBUtils.TEST_DB_INVALID_URL);
```
##
flink-connectors/flink-connector-jdbc/src/test/java/org/apache/flink/connector/jdbc/utils/JdbcTypeUtilTest.java:
##
@@ -25,23 +25,23 @@
import java.sql.Types;
import static
org.apache.flink.connector.jdbc.utils.JdbcTypeUtil.logicalTypeToSqlType;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.fail;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.fail;
/** Testing the type conversions from Flink to SQL types. */
public class JdbcTypeUtilTest {
@Test
public void testTypeConversions() {
-assertEquals(Types.INTEGER,
logicalTypeToSqlType(LogicalTypeRoot.INTEGER));
+
assertThat(logicalTypeToSqlType(LogicalTypeRoot.INTEGER)).isEqualTo(Types.INTEGER);
testUnsupportedType(LogicalTypeRoot.RAW);
testUnsupportedType(LogicalTypeRoot.MAP);
}
private static void testUnsupportedType(LogicalTypeRoot logicalTypeRoot) {
try {
logicalTypeToSqlType(logicalTypeRoot);
-fail();
+fail("unknown failure");
} catch (IllegalArgumentException ignored) {
}
Review Comment:
Probably `try ... catch` could be replaced with `assertThatThrownBy`
##
flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaShortRetentionTestBase.java:
##
@@ -286,7 +286,7 @@ public void runFailOnAutoOffsetResetNoneEager() throws
Exception {
fail("should fail with an exception");
} catch (IllegalArgumentException e) {
// expected
-assertTrue(e.getMessage().contains("none"));
+assertThat(e.getMessage()).contains("none");
Review Comment:
Probably `try ... catch` could be replaced with `assertThatThrownBy`
##
flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/table/KafkaConnectorOptionsUtilTest.java:
##
@@ -118,14 +120,15 @@ public void testInvalidKeyFormatPrefixProjection() {
try {
createKeyFormatProjection(config, dataType);
-fail();
+fail("unknown failure");
Review Comment:
Probably `try ... catch` could be replaced with `assertThatThrownBy`
##
flink-formats/flink-csv/src/test/java/org/apache/flink/formats/csv/CsvRowDataSerDeSchemaTest.java:
##
@@ -196,11 +194,11 @@ public void
testSerializeDeserializeCustomizedProperties() throws Exception {
try {
testFieldDeserialization(
TIME(precision), "12:12:12.45",
LocalTime.parse("12:12:12"), deserConfig, ";");
-fail();
+fail("unknown failure");
Review Comment:
Probably `try ... catch` could
[jira] [Assigned] (FLINK-26052) Update chinese documentation regarding FLIP-203
[ https://issues.apache.org/jira/browse/FLINK-26052?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Dawid Wysakowicz reassigned FLINK-26052: Assignee: Feifan Wang > Update chinese documentation regarding FLIP-203 > --- > > Key: FLINK-26052 > URL: https://issues.apache.org/jira/browse/FLINK-26052 > Project: Flink > Issue Type: Sub-task > Components: Documentation, Runtime / Checkpointing >Reporter: Dawid Wysakowicz >Assignee: Feifan Wang >Priority: Minor > Labels: translation-zh > > Relevant english commits: > * c1f5c5320150402fc0cb4fbf3a31f9a27b1e4d9a > * cd8ea8d5b207569f68acc5a3c8db95cd2ca47ba6 -- This message was sent by Atlassian Jira (v8.20.7#820007)
[jira] [Commented] (FLINK-26052) Update chinese documentation regarding FLIP-203
[ https://issues.apache.org/jira/browse/FLINK-26052?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17526480#comment-17526480 ] Dawid Wysakowicz commented on FLINK-26052: -- Sure, I've assigned it to you. Thank you for your help! > Update chinese documentation regarding FLIP-203 > --- > > Key: FLINK-26052 > URL: https://issues.apache.org/jira/browse/FLINK-26052 > Project: Flink > Issue Type: Sub-task > Components: Documentation, Runtime / Checkpointing >Reporter: Dawid Wysakowicz >Priority: Minor > Labels: translation-zh > > Relevant english commits: > * c1f5c5320150402fc0cb4fbf3a31f9a27b1e4d9a > * cd8ea8d5b207569f68acc5a3c8db95cd2ca47ba6 -- This message was sent by Atlassian Jira (v8.20.7#820007)
[jira] [Resolved] (FLINK-27187) The attemptsPerUpload metric may be lower than it actually is
[ https://issues.apache.org/jira/browse/FLINK-27187?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Roman Khachatryan resolved FLINK-27187. --- Resolution: Fixed Thanks for adding this metric [~Feifan Wang], merged as cb68ccf1b2cb879148fb17d2fd6394e15d1ae46c. > The attemptsPerUpload metric may be lower than it actually is > - > > Key: FLINK-27187 > URL: https://issues.apache.org/jira/browse/FLINK-27187 > Project: Flink > Issue Type: Improvement > Components: Runtime / Metrics, Runtime / State Backends >Reporter: Feifan Wang >Assignee: Feifan Wang >Priority: Minor > Labels: pull-request-available > Fix For: 1.16.0 > > > The attemptsPerUpload metric in ChangelogStorageMetricGroup indicate > distributions of number of attempts per upload. > In the current implementation, each successful attempt try to update > attemptsPerUpload with its attemptNumber. > But consider this case: > # attempt 1 timeout, then schedule attempt 2 > # attempt 1 completed before attempt 2 and update attemptsPerUpload with 1 > In fact there are two attempts, but attemptsPerUpload updated with 1. > So, I think we should add "actionAttemptsCount" to > RetryExecutor.RetriableActionAttempt, this field shared across all attempts > to execute the same upload action representing the number of upload attempts. > And completed attempt should use this field update attemptsPerUpload. > > How do you think about ? [~ym] , [~roman] -- This message was sent by Atlassian Jira (v8.20.7#820007)
[jira] [Commented] (FLINK-26052) Update chinese documentation regarding FLIP-203
[ https://issues.apache.org/jira/browse/FLINK-26052?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17526478#comment-17526478 ] Feifan Wang commented on FLINK-26052: - Hi [~pnowojski] and [~dwysakowicz] , I'm a Chinese speaker, and I am glad to pick up this work, can you assign this ticket to me ? > Update chinese documentation regarding FLIP-203 > --- > > Key: FLINK-26052 > URL: https://issues.apache.org/jira/browse/FLINK-26052 > Project: Flink > Issue Type: Sub-task > Components: Documentation, Runtime / Checkpointing >Reporter: Dawid Wysakowicz >Priority: Minor > Labels: translation-zh > > Relevant english commits: > * c1f5c5320150402fc0cb4fbf3a31f9a27b1e4d9a > * cd8ea8d5b207569f68acc5a3c8db95cd2ca47ba6 -- This message was sent by Atlassian Jira (v8.20.7#820007)
[jira] [Updated] (FLINK-27343) flink jdbc sink will lead to unordered result, because the sink buffer records execute unorder
[ https://issues.apache.org/jira/browse/FLINK-27343?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] pengyusong updated FLINK-27343: --- Priority: Minor (was: Critical) > flink jdbc sink will lead to unordered result, because the sink buffer > records execute unorder > -- > > Key: FLINK-27343 > URL: https://issues.apache.org/jira/browse/FLINK-27343 > Project: Flink > Issue Type: Improvement > Components: Connectors / JDBC >Affects Versions: 1.13.6 > Environment: flink 1.13.6 > kafka > postgres jdbc sink >Reporter: pengyusong >Priority: Minor > > * situation one > when i use flink sql kafka connector re-consume a topic, the topic > already has many messages. > jdbc sink param with default. > kafka topic is a compact topic, which contents is a mysql table cdc > events. > there some records with same key in one batch, buffer within one batch, > finnaly sink to postgres with unorder, later record in the buffer batch are > executed first. > this will lead to the older message in kafka deal with after the newer > message, the results are inconsistent with kafka message orders. > * situation two > If i set > h5. sink.buffer-flush.interval = 0 > h5. sink.buffer-flush.max-rows = 1 > the result are inconsistent with kafka message orders. > > So, I have a suspicion that the order in jdbc buffer execute is > non-deterministic, lead to result in jdbc unordered. > > updated!!! > I found the order is my left join operator disorder the record order. The > question is left join why disorder the order -- This message was sent by Atlassian Jira (v8.20.7#820007)
[GitHub] [flink] rkhachatryan merged pull request #19441: [FLINK-27187][state/changelog] Add changelog storage metric totalAttemptsPerUpload
rkhachatryan merged PR #19441: URL: https://github.com/apache/flink/pull/19441 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[jira] [Resolved] (FLINK-27218) Serializer in OperatorState has not been updated when new Serializers are NOT incompatible
[
https://issues.apache.org/jira/browse/FLINK-27218?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Roman Khachatryan resolved FLINK-27218.
---
Fix Version/s: 1.16.0
1.15.1
Resolution: Fixed
Merged into master as 4033ddc5fa682a1619f8f22348e2ee38afcc1c85,
into 1.15 as 3d4b3a495b273c3a15ce7d35ba5a5b2e4ddc4c20.
> Serializer in OperatorState has not been updated when new Serializers are NOT
> incompatible
> --
>
> Key: FLINK-27218
> URL: https://issues.apache.org/jira/browse/FLINK-27218
> Project: Flink
> Issue Type: Bug
> Components: Runtime / State Backends
>Affects Versions: 1.15.1
>Reporter: Yue Ma
>Assignee: Yue Ma
>Priority: Major
> Labels: pull-request-available
> Fix For: 1.16.0, 1.15.1
>
> Attachments: image-2022-04-13-14-50-10-921.png,
> image-2022-04-18-21-48-30-519.png
>
>
> OperatorState such as *BroadcastState* or *PartitionableListState* can only
> be constructed via {*}DefaultOperatorStateBackend{*}. But when
> *BroadcastState* or *PartitionableListState* Serializer changes after we
> restart the job , it seems to have the following problems .
> As an example, we can see how PartitionableListState is initialized.
> First, RestoreOperation will construct a restored PartitionableListState
> based on the information in the snapshot.
> Then StateMetaInfo in partitionableListState will be updated as the
> following code
> {code:java}
> TypeSerializerSchemaCompatibility stateCompatibility =
>
> restoredPartitionableListStateMetaInfo.updatePartitionStateSerializer(newPartitionStateSerializer);
> partitionableListState.setStateMetaInfo(restoredPartitionableListStateMetaInfo);{code}
> The main problem is that there is also an *internalListCopySerializer* in
> *PartitionableListState* that is built using the previous Serializer and it
> has not been updated.
> Therefore, when we update the StateMetaInfo, the *internalListCopySerializer*
> also needs to be updated.
>
--
This message was sent by Atlassian Jira
(v8.20.7#820007)
[jira] [Closed] (FLINK-22984) UnsupportedOperationException when using Python UDF to generate watermark
[
https://issues.apache.org/jira/browse/FLINK-22984?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Dian Fu closed FLINK-22984.
---
Fix Version/s: 1.16.0
1.13.7
1.14.5
1.15.1
Assignee: Juntao Hu
Resolution: Fixed
Fixed in:
- master via 7ce5a7c6e1eab6823094a94bc0bca30d0ee618f1
- release-1.15 via 703b10ca5d004e8e79059e814fcf8503f84e2da8
- release-1.14 via 0806ad5a154e37d09b53ce56d59cec8dc11209da
- release-1.13 via 79a86f35fb321cb5f8dd40442db8c6bafb00153c
> UnsupportedOperationException when using Python UDF to generate watermark
> -
>
> Key: FLINK-22984
> URL: https://issues.apache.org/jira/browse/FLINK-22984
> Project: Flink
> Issue Type: Bug
> Components: API / Python
>Affects Versions: 1.13.0, 1.13.1
>Reporter: Maciej Bryński
>Assignee: Juntao Hu
>Priority: Minor
> Labels: auto-deprioritized-critical, auto-deprioritized-major,
> pull-request-available
> Fix For: 1.16.0, 1.13.7, 1.14.5, 1.15.1
>
>
> Hi,
> I'm trying to use output of Python UDF (parse_data) to set watermark for the
> table
> {code:java}
> CREATE TABLE test (
> data BYTES,
> ts as parse_data(data).ts,
> WATERMARK for ts as ts
> ) WITH (
>'connector' = 'kafka',
>'topic' = 'test',
>'properties.bootstrap.servers' = 'localhost:9092',
>'properties.group.id' = 'flink',
>'scan.startup.mode' = 'earliest-offset',
>'format' = 'raw'
> ){code}
> Then running SELECT on this table gives me exception
> {code:java}
> Py4JJavaError: An error occurred while calling o311.hasNext.
> : java.lang.RuntimeException: Failed to fetch next result
> at
> org.apache.flink.streaming.api.operators.collect.CollectResultIterator.nextResultFromFetcher(CollectResultIterator.java:109)
> at
> org.apache.flink.streaming.api.operators.collect.CollectResultIterator.hasNext(CollectResultIterator.java:80)
> at
> org.apache.flink.table.api.internal.TableResultImpl$CloseableRowIteratorWrapper.hasNext(TableResultImpl.java:370)
> at
> java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
> at
> java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
> at
> java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
> at java.base/java.lang.reflect.Method.invoke(Method.java:566)
> at
> org.apache.flink.api.python.shaded.py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)
> at
> org.apache.flink.api.python.shaded.py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)
> at
> org.apache.flink.api.python.shaded.py4j.Gateway.invoke(Gateway.java:282)
> at
> org.apache.flink.api.python.shaded.py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
> at
> org.apache.flink.api.python.shaded.py4j.commands.CallCommand.execute(CallCommand.java:79)
> at
> org.apache.flink.api.python.shaded.py4j.GatewayConnection.run(GatewayConnection.java:238)
> at java.base/java.lang.Thread.run(Thread.java:829)
> Caused by: java.io.IOException: Failed to fetch job execution result
> at
> org.apache.flink.streaming.api.operators.collect.CollectResultFetcher.getAccumulatorResults(CollectResultFetcher.java:177)
> at
> org.apache.flink.streaming.api.operators.collect.CollectResultFetcher.next(CollectResultFetcher.java:120)
> at
> org.apache.flink.streaming.api.operators.collect.CollectResultIterator.nextResultFromFetcher(CollectResultIterator.java:106)
> ... 13 more
> Caused by: java.util.concurrent.ExecutionException:
> org.apache.flink.runtime.client.JobExecutionException: Job execution failed.
> at
> java.base/java.util.concurrent.CompletableFuture.reportGet(CompletableFuture.java:395)
> at
> java.base/java.util.concurrent.CompletableFuture.get(CompletableFuture.java:2022)
> at
> org.apache.flink.streaming.api.operators.collect.CollectResultFetcher.getAccumulatorResults(CollectResultFetcher.java:175)
> ... 15 more
> Caused by: org.apache.flink.runtime.client.JobExecutionException: Job
> execution failed.
> at
> org.apache.flink.runtime.jobmaster.JobResult.toJobExecutionResult(JobResult.java:144)
> at
> org.apache.flink.runtime.minicluster.MiniClusterJobClient.lambda$getJobExecutionResult$3(MiniClusterJobClient.java:137)
> at
> java.base/java.util.concurrent.CompletableFuture.uniApplyNow(CompletableFuture.java:680)
> at
> java.base/java.util.concurrent.CompletableFuture.uniApplyStage(CompletableFuture.java:658)
> at
> java.base/java.util.concurrent.CompletableFuture.thenApply(CompletableFuture.java:2094)
> at
>
[GitHub] [flink-kubernetes-operator] SteNicholas commented on a diff in pull request #178: [FLINK-27334] Support auto generate the doc for the `KubernetesOperatorConfigOptions`
SteNicholas commented on code in PR #178: URL: https://github.com/apache/flink-kubernetes-operator/pull/178#discussion_r856126535 ## Dockerfile: ## @@ -23,21 +23,23 @@ WORKDIR /app ENV SHADED_DIR=flink-kubernetes-shaded ENV OPERATOR_DIR=flink-kubernetes-operator ENV WEBHOOK_DIR=flink-kubernetes-webhook +ENV DOCS_DIR=flink-kubernetes-docs RUN mkdir $OPERATOR_DIR $WEBHOOK_DIR COPY pom.xml . COPY $SHADED_DIR/pom.xml ./$SHADED_DIR/ COPY $WEBHOOK_DIR/pom.xml ./$WEBHOOK_DIR/ COPY $OPERATOR_DIR/pom.xml ./$OPERATOR_DIR/ +COPY $DOCS_DIR/pom.xml ./$DOCS_DIR/ Review Comment: @wangyang0918, only the pom.xml of `flink-kubernetes-docs`is copied into Dockerfile, which is introduced as module in pom.xml of parent and need to copy. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [flink] dianfu closed pull request #19551: [FLINK-22984][python] Disable PushWatermarkIntoTableSourceScanAcrossCalcRule when having Python UDF
dianfu closed pull request #19551: [FLINK-22984][python] Disable PushWatermarkIntoTableSourceScanAcrossCalcRule when having Python UDF URL: https://github.com/apache/flink/pull/19551 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [flink-kubernetes-operator] SteNicholas commented on a diff in pull request #178: [FLINK-27334] Support auto generate the doc for the `KubernetesOperatorConfigOptions`
SteNicholas commented on code in PR #178: URL: https://github.com/apache/flink-kubernetes-operator/pull/178#discussion_r856126535 ## Dockerfile: ## @@ -23,21 +23,23 @@ WORKDIR /app ENV SHADED_DIR=flink-kubernetes-shaded ENV OPERATOR_DIR=flink-kubernetes-operator ENV WEBHOOK_DIR=flink-kubernetes-webhook +ENV DOCS_DIR=flink-kubernetes-docs RUN mkdir $OPERATOR_DIR $WEBHOOK_DIR COPY pom.xml . COPY $SHADED_DIR/pom.xml ./$SHADED_DIR/ COPY $WEBHOOK_DIR/pom.xml ./$WEBHOOK_DIR/ COPY $OPERATOR_DIR/pom.xml ./$OPERATOR_DIR/ +COPY $DOCS_DIR/pom.xml ./$DOCS_DIR/ Review Comment: @wangyang0918, only the `pom.xml` of `flink-kubernetes-docs`is copied into Dockerfile, which is introduced as module in `pom.xml` of parent and need to copy. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [flink] rkhachatryan commented on pull request #19508: [FLINK-27218] fix the problem that the internal Serializer in Operato…
rkhachatryan commented on PR #19508: URL: https://github.com/apache/flink/pull/19508#issuecomment-1106582214 Thanks @mayuehappy. Merged the PR. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [flink] rkhachatryan merged pull request #19508: [FLINK-27218] fix the problem that the internal Serializer in Operato…
rkhachatryan merged PR #19508: URL: https://github.com/apache/flink/pull/19508 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [flink] rkhachatryan commented on a diff in pull request #18539: [FLINK-25745] Support RocksDB incremental native savepoints
rkhachatryan commented on code in PR #18539:
URL: https://github.com/apache/flink/pull/18539#discussion_r856290060
##
flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/Checkpoints.java:
##
@@ -369,4 +373,37 @@ public static CheckpointStorage loadCheckpointStorage(
/** This class contains only static utility methods and is not meant to be
instantiated. */
private Checkpoints() {}
+
+private static class ClaimModeCompletedStorageLocation
+implements CompletedCheckpointStorageLocation {
+
+private final CompletedCheckpointStorageLocation wrapped;
+
+private
ClaimModeCompletedStorageLocation(CompletedCheckpointStorageLocation location) {
+wrapped = location;
+}
+
+@Override
+public String getExternalPointer() {
+return wrapped.getExternalPointer();
+}
+
+@Override
+public StreamStateHandle getMetadataHandle() {
+return wrapped.getMetadataHandle();
+}
+
+@Override
+public void disposeStorageLocation() throws IOException {
+try {
+wrapped.disposeStorageLocation();
+} catch (Exception ex) {
+LOG.debug(
+"We could not delete the storage location: {} in CLAIM
restore mode. It is"
++ " most probably because of shared files
still being used by newer"
++ " checkpoints",
Review Comment:
Got it. Thanks for the explanation.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[jira] [Commented] (FLINK-27352) [JUnit5 Migration] Module: flink-json
[ https://issues.apache.org/jira/browse/FLINK-27352?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17526450#comment-17526450 ] EMing Zhou commented on FLINK-27352: Hi, Can you assigne to me?Thank you. > [JUnit5 Migration] Module: flink-json > - > > Key: FLINK-27352 > URL: https://issues.apache.org/jira/browse/FLINK-27352 > Project: Flink > Issue Type: Sub-task > Components: Formats (JSON, Avro, Parquet, ORC, SequenceFile) >Reporter: EMing Zhou >Priority: Minor > -- This message was sent by Atlassian Jira (v8.20.7#820007)
[GitHub] [flink-kubernetes-operator] gyfora commented on pull request #177: [FLINK-27303][FLINK-27309] Introduce FlinkConfigManager for efficient config management
gyfora commented on PR #177: URL: https://github.com/apache/flink-kubernetes-operator/pull/177#issuecomment-1106522084 Reworked the watcher logic and added some more tests -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [flink] dawidwys commented on a diff in pull request #18539: [FLINK-25745] Support RocksDB incremental native savepoints
dawidwys commented on code in PR #18539:
URL: https://github.com/apache/flink/pull/18539#discussion_r856213164
##
flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/Checkpoints.java:
##
@@ -369,4 +373,37 @@ public static CheckpointStorage loadCheckpointStorage(
/** This class contains only static utility methods and is not meant to be
instantiated. */
private Checkpoints() {}
+
+private static class ClaimModeCompletedStorageLocation
+implements CompletedCheckpointStorageLocation {
+
+private final CompletedCheckpointStorageLocation wrapped;
+
+private
ClaimModeCompletedStorageLocation(CompletedCheckpointStorageLocation location) {
+wrapped = location;
+}
+
+@Override
+public String getExternalPointer() {
+return wrapped.getExternalPointer();
+}
+
+@Override
+public StreamStateHandle getMetadataHandle() {
+return wrapped.getMetadataHandle();
+}
+
+@Override
+public void disposeStorageLocation() throws IOException {
+try {
+wrapped.disposeStorageLocation();
+} catch (Exception ex) {
+LOG.debug(
+"We could not delete the storage location: {} in CLAIM
restore mode. It is"
++ " most probably because of shared files
still being used by newer"
++ " checkpoints",
Review Comment:
This is the case for native savepoints. Native "incremental" RocksDB
savepoints place all files in a single savepoint folder. They do not place
shared files in the shared folder and thus they can prevent deleting the
savepoint folder.
This is the case because savepoints need to be self contained to be
relocatable.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [flink] dawidwys commented on a diff in pull request #18539: [FLINK-25745] Support RocksDB incremental native savepoints
dawidwys commented on code in PR #18539:
URL: https://github.com/apache/flink/pull/18539#discussion_r856213164
##
flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/Checkpoints.java:
##
@@ -369,4 +373,37 @@ public static CheckpointStorage loadCheckpointStorage(
/** This class contains only static utility methods and is not meant to be
instantiated. */
private Checkpoints() {}
+
+private static class ClaimModeCompletedStorageLocation
+implements CompletedCheckpointStorageLocation {
+
+private final CompletedCheckpointStorageLocation wrapped;
+
+private
ClaimModeCompletedStorageLocation(CompletedCheckpointStorageLocation location) {
+wrapped = location;
+}
+
+@Override
+public String getExternalPointer() {
+return wrapped.getExternalPointer();
+}
+
+@Override
+public StreamStateHandle getMetadataHandle() {
+return wrapped.getMetadataHandle();
+}
+
+@Override
+public void disposeStorageLocation() throws IOException {
+try {
+wrapped.disposeStorageLocation();
+} catch (Exception ex) {
+LOG.debug(
+"We could not delete the storage location: {} in CLAIM
restore mode. It is"
++ " most probably because of shared files
still being used by newer"
++ " checkpoints",
Review Comment:
This is the case for native savepoints. Native "incremental" RocksDB
savepoints place all files in a single savepoint folder. They do not place
shared files in the shared folder and thus they can prevent deleting the
savepoint folder.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[jira] (FLINK-27340) [JUnit5 Migration] Module: flink-python
[ https://issues.apache.org/jira/browse/FLINK-27340 ] EMing Zhou deleted comment on FLINK-27340: was (Author: zsigner): Hi [~Sergey Nuyanzin] Can I get the ticket? > [JUnit5 Migration] Module: flink-python > --- > > Key: FLINK-27340 > URL: https://issues.apache.org/jira/browse/FLINK-27340 > Project: Flink > Issue Type: Sub-task > Components: API / Python, Tests >Reporter: Sergey Nuyanzin >Priority: Major > -- This message was sent by Atlassian Jira (v8.20.7#820007)
[GitHub] [flink] rkhachatryan commented on a diff in pull request #18539: [FLINK-25745] Support RocksDB incremental native savepoints
rkhachatryan commented on code in PR #18539:
URL: https://github.com/apache/flink/pull/18539#discussion_r856196969
##
flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/Checkpoints.java:
##
@@ -369,4 +373,37 @@ public static CheckpointStorage loadCheckpointStorage(
/** This class contains only static utility methods and is not meant to be
instantiated. */
private Checkpoints() {}
+
+private static class ClaimModeCompletedStorageLocation
+implements CompletedCheckpointStorageLocation {
+
+private final CompletedCheckpointStorageLocation wrapped;
+
+private
ClaimModeCompletedStorageLocation(CompletedCheckpointStorageLocation location) {
+wrapped = location;
+}
+
+@Override
+public String getExternalPointer() {
+return wrapped.getExternalPointer();
+}
+
+@Override
+public StreamStateHandle getMetadataHandle() {
+return wrapped.getMetadataHandle();
+}
+
+@Override
+public void disposeStorageLocation() throws IOException {
+try {
+wrapped.disposeStorageLocation();
+} catch (Exception ex) {
+LOG.debug(
+"We could not delete the storage location: {} in CLAIM
restore mode. It is"
++ " most probably because of shared files
still being used by newer"
++ " checkpoints",
Review Comment:
@dawidwys could you please explain this scenario?
AFAIK, shared state files should be placed in a separate
`checkpoints/shared` folder and therefore should not prevent
`checkpoints/chk-xxx` folder from being deleted.
However, this is not true when migrating to Changelog, when private state
becomes "re-usable".
I'd like to understand whether it's a general case or only related to
Changelog.
cc: @zoltar9264
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [flink] JulioPerezGitHub commented on pull request #17020: ---
JulioPerezGitHub commented on PR #17020: URL: https://github.com/apache/flink/pull/17020#issuecomment-1106481367 https://discord.gg/ccZe2WMd -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [flink] flinkbot commented on pull request #19557: [hotfix][docs] Fix class name in docs for ExecutionEnvironment class
flinkbot commented on PR #19557: URL: https://github.com/apache/flink/pull/19557#issuecomment-1106473407 ## CI report: * 1940799765d3b35c3e93f7bc6f78138f1e33fbbb UNKNOWN Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run azure` re-run the last Azure build -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [flink] flinkbot commented on pull request #19556: [FLINK-26413][hive] Hive dialect supports "LOAD DATA INPATH"
flinkbot commented on PR #19556: URL: https://github.com/apache/flink/pull/19556#issuecomment-1106473307 ## CI report: * 4d99ef8a1ea72740067fb5838d9c854a622be5d5 UNKNOWN Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run azure` re-run the last Azure build -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [flink] snuyanzin opened a new pull request, #19557: [hotfix][docs] Fix class name in docs for ExecutionEnvironment class
snuyanzin opened a new pull request, #19557: URL: https://github.com/apache/flink/pull/19557 ## What is the purpose of the change Trivial misprint fix class name for `ExecutionEnvironment` class in docs/content.zh/docs/connectors/datastream/formats/hadoop.md docs/content.zh/docs/dev/dataset/hadoop_compatibility.md docs/content/docs/connectors/dataset/formats/hadoop.md docs/content/docs/connectors/datastream/formats/hadoop.md `ExecutionEnvironmen` => `ExecutionEnvironment` Verifying this change This change is a trivial rework / code cleanup without any test coverage. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[jira] [Updated] (FLINK-26413) Hive dialect support "LOAD DATA INPATH"
[
https://issues.apache.org/jira/browse/FLINK-26413?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
ASF GitHub Bot updated FLINK-26413:
---
Labels: pull-request-available (was: )
> Hive dialect support "LOAD DATA INPATH"
>
>
> Key: FLINK-26413
> URL: https://issues.apache.org/jira/browse/FLINK-26413
> Project: Flink
> Issue Type: Sub-task
>Reporter: luoyuxia
>Priority: Major
> Labels: pull-request-available
>
> In Hive, it's supported to use such sql like
> {code:java}
> LOAD DATA INPATH
> {code}
> to import data to hive table.
> It's also need to support it using Hive dialect in Flink.
--
This message was sent by Atlassian Jira
(v8.20.7#820007)
[GitHub] [flink] luoyuxia opened a new pull request, #19556: [FLINK-26413][hive] Hive dialect supports "LOAD DATA INPATH"
luoyuxia opened a new pull request, #19556: URL: https://github.com/apache/flink/pull/19556 ## What is the purpose of the change *(For example: This pull request makes task deployment go through the blob server, rather than through RPC. That way we avoid re-transferring them on each deployment (during recovery).)* ## Brief change log *(for example:)* - *The TaskInfo is stored in the blob store on job creation time as a persistent artifact* - *Deployments RPC transmits only the blob storage reference* - *TaskManagers retrieve the TaskInfo from the blob cache* ## Verifying this change Please make sure both new and modified tests in this PR follows the conventions defined in our code quality guide: https://flink.apache.org/contributing/code-style-and-quality-common.html#testing *(Please pick either of the following options)* This change is a trivial rework / code cleanup without any test coverage. *(or)* This change is already covered by existing tests, such as *(please describe tests)*. *(or)* This change added tests and can be verified as follows: *(example:)* - *Added integration tests for end-to-end deployment with large payloads (100MB)* - *Extended integration test for recovery after master (JobManager) failure* - *Added test that validates that TaskInfo is transferred only once across recoveries* - *Manually verified the change by running a 4 node cluster with 2 JobManagers and 4 TaskManagers, a stateful streaming program, and killing one JobManager and two TaskManagers during the execution, verifying that recovery happens correctly.* ## Does this pull request potentially affect one of the following parts: - Dependencies (does it add or upgrade a dependency): (yes / no) - The public API, i.e., is any changed class annotated with `@Public(Evolving)`: (yes / no) - The serializers: (yes / no / don't know) - The runtime per-record code paths (performance sensitive): (yes / no / don't know) - Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: (yes / no / don't know) - The S3 file system connector: (yes / no / don't know) ## Documentation - Does this pull request introduce a new feature? (yes / no) - If yes, how is the feature documented? (not applicable / docs / JavaDocs / not documented) -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [flink-table-store] JingsongLi commented on pull request #99: [FLINK-27307] Flink table store support append-only ingestion without primary keys.
JingsongLi commented on PR #99: URL: https://github.com/apache/flink-table-store/pull/99#issuecomment-1106466192 > The different manifests design for both two kinds of tables. Can you clarify which parts of the design are different? > What's the read API abstraction for those two kinds of tables. I still don't have a clearly propose for it. Will try to update this PR for this. A new `RecordReader` too? -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [flink-table-store] JingsongLi commented on a diff in pull request #99: [FLINK-27307] Flink table store support append-only ingestion without primary keys.
JingsongLi commented on code in PR #99:
URL: https://github.com/apache/flink-table-store/pull/99#discussion_r856178226
##
flink-table-store-core/src/main/java/org/apache/flink/table/store/file/WriteMode.java:
##
@@ -0,0 +1,45 @@
+/*
+ * 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.flink.table.store.file;
+
+/** Defines the write mode for flink table store. */
+public enum WriteMode {
+INSERT_ONLY(
+"insert-only",
+"The table can only accept append-only insert operations. All rows
will be "
++ "inserted into the table store without any deduplication
or primary/unique key constraint"),
+DELETABLE("deletable", "The table can accept both insert and operations.");
Review Comment:
Maybe just `changelog`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [flink-table-store] JingsongLi commented on pull request #99: [FLINK-27307] Flink table store support append-only ingestion without primary keys.
JingsongLi commented on PR #99: URL: https://github.com/apache/flink-table-store/pull/99#issuecomment-1106461636 Can we set the append-only write file to an empty key? This allows for a good integration of these two modes. Actually, `SstFileMeta` can be renamed to `DataFileMeta`. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [flink] zoltar9264 commented on pull request #19441: [FLINK-27187][state/changelog] Add changelog storage metric totalAttemptsPerUpload
zoltar9264 commented on PR #19441: URL: https://github.com/apache/flink/pull/19441#issuecomment-1106451265 @flinkbot run azure -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[jira] [Comment Edited] (FLINK-27174) Non-null check for bootstrapServers field is incorrect in KafkaSink
[
https://issues.apache.org/jira/browse/FLINK-27174?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17526356#comment-17526356
]
Zhengqi Zhang edited comment on FLINK-27174 at 4/22/22 11:57 AM:
-
Yes. In the current code, if the user does not use the setBootstrapServers
method to set bootstrapServers, even if he provides it in a separate property,
the non-null check on bootstrapServers will fail, which is obviously
unreasonable. In fact, we can just check bootstrapServers in the final property.
was (Author: tony giao):
In the current code, if the user does not use the setBootstrapServers method to
set bootstrapServers, even if he provides it in a separate property, the
non-null check on bootstrapServers will fail, which is obviously unreasonable.
In fact, we can just check bootstrapServers in the final property.
> Non-null check for bootstrapServers field is incorrect in KafkaSink
> ---
>
> Key: FLINK-27174
> URL: https://issues.apache.org/jira/browse/FLINK-27174
> Project: Flink
> Issue Type: Bug
> Components: Connectors / Kafka
>Affects Versions: 1.14.4
>Reporter: Zhengqi Zhang
>Priority: Major
> Labels: easyfix
> Attachments: image-2022-04-11-18-11-18-576.png,
> image-2022-04-11-18-17-48-514.png
>
>
> If the user-supplied kafkaProducerConfig contains bootstrapServers
> information, there is no need to define the value of this field separately
> through the setBootstrapServers method. Obviously, the current code doesn't
> notice this.
> !image-2022-04-11-18-11-18-576.png|width=859,height=261!
>
> Perhaps we can check bootstrapServers as follows:
> !image-2022-04-11-18-17-48-514.png|width=861,height=322!
>
> {color:#172b4d}Or check bootstrapServers like KafkaSourceBuilder.{color}
>
>
--
This message was sent by Atlassian Jira
(v8.20.7#820007)
[jira] [Commented] (FLINK-27174) Non-null check for bootstrapServers field is incorrect in KafkaSink
[
https://issues.apache.org/jira/browse/FLINK-27174?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17526356#comment-17526356
]
Zhengqi Zhang commented on FLINK-27174:
---
In the current code, if the user does not use the setBootstrapServers method to
set bootstrapServers, even if he provides it in a separate property, the
non-null check on bootstrapServers will fail, which is obviously unreasonable.
In fact, we can just check bootstrapServers in the final property.
> Non-null check for bootstrapServers field is incorrect in KafkaSink
> ---
>
> Key: FLINK-27174
> URL: https://issues.apache.org/jira/browse/FLINK-27174
> Project: Flink
> Issue Type: Bug
> Components: Connectors / Kafka
>Affects Versions: 1.14.4
>Reporter: Zhengqi Zhang
>Priority: Major
> Labels: easyfix
> Attachments: image-2022-04-11-18-11-18-576.png,
> image-2022-04-11-18-17-48-514.png
>
>
> If the user-supplied kafkaProducerConfig contains bootstrapServers
> information, there is no need to define the value of this field separately
> through the setBootstrapServers method. Obviously, the current code doesn't
> notice this.
> !image-2022-04-11-18-11-18-576.png|width=859,height=261!
>
> Perhaps we can check bootstrapServers as follows:
> !image-2022-04-11-18-17-48-514.png|width=861,height=322!
>
> {color:#172b4d}Or check bootstrapServers like KafkaSourceBuilder.{color}
>
>
--
This message was sent by Atlassian Jira
(v8.20.7#820007)
[GitHub] [flink] rkhachatryan commented on a diff in pull request #19441: [FLINK-27187][state/changelog] Add changelog storage metric totalAttemptsPerUpload
rkhachatryan commented on code in PR #19441:
URL: https://github.com/apache/flink/pull/19441#discussion_r856152839
##
flink-dstl/flink-dstl-dfs/src/test/java/org/apache/flink/changelog/fs/ChangelogStorageMetricsTest.java:
##
@@ -295,6 +346,55 @@ public void close() {
}
}
+private static class WaitingMaxAttemptUploader implements
StateChangeUploader {
+private final ConcurrentHashMap
remainingAttemptsPerTask;
+private final int maxAttempts;
+
+public WaitingMaxAttemptUploader(int maxAttempts) {
+if (maxAttempts < 1) {
+throw new IllegalArgumentException("maxAttempts < 0");
+}
+this.maxAttempts = maxAttempts;
+this.remainingAttemptsPerTask = new ConcurrentHashMap<>();
+}
+
+@Override
+public UploadTasksResult upload(Collection tasks) throws
IOException {
+
+for (UploadTask uploadTask : tasks) {
+CountDownLatch remainingAttempts =
remainingAttemptsPerTask.get(uploadTask);
+if (remainingAttempts == null) {
+remainingAttempts = new CountDownLatch(maxAttempts - 1);
+remainingAttemptsPerTask.put(uploadTask,
remainingAttempts);
+} else {
+remainingAttempts.countDown();
+}
+}
+for (UploadTask uploadTask : tasks) {
+CountDownLatch remainingAttempts =
remainingAttemptsPerTask.get(uploadTask);
+try {
+remainingAttempts.await();
Review Comment:
You're right, completing first and last attempts sounds good.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [flink-kubernetes-operator] gyfora commented on a diff in pull request #176: [FLINK-27279] Extract common status interfaces
gyfora commented on code in PR #176:
URL:
https://github.com/apache/flink-kubernetes-operator/pull/176#discussion_r856146587
##
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/status/ReconciliationStatus.java:
##
@@ -54,19 +49,20 @@ public class ReconciliationStatus {
private ReconciliationState state = ReconciliationState.DEPLOYED;
@JsonIgnore
-public FlinkDeploymentSpec deserializeLastReconciledSpec() {
-return ReconciliationUtils.deserializedSpecWithVersion(
-lastReconciledSpec, FlinkDeploymentSpec.class);
+public abstract Class getSpecClass();
Review Comment:
You could also simply add a constructor that takes the spec class (you have
that when you call initStatus by calling getSpec().getClass()). Then you would
not need 2 subclasses just to implement this.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [flink] fapaul commented on a diff in pull request #19405: [FLINK-27066] Reintroduce e2e tests in ES as Java tests.
fapaul commented on code in PR #19405:
URL: https://github.com/apache/flink/pull/19405#discussion_r856141131
##
flink-end-to-end-tests/flink-end-to-end-tests-elasticsearch7/pom.xml:
##
@@ -30,10 +30,14 @@ under the License.
..
- flink-elasticsearch7-test
- Flink : E2E Tests : Elasticsearch 7
+ flink-end-to-end-tests-elasticsearch7
+ Flink : E2E Tests : Elasticsearch 7 Java
Review Comment:
Is `Java` really important here?
##
flink-end-to-end-tests/flink-end-to-end-tests-common-elasticsearch/pom.xml:
##
@@ -43,50 +43,52 @@ under the License.
org.apache.flink
- flink-connector-elasticsearch6
+ flink-connector-test-utils
${project.version}
+ compile
+
+ org.apache.flink
+
flink-connector-elasticsearch-base
+ ${project.version}
+
+
+ org.testcontainers
+ elasticsearch
+ ${testcontainers.version}
+
+
+ org.apache.flink
+ flink-end-to-end-tests-common
+ ${project.version}
+ test
+
+
+
+org.apache.httpcomponents
+httpcore-nio
+4.4.12
+
+
+
+
org.apache.maven.plugins
- maven-shade-plugin
+ maven-jar-plugin
-
Elasticsearch6SinkExample
+ Jar Package
package
- shade
+ test-jar
Review Comment:
Can you avoid introducing more test jars and maybe move the common utilities
to the main package?
##
flink-end-to-end-tests/flink-end-to-end-tests-common-elasticsearch/src/test/java/org/apache/flink/streaming/tests/ElasticsearchSinkE2ECaseBase.java:
##
@@ -0,0 +1,92 @@
+/*
+ * 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.flink.streaming.tests;
+
+import
org.apache.flink.connector.testframe.external.DefaultContainerizedExternalSystem;
+import org.apache.flink.connector.testframe.external.ExternalSystemDataReader;
+import org.apache.flink.connector.testframe.junit.annotations.TestEnv;
+import
org.apache.flink.connector.testframe.junit.annotations.TestExternalSystem;
+import org.apache.flink.connector.testframe.junit.annotations.TestSemantics;
+import org.apache.flink.connector.testframe.testsuites.SinkTestSuiteBase;
+import org.apache.flink.streaming.api.CheckpointingMode;
+import org.apache.flink.tests.util.flink.FlinkContainerTestEnvironment;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testcontainers.elasticsearch.ElasticsearchContainer;
+import org.testcontainers.utility.DockerImageName;
+
+import java.time.Duration;
+import java.util.Collections;
+import java.util.List;
+
+import static
org.apache.flink.connector.testframe.utils.CollectIteratorAssertions.assertThat;
+import static
org.apache.flink.runtime.testutils.CommonTestUtils.waitUntilCondition;
+
+/** Base classs for end to end ElasticsearchSink tests based on connector
testing framework. */
+@SuppressWarnings("unused")
+public abstract class ElasticsearchSinkE2ECaseBase>
+extends SinkTestSuiteBase {
+
+private static final Logger LOG =
LoggerFactory.getLogger(ElasticsearchSinkE2ECaseBase.class);
+private static final int READER_RETRY_ATTEMPTS = 10;
+private static final int READER_TIMEOUT = 10;
+
+protected
[GitHub] [flink-kubernetes-operator] gyfora commented on a diff in pull request #176: [FLINK-27279] Extract common status interfaces
gyfora commented on code in PR #176:
URL:
https://github.com/apache/flink-kubernetes-operator/pull/176#discussion_r856144200
##
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/ReconcileTarget.java:
##
@@ -0,0 +1,62 @@
+/*
+ * 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.flink.kubernetes.operator.reconciler;
+
+import org.apache.flink.kubernetes.operator.crd.spec.JobSpec;
+import org.apache.flink.kubernetes.operator.crd.status.ReconciliationStatus;
+
+import javax.annotation.Nullable;
+
+/**
+ * The interface is responsible to handle the reconciliation result. For the
common logic, it
+ * provides method to extract the common view between the {@link
+ * org.apache.flink.kubernetes.operator.crd.FlinkDeployment} and {@link
+ * org.apache.flink.kubernetes.operator.crd.FlinkSessionJob} to simplify the
custom resource
+ * manipulation. For the special part of each custom resource, we can extend
the interface to let
+ * the target custom resource react to the reconciliation result
correspondingly.
+ *
+ * @param the common view of the custom resource getSpec
+ */
+public interface ReconcileTarget {
+
+/** The common view of the spec. */
+interface SpecView {
+JobSpec getJobSpec();
+}
+
+/**
+ * Get the current getSpec of the custom resource.
+ *
+ * @return the current getSpec.
+ */
+SPEC getSpec();
+
+/**
+ * Get the current reconciliation status.
+ *
+ * @return the current reconciliation status.
+ */
+ReconciliationStatus getReconcileStatus();
+
+/**
+ * Let the target custom resource handle the reconciliation error.
+ *
+ * @param error The error to be handled.
+ */
+void handleError(@Nullable String error);
Review Comment:
That way the ReconcileTerget interface could be also removed and use
`CustomResource`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [flink-kubernetes-operator] gyfora commented on a diff in pull request #176: [FLINK-27279] Extract common status interfaces
gyfora commented on code in PR #176:
URL:
https://github.com/apache/flink-kubernetes-operator/pull/176#discussion_r856142721
##
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/ReconcileTarget.java:
##
@@ -0,0 +1,62 @@
+/*
+ * 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.flink.kubernetes.operator.reconciler;
+
+import org.apache.flink.kubernetes.operator.crd.spec.JobSpec;
+import org.apache.flink.kubernetes.operator.crd.status.ReconciliationStatus;
+
+import javax.annotation.Nullable;
+
+/**
+ * The interface is responsible to handle the reconciliation result. For the
common logic, it
+ * provides method to extract the common view between the {@link
+ * org.apache.flink.kubernetes.operator.crd.FlinkDeployment} and {@link
+ * org.apache.flink.kubernetes.operator.crd.FlinkSessionJob} to simplify the
custom resource
+ * manipulation. For the special part of each custom resource, we can extend
the interface to let
+ * the target custom resource react to the reconciliation result
correspondingly.
+ *
+ * @param the common view of the custom resource getSpec
+ */
+public interface ReconcileTarget {
+
+/** The common view of the spec. */
+interface SpecView {
+JobSpec getJobSpec();
+}
+
+/**
+ * Get the current getSpec of the custom resource.
+ *
+ * @return the current getSpec.
+ */
+SPEC getSpec();
+
+/**
+ * Get the current reconciliation status.
+ *
+ * @return the current reconciliation status.
+ */
+ReconciliationStatus getReconcileStatus();
+
+/**
+ * Let the target custom resource handle the reconciliation error.
+ *
+ * @param error The error to be handled.
+ */
+void handleError(@Nullable String error);
Review Comment:
I think these methods should be part of the CommonStatus interface instead.
Simply have the ReonciliationStatus and error fields there and use get/set
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [flink] snuyanzin commented on pull request #18109: [FLINK-25284][Table SQL / API] Add proportion of null values to generate with datagen
snuyanzin commented on PR #18109: URL: https://github.com/apache/flink/pull/18109#issuecomment-1106425611 @JingsongLi sorry for the poke Since you are one of the committers dealing with datagen, could you please have a look here, once you have time? -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [flink-web] NicoK commented on a diff in pull request #526: Announcement blogpost for the 1.15 release
NicoK commented on code in PR #526: URL: https://github.com/apache/flink-web/pull/526#discussion_r856088102 ## _posts/2022-04-11-1.15-announcement.md: ## @@ -0,0 +1,431 @@ +--- +layout: post +title: "Announcing the Release of Apache Flink 1.15" +subtitle: "" +date: 2022-04-11T08:00:00.000Z +categories: news +authors: +- yungao: + name: "Yun Gao" + twitter: "YunGao16" +- joemoe: + name: "Joe Moser" + twitter: "JoemoeAT" + +--- + +Thanks to our well-organized and open community, Apache Flink continues +[to grow](https://www.apache.org/foundation/docs/FY2021AnnualReport.pdf) as a +technology and remain one of the most active projects in +the Apache community. With the release of Flink 1.15, we are proud to announce a number of +exciting changes. + +One of the main concepts that makes Apache Flink stand out is the unification of +batch (aka bounded) and stream (aka unbounded) data processing. A lot of +effort went into this unification in the previous releases but you can expect more efforts in this direction. +Apache Flink is not only growing when it comes to contributions and users, but +also out of the original use cases. We are seeing a trend towards more business/analytics +use cases implemented in low-/no-code. Flink SQL is the feature in the Flink ecosystem +that enables such uses cases and this is why its popularity continues to grow. + +Apache Flink is an essential building block in data pipelines/architectures and +is used with many other technologies in order to drive all sorts of use cases. While new ideas/products +may appear in this domain, existing technologies continue to establish themselves as standards for solving +mission-critical problems. Knowing that we have such a wide reach and play a role in the success of many +projects, it is important that the experience of +integrating with Apache Flink is as seamless and easy as possible. + +In the 1.15 release the Apache Flink community made significant progress across all +these areas. Still those are not the only things that made it into 1.15. The +contributors improved the experience of operating Apache Flink by making it much +easier and more transparent to handle checkpoints and savepoints and their ownership, +making auto scaling more seamless and complete, by removing side effects of use cases +in which different data sources produce varying amounts of data, and - finally - the +ability to upgrade SQL jobs without losing the state. By continuing on supporting +checkpoints after tasks finished and adding window table valued functions in batch +mode, the experience of unified stream and batch processing was once more improved +making hybrid use cases way easier. In the SQL space, not only the first step in +version upgrades have been added but also JSON functions to make it easier to import +and export structured data in SQL. Both will allow users to better rely on Flink SQL +for production use cases in the long term. To establish Apache Flink as part of the +data processing ecosystem we improved the cloud interoperability and added more sink +connectors and formats. And yes we enabled a Scala-free runtime +([the hype is real](https://flink.apache.org/2022/02/22/scala-free.html)). + + +## Operating Apache Flink with ease + +Even Flink jobs that have been built and tuned by the best engineering teams still need to +be operated, usually on a long-term basis. The many deployment +patterns, APIs, tuneable configs, and use cases covered by Apache Flink mean that operation +support is vital and can be burdensome. + +In this release, we listened to user feedback and now operating Flink is made much +easier. It is now more transparent in terms of handling checkpoints and savepoints and their ownership, +which makes auto-scaling more seamless and complete (by removing side effects of use cases +where different data sources produce varying amounts of data) and enables the +ability to upgrade SQL jobs without losing the state. + + +### Clarification of checkpoint and savepoint semantics + +An essential cornerstone of Flink’s fault tolerance strategy is based on +[checkpoints](https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/ops/state/checkpoints/) and +[savepoints](https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/ops/state/savepoints/) (see [the comparison](https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/ops/state/checkpoints_vs_savepoints/). Review Comment: ```suggestion [savepoints](https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/ops/state/savepoints/) (see [the comparison](https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/ops/state/checkpoints_vs_savepoints/)). ``` ## _posts/2022-04-11-1.15-announcement.md: ## @@ -0,0 +1,431 @@ +--- +layout: post +title: "Announcing the Release of Apache Flink 1.15" +subtitle: "" +date:
[jira] [Created] (FLINK-27355) JobManagerRunnerRegistry.localCleanupAsync does not call the JobManagerRunner.close method repeatedly
Matthias Pohl created FLINK-27355:
-
Summary: JobManagerRunnerRegistry.localCleanupAsync does not call
the JobManagerRunner.close method repeatedly
Key: FLINK-27355
URL: https://issues.apache.org/jira/browse/FLINK-27355
Project: Flink
Issue Type: Bug
Components: Runtime / Coordination
Affects Versions: 1.15.0
Reporter: Matthias Pohl
The {{DefaultJobManagerRunner.localCleanupAsync}} method deregisters the
JobManagerRunner and calls close on it. If close fails for whatever reason, it
will be identified but the next retry would just notice that the
JobManagerRunner is already deregistered and not do anything.
Hence, JobMaster shutdown won't be retriggered (i.e. errors in the
{{CompletedCheckpointStore}} or the {{CheckpointIDCounter}} won't be handled).
FLINK-26114 is related: Both components don't expose any errors right now,
anyway.
--
This message was sent by Atlassian Jira
(v8.20.7#820007)
[GitHub] [flink-kubernetes-operator] SteNicholas commented on a diff in pull request #178: [FLINK-27334] Support auto generate the doc for the `KubernetesOperatorConfigOptions`
SteNicholas commented on code in PR #178: URL: https://github.com/apache/flink-kubernetes-operator/pull/178#discussion_r856126535 ## Dockerfile: ## @@ -23,21 +23,23 @@ WORKDIR /app ENV SHADED_DIR=flink-kubernetes-shaded ENV OPERATOR_DIR=flink-kubernetes-operator ENV WEBHOOK_DIR=flink-kubernetes-webhook +ENV DOCS_DIR=flink-kubernetes-docs RUN mkdir $OPERATOR_DIR $WEBHOOK_DIR COPY pom.xml . COPY $SHADED_DIR/pom.xml ./$SHADED_DIR/ COPY $WEBHOOK_DIR/pom.xml ./$WEBHOOK_DIR/ COPY $OPERATOR_DIR/pom.xml ./$OPERATOR_DIR/ +COPY $DOCS_DIR/pom.xml ./$DOCS_DIR/ Review Comment: @wangyang0918, this only adds the pom.xml of the `flink-kubernetes-docs`, which is used in parent pom.xml. I have tried not to copy this file but the maven command throwed the module not existing. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[jira] [Assigned] (FLINK-26114) DefaultScheduler fails fatally in case of an error when shutting down the checkpoint-related resources
[
https://issues.apache.org/jira/browse/FLINK-26114?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Matthias Pohl reassigned FLINK-26114:
-
Assignee: (was: Niklas Semmler)
> DefaultScheduler fails fatally in case of an error when shutting down the
> checkpoint-related resources
> --
>
> Key: FLINK-26114
> URL: https://issues.apache.org/jira/browse/FLINK-26114
> Project: Flink
> Issue Type: Bug
> Components: Runtime / Coordination
>Affects Versions: 1.15.0
>Reporter: Matthias Pohl
>Priority: Critical
>
> In contrast to the {{AdaptiveScheduler}}, the {{DefaultScheduler}} fails
> fatally in case of an error while cleaning up the checkpoint-related
> resources. This contradicts our new approach of retrying the cleanup of
> job-related data (see FLINK-25433). Instead, we would want the
> {{DefaultScheduler}} to return an exceptionally completed future with the
> exception. This enables the {{DefaultResourceCleaner}} to trigger a retry.
> Both scheduler implementations do not expose the error during shutdown of the
> {{CompletedCheckpointStore}} or {{CheckpointIDCounter}} right now. This would
> need to be addressed as well.
--
This message was sent by Atlassian Jira
(v8.20.7#820007)
[jira] [Updated] (FLINK-26114) DefaultScheduler fails fatally in case of an error when shutting down the checkpoint-related resources
[
https://issues.apache.org/jira/browse/FLINK-26114?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Matthias Pohl updated FLINK-26114:
--
Description:
In contrast to the {{AdaptiveScheduler}}, the {{DefaultScheduler}} fails
fatally in case of an error while cleaning up the checkpoint-related resources.
This contradicts our new approach of retrying the cleanup of job-related data
(see FLINK-25433). Instead, we would want the {{DefaultScheduler}} to return an
exceptionally completed future with the exception. This enables the
{{DefaultResourceCleaner}} to trigger a retry.
Both scheduler implementations do not expose the error during shutdown of the
{{CompletedCheckpointStore}} or {{CheckpointIDCounter}} right now. This would
need to be addressed as well.
was:In contrast to the {{AdaptiveScheduler}}, the {{DefaultScheduler}} fails
fatally in case of an error while cleaning up the checkpoint-related resources.
This contradicts our new approach of retrying the cleanup of job-related data
(see FLINK-25433). Instead, we would want the {{DefaultScheduler}} to return an
exceptionally completed future with the exception. This enables the
{{DefaultResourceCleaner}} to trigger a retry.
> DefaultScheduler fails fatally in case of an error when shutting down the
> checkpoint-related resources
> --
>
> Key: FLINK-26114
> URL: https://issues.apache.org/jira/browse/FLINK-26114
> Project: Flink
> Issue Type: Bug
> Components: Runtime / Coordination
>Affects Versions: 1.15.0
>Reporter: Matthias Pohl
>Assignee: Niklas Semmler
>Priority: Critical
>
> In contrast to the {{AdaptiveScheduler}}, the {{DefaultScheduler}} fails
> fatally in case of an error while cleaning up the checkpoint-related
> resources. This contradicts our new approach of retrying the cleanup of
> job-related data (see FLINK-25433). Instead, we would want the
> {{DefaultScheduler}} to return an exceptionally completed future with the
> exception. This enables the {{DefaultResourceCleaner}} to trigger a retry.
> Both scheduler implementations do not expose the error during shutdown of the
> {{CompletedCheckpointStore}} or {{CheckpointIDCounter}} right now. This would
> need to be addressed as well.
--
This message was sent by Atlassian Jira
(v8.20.7#820007)
[jira] [Commented] (FLINK-27174) Non-null check for bootstrapServers field is incorrect in KafkaSink
[
https://issues.apache.org/jira/browse/FLINK-27174?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17526331#comment-17526331
]
Fabian Paul commented on FLINK-27174:
-
I think that is a valid point. We introduced the setBootstrapServers method to
keep in consistent with the KafkaSource. Do you want to work on relaxing the
check that it also allows the bootstrap servers as part of the properties?
> Non-null check for bootstrapServers field is incorrect in KafkaSink
> ---
>
> Key: FLINK-27174
> URL: https://issues.apache.org/jira/browse/FLINK-27174
> Project: Flink
> Issue Type: Bug
> Components: Connectors / Kafka
>Affects Versions: 1.14.4
>Reporter: Zhengqi Zhang
>Priority: Major
> Labels: easyfix
> Attachments: image-2022-04-11-18-11-18-576.png,
> image-2022-04-11-18-17-48-514.png
>
>
> If the user-supplied kafkaProducerConfig contains bootstrapServers
> information, there is no need to define the value of this field separately
> through the setBootstrapServers method. Obviously, the current code doesn't
> notice this.
> !image-2022-04-11-18-11-18-576.png|width=859,height=261!
>
> Perhaps we can check bootstrapServers as follows:
> !image-2022-04-11-18-17-48-514.png|width=861,height=322!
>
> {color:#172b4d}Or check bootstrapServers like KafkaSourceBuilder.{color}
>
>
--
This message was sent by Atlassian Jira
(v8.20.7#820007)
[GitHub] [flink-kubernetes-operator] gyfora commented on pull request #176: [FLINK-27279] Extract common status interfaces
gyfora commented on PR #176: URL: https://github.com/apache/flink-kubernetes-operator/pull/176#issuecomment-1106387142 Thank you, I will check this later today :) -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [flink] afedulov commented on pull request #19405: [FLINK-27066] Reintroduce e2e tests in ES as Java tests.
afedulov commented on PR #19405: URL: https://github.com/apache/flink/pull/19405#issuecomment-1106341694 @flinkbot run azure -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [flink-kubernetes-operator] Aitozi commented on pull request #176: [FLINK-27279] Extract common status interfaces
Aitozi commented on PR #176: URL: https://github.com/apache/flink-kubernetes-operator/pull/176#issuecomment-1106331702 Due to the java type erasure, I can't not completly eliminate the `FlinkDeploymentReconciliationStatus` and `FlinkSessionJobReconciliationStatus` because I can't get the class at runtime . The new interface `ReconcileTarget` is responsible for get the common view of the custom resource. If we have special reconcile update logic for different custom resource, we can extend the interface to make it happen. What do you think about the current implementation @gyfora -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[jira] [Commented] (FLINK-27174) Non-null check for bootstrapServers field is incorrect in KafkaSink
[
https://issues.apache.org/jira/browse/FLINK-27174?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17526313#comment-17526313
]
Zhengqi Zhang commented on FLINK-27174:
---
[~fpaul] , please take a look
> Non-null check for bootstrapServers field is incorrect in KafkaSink
> ---
>
> Key: FLINK-27174
> URL: https://issues.apache.org/jira/browse/FLINK-27174
> Project: Flink
> Issue Type: Bug
> Components: Connectors / Kafka
>Affects Versions: 1.14.4
>Reporter: Zhengqi Zhang
>Priority: Major
> Labels: easyfix
> Attachments: image-2022-04-11-18-11-18-576.png,
> image-2022-04-11-18-17-48-514.png
>
>
> If the user-supplied kafkaProducerConfig contains bootstrapServers
> information, there is no need to define the value of this field separately
> through the setBootstrapServers method. Obviously, the current code doesn't
> notice this.
> !image-2022-04-11-18-11-18-576.png|width=859,height=261!
>
> Perhaps we can check bootstrapServers as follows:
> !image-2022-04-11-18-17-48-514.png|width=861,height=322!
>
> {color:#172b4d}Or check bootstrapServers like KafkaSourceBuilder.{color}
>
>
--
This message was sent by Atlassian Jira
(v8.20.7#820007)
[jira] [Created] (FLINK-27354) JobMaster still processes requests while terminating
Matthias Pohl created FLINK-27354:
-
Summary: JobMaster still processes requests while terminating
Key: FLINK-27354
URL: https://issues.apache.org/jira/browse/FLINK-27354
Project: Flink
Issue Type: Bug
Components: Runtime / Coordination
Affects Versions: 1.14.4, 1.13.6, 1.15.0
Reporter: Matthias Pohl
An issue was reported in the [user
ML|https://lists.apache.org/thread/5pm3crntmb1hl17h4txnlhjz34clghrg] about the
JobMaster trying to reconnect to the ResourceManager during shutdown.
The JobMaster is disconnecting from the ResourceManager during shutdown (see
[JobMaster:1182|https://github.com/apache/flink/blob/da532423487e0534b5fe61f5a02366833f76193a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java#L1182]).
This triggers the deregistration of the job in the {{ResourceManager}}. The RM
responses asynchronously at the end of this deregistration through
{{disconnectResourceManager}} (see
[ResourceManager:993|https://github.com/apache/flink/blob/da532423487e0534b5fe61f5a02366833f76193a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManager.java#L993])
which will trigger a reconnect on the {{JobMaster}}'s side (see
[JobMaster::disconnectResourceManager|https://github.com/apache/flink/blob/da532423487e0534b5fe61f5a02366833f76193a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java#L789])
if it's still around because the {{resourceManagerAddress}} (used in
{{isConnectingToResourceManager}}) is not cleared. This would only happen
during a RM leader change.
The {{disconnectResourceManager}} will be ignored if the {{JobMaster}} is gone
already.
We should add a guard in some way to {{JobMaster}} to avoid reconnecting to
other components during shutdown. This might not only include the
ResourceManager connection but might also affect other parts of the
{{JobMaster}} API.
--
This message was sent by Atlassian Jira
(v8.20.7#820007)
[jira] [Closed] (FLINK-27348) Flink KafkaSource doesn't set groupId
[
https://issues.apache.org/jira/browse/FLINK-27348?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Ahmet Gürbüz closed FLINK-27348.
Resolution: Done
> Flink KafkaSource doesn't set groupId
> -
>
> Key: FLINK-27348
> URL: https://issues.apache.org/jira/browse/FLINK-27348
> Project: Flink
> Issue Type: Bug
> Components: API / Scala
>Affects Versions: 1.14.4
> Environment: OS: windows 8.1.
> Java version:
> java version "11.0.13" 2021-10-19 LTS
> Java(TM) SE Runtime Environment 18.9 (build 11.0.13+10-LTS-370)
> Java HotSpot(TM) 64-Bit Server VM 18.9 (build 11.0.13+10-LTS-370, mixed mode)
>
>
>Reporter: Ahmet Gürbüz
>Priority: Major
> Attachments: image-2022-04-22-05-43-06-475.png,
> image-2022-04-22-05-44-56-494.png, image-2022-04-22-05-46-45-592.png,
> image-2022-04-22-05-52-04-760.png
>
>
> I have one very simple Flink application. I have installed kafka in my local
> and I am reading data from kafka with flink. I am using KafkaSource class in
> Flink. Although I have assigned GroupId with setGroupId, this groupId does
> not appear in Kafka.
>
> {code:java}
> object FlinkKafkaSource extends App {
> val env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI()
> case class Event(partitionNo:Long, eventTime:String, eventTimestamp:Long,
> userId:String, firstName:String)
> implicit val readsEvent: Reads[Event] = Json.reads[Event]
> env
> .fromSource(KafkaSource.builder[Event]
> .setBootstrapServers("localhost:9092")
> .setTopics("flink-connection")
> .setGroupId("test-group") // I can't see this groupId in
> kafka-consumer-groups
> .setStartingOffsets(OffsetsInitializer.latest)
> .setDeserializer(new KafkaRecordDeserializationSchema[Event] {
> override def deserialize(record: ConsumerRecord[Array[Byte],
> Array[Byte]], out: Collector[Event]): Unit = {
> val rec = record.value.map(_.toChar).mkString
> Try(Json.fromJson[Event](Json.parse(rec)).get) match {
> case Success(event) => out.collect(event)
> case Failure(exception) => println(s"Couldn't parse string: $rec,
> error: ${exception.toString}")
> }
> }
> override def getProducedType: TypeInformation[Event] =
> createTypeInformation[Event]
> })
> .build,
> WatermarkStrategy.noWatermarks[Event],
> "kafka-source"
> )
> .keyBy(l => l.userId)
> .print
> env.execute("flink-kafka-source")
> } {code}
> I have created a topic in kafka named "flink-connection".
>
> I am using a simple kafka-python producer to produce data flink-connection
> topic.
> !image-2022-04-22-05-52-04-760.png!
> I am able to consume data from kafka to flink.
> !image-2022-04-22-05-44-56-494.png!
> But can't see the groupId in kafka-consumer-groups
> !image-2022-04-22-05-46-45-592.png!
> Does anyone has any idea why groupid is not setting?
>
--
This message was sent by Atlassian Jira
(v8.20.7#820007)
[jira] [Comment Edited] (FLINK-27348) Flink KafkaSource doesn't set groupId
[
https://issues.apache.org/jira/browse/FLINK-27348?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17526291#comment-17526291
]
Ahmet Gürbüz edited comment on FLINK-27348 at 4/22/22 9:49 AM:
---
Yes i found the necessary properties for that [~Jiangang] ,
{code:java}
.setProperty("enable.auto.commit", "true") {code}
doing that, many thanks ...
was (Author: JIRAUSER288443):
How will i do that with KafkaSource? [~Jiangang]
> Flink KafkaSource doesn't set groupId
> -
>
> Key: FLINK-27348
> URL: https://issues.apache.org/jira/browse/FLINK-27348
> Project: Flink
> Issue Type: Bug
> Components: API / Scala
>Affects Versions: 1.14.4
> Environment: OS: windows 8.1.
> Java version:
> java version "11.0.13" 2021-10-19 LTS
> Java(TM) SE Runtime Environment 18.9 (build 11.0.13+10-LTS-370)
> Java HotSpot(TM) 64-Bit Server VM 18.9 (build 11.0.13+10-LTS-370, mixed mode)
>
>
>Reporter: Ahmet Gürbüz
>Priority: Major
> Attachments: image-2022-04-22-05-43-06-475.png,
> image-2022-04-22-05-44-56-494.png, image-2022-04-22-05-46-45-592.png,
> image-2022-04-22-05-52-04-760.png
>
>
> I have one very simple Flink application. I have installed kafka in my local
> and I am reading data from kafka with flink. I am using KafkaSource class in
> Flink. Although I have assigned GroupId with setGroupId, this groupId does
> not appear in Kafka.
>
> {code:java}
> object FlinkKafkaSource extends App {
> val env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI()
> case class Event(partitionNo:Long, eventTime:String, eventTimestamp:Long,
> userId:String, firstName:String)
> implicit val readsEvent: Reads[Event] = Json.reads[Event]
> env
> .fromSource(KafkaSource.builder[Event]
> .setBootstrapServers("localhost:9092")
> .setTopics("flink-connection")
> .setGroupId("test-group") // I can't see this groupId in
> kafka-consumer-groups
> .setStartingOffsets(OffsetsInitializer.latest)
> .setDeserializer(new KafkaRecordDeserializationSchema[Event] {
> override def deserialize(record: ConsumerRecord[Array[Byte],
> Array[Byte]], out: Collector[Event]): Unit = {
> val rec = record.value.map(_.toChar).mkString
> Try(Json.fromJson[Event](Json.parse(rec)).get) match {
> case Success(event) => out.collect(event)
> case Failure(exception) => println(s"Couldn't parse string: $rec,
> error: ${exception.toString}")
> }
> }
> override def getProducedType: TypeInformation[Event] =
> createTypeInformation[Event]
> })
> .build,
> WatermarkStrategy.noWatermarks[Event],
> "kafka-source"
> )
> .keyBy(l => l.userId)
> .print
> env.execute("flink-kafka-source")
> } {code}
> I have created a topic in kafka named "flink-connection".
>
> I am using a simple kafka-python producer to produce data flink-connection
> topic.
> !image-2022-04-22-05-52-04-760.png!
> I am able to consume data from kafka to flink.
> !image-2022-04-22-05-44-56-494.png!
> But can't see the groupId in kafka-consumer-groups
> !image-2022-04-22-05-46-45-592.png!
> Does anyone has any idea why groupid is not setting?
>
--
This message was sent by Atlassian Jira
(v8.20.7#820007)
[jira] [Commented] (FLINK-27155) Reduce multiple reads to the same Changelog file in the same taskmanager during restore
[ https://issues.apache.org/jira/browse/FLINK-27155?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17526298#comment-17526298 ] Roman Khachatryan commented on FLINK-27155: --- I think Yun Tang means waiting for all task of a TM to switch from INITIALIZING to RUNNING. I think it could be problematic, because the set of tasks is dynamic. Besides that, because the local space can be limited, and changelog can be large, it's better to clean up the cache earlier. Probably, we should combine reference counting and timeouts. periodic-materialize.interval can still be less than the recovery time. As for serializing file accesses, I'm afraid that without it the ticket can introduce regression. Furthermore, the need to serialize can affect the design of cache. So I'd at least design this at once. > Reduce multiple reads to the same Changelog file in the same taskmanager > during restore > --- > > Key: FLINK-27155 > URL: https://issues.apache.org/jira/browse/FLINK-27155 > Project: Flink > Issue Type: Sub-task > Components: Runtime / State Backends >Reporter: Feifan Wang >Assignee: Feifan Wang >Priority: Major > Fix For: 1.16.0 > > > h3. Background > In the current implementation, State changes of different operators in the > same taskmanager may be written to the same changelog file, which effectively > reduces the number of files and requests to DFS. > But on the other hand, the current implementation also reads the same > changelog file multiple times on recovery. More specifically, the number of > times the same changelog file is accessed is related to the number of > ChangeSets contained in it. And since each read needs to skip the preceding > bytes, this network traffic is also wasted. > The result is a lot of unnecessary request to DFS when there are multiple > slots and keyed state in the same taskmanager. > h3. Proposal > We can reduce multiple reads to the same changelog file in the same > taskmanager during restore. > One possible approach is to read the changelog file all at once and cache it > in memory or local file for a period of time when reading the changelog file. > I think this could be a subtask of [v2 FLIP-158: Generalized incremental > checkpoints|https://issues.apache.org/jira/browse/FLINK-25842] . > Hi [~ym] , [~roman] how do you think about ? -- This message was sent by Atlassian Jira (v8.20.7#820007)
[jira] [Updated] (FLINK-27353) Update training exercises to use Flink 1.15
[ https://issues.apache.org/jira/browse/FLINK-27353?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] ASF GitHub Bot updated FLINK-27353: --- Labels: pull-request-available (was: ) > Update training exercises to use Flink 1.15 > --- > > Key: FLINK-27353 > URL: https://issues.apache.org/jira/browse/FLINK-27353 > Project: Flink > Issue Type: New Feature > Components: Documentation / Training / Exercises >Reporter: Nico Kruber >Assignee: Nico Kruber >Priority: Major > Labels: pull-request-available > Fix For: 1.15.0 > > -- This message was sent by Atlassian Jira (v8.20.7#820007)
[GitHub] [flink-training] NicoK opened a new pull request, #48: [FLINK-27353] Update to Flink 1.15
NicoK opened a new pull request, #48: URL: https://github.com/apache/flink-training/pull/48 This PR will only go through CI once 1.15.0 is released. Until then, you can apply the following change to `build.gradle` to see it in action: ``` flinkVersion = '1.15-SNAPSHOT' ``` -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[jira] [Created] (FLINK-27353) Update training exercises to use Flink 1.15
Nico Kruber created FLINK-27353: --- Summary: Update training exercises to use Flink 1.15 Key: FLINK-27353 URL: https://issues.apache.org/jira/browse/FLINK-27353 Project: Flink Issue Type: New Feature Components: Documentation / Training / Exercises Reporter: Nico Kruber Assignee: Nico Kruber Fix For: 1.15.0 -- This message was sent by Atlassian Jira (v8.20.7#820007)
[jira] [Closed] (FLINK-11694) Enforce space before left curly brace
[
https://issues.apache.org/jira/browse/FLINK-11694?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Chesnay Schepler closed FLINK-11694.
Resolution: Fixed
Should be covered by spotless.
> Enforce space before left curly brace
> -
>
> Key: FLINK-11694
> URL: https://issues.apache.org/jira/browse/FLINK-11694
> Project: Flink
> Issue Type: Improvement
> Components: Build System
>Reporter: Chesnay Schepler
>Priority: Not a Priority
> Labels: auto-deprioritized-major, auto-deprioritized-minor
> Time Spent: 10m
> Remaining Estimate: 0h
>
> We currently don't enforce spaces before left curly brace (\{), as a result
> of which the following code block would not raise any error:
> {code}
> void someMethod(){
> if (someCondition){
> try {
> ...
> } catch (Exception e){
> ...
> }
> }
> }
> {code}
> It is not the aim of this JIRA to forbid {} as a shorthand for empty bodies,
> such as:
> {code}
> new TypeHint>(){}
> {code}
--
This message was sent by Atlassian Jira
(v8.20.7#820007)
[jira] [Commented] (FLINK-27348) Flink KafkaSource doesn't set groupId
[
https://issues.apache.org/jira/browse/FLINK-27348?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17526291#comment-17526291
]
Ahmet Gürbüz commented on FLINK-27348:
--
How will i do that with KafkaSource? [~Jiangang]
> Flink KafkaSource doesn't set groupId
> -
>
> Key: FLINK-27348
> URL: https://issues.apache.org/jira/browse/FLINK-27348
> Project: Flink
> Issue Type: Bug
> Components: API / Scala
>Affects Versions: 1.14.4
> Environment: OS: windows 8.1.
> Java version:
> java version "11.0.13" 2021-10-19 LTS
> Java(TM) SE Runtime Environment 18.9 (build 11.0.13+10-LTS-370)
> Java HotSpot(TM) 64-Bit Server VM 18.9 (build 11.0.13+10-LTS-370, mixed mode)
>
>
>Reporter: Ahmet Gürbüz
>Priority: Major
> Attachments: image-2022-04-22-05-43-06-475.png,
> image-2022-04-22-05-44-56-494.png, image-2022-04-22-05-46-45-592.png,
> image-2022-04-22-05-52-04-760.png
>
>
> I have one very simple Flink application. I have installed kafka in my local
> and I am reading data from kafka with flink. I am using KafkaSource class in
> Flink. Although I have assigned GroupId with setGroupId, this groupId does
> not appear in Kafka.
>
> {code:java}
> object FlinkKafkaSource extends App {
> val env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI()
> case class Event(partitionNo:Long, eventTime:String, eventTimestamp:Long,
> userId:String, firstName:String)
> implicit val readsEvent: Reads[Event] = Json.reads[Event]
> env
> .fromSource(KafkaSource.builder[Event]
> .setBootstrapServers("localhost:9092")
> .setTopics("flink-connection")
> .setGroupId("test-group") // I can't see this groupId in
> kafka-consumer-groups
> .setStartingOffsets(OffsetsInitializer.latest)
> .setDeserializer(new KafkaRecordDeserializationSchema[Event] {
> override def deserialize(record: ConsumerRecord[Array[Byte],
> Array[Byte]], out: Collector[Event]): Unit = {
> val rec = record.value.map(_.toChar).mkString
> Try(Json.fromJson[Event](Json.parse(rec)).get) match {
> case Success(event) => out.collect(event)
> case Failure(exception) => println(s"Couldn't parse string: $rec,
> error: ${exception.toString}")
> }
> }
> override def getProducedType: TypeInformation[Event] =
> createTypeInformation[Event]
> })
> .build,
> WatermarkStrategy.noWatermarks[Event],
> "kafka-source"
> )
> .keyBy(l => l.userId)
> .print
> env.execute("flink-kafka-source")
> } {code}
> I have created a topic in kafka named "flink-connection".
>
> I am using a simple kafka-python producer to produce data flink-connection
> topic.
> !image-2022-04-22-05-52-04-760.png!
> I am able to consume data from kafka to flink.
> !image-2022-04-22-05-44-56-494.png!
> But can't see the groupId in kafka-consumer-groups
> !image-2022-04-22-05-46-45-592.png!
> Does anyone has any idea why groupid is not setting?
>
--
This message was sent by Atlassian Jira
(v8.20.7#820007)
[jira] [Assigned] (FLINK-26268) Add classfication algorithm support for LogisticRegression, KNN and NaiveBayes in ML Python API
[ https://issues.apache.org/jira/browse/FLINK-26268?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Huang Xingbo reassigned FLINK-26268: Assignee: Huang Xingbo > Add classfication algorithm support for LogisticRegression, KNN and > NaiveBayes in ML Python API > --- > > Key: FLINK-26268 > URL: https://issues.apache.org/jira/browse/FLINK-26268 > Project: Flink > Issue Type: Sub-task > Components: API / Python, Library / Machine Learning >Affects Versions: ml-2.1.0 >Reporter: Huang Xingbo >Assignee: Huang Xingbo >Priority: Major > Labels: pull-request-available > Fix For: ml-2.1.0 > > -- This message was sent by Atlassian Jira (v8.20.7#820007)
[jira] [Closed] (FLINK-26268) Add classfication algorithm support for LogisticRegression, KNN and NaiveBayes in ML Python API
[ https://issues.apache.org/jira/browse/FLINK-26268?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Huang Xingbo closed FLINK-26268. Resolution: Done Merged into master via aedf66335ff3475e449ebeee795d43372c9fa703 > Add classfication algorithm support for LogisticRegression, KNN and > NaiveBayes in ML Python API > --- > > Key: FLINK-26268 > URL: https://issues.apache.org/jira/browse/FLINK-26268 > Project: Flink > Issue Type: Sub-task > Components: API / Python, Library / Machine Learning >Affects Versions: ml-2.1.0 >Reporter: Huang Xingbo >Priority: Major > Labels: pull-request-available > Fix For: ml-2.1.0 > > -- This message was sent by Atlassian Jira (v8.20.7#820007)
[jira] [Updated] (FLINK-26268) Add classfication algorithm support for LogisticRegression, KNN and NaiveBayes in ML Python API
[ https://issues.apache.org/jira/browse/FLINK-26268?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] ASF GitHub Bot updated FLINK-26268: --- Labels: pull-request-available (was: ) > Add classfication algorithm support for LogisticRegression, KNN and > NaiveBayes in ML Python API > --- > > Key: FLINK-26268 > URL: https://issues.apache.org/jira/browse/FLINK-26268 > Project: Flink > Issue Type: Sub-task > Components: API / Python, Library / Machine Learning >Affects Versions: ml-2.1.0 >Reporter: Huang Xingbo >Priority: Major > Labels: pull-request-available > Fix For: ml-2.1.0 > > -- This message was sent by Atlassian Jira (v8.20.7#820007)
[GitHub] [flink-ml] HuangXingBo closed pull request #88: [FLINK-26268][ml][python] Add classfication algorithm support for LogisticRegression, KNN and NaiveBayes in ML Python API
HuangXingBo closed pull request #88: [FLINK-26268][ml][python] Add classfication algorithm support for LogisticRegression, KNN and NaiveBayes in ML Python API URL: https://github.com/apache/flink-ml/pull/88 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[jira] [Assigned] (FLINK-24144) Improve DataGenerator to prevent excessive creation of new Random objects
[
https://issues.apache.org/jira/browse/FLINK-24144?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Nico Kruber reassigned FLINK-24144:
---
Assignee: (was: Nico Kruber)
> Improve DataGenerator to prevent excessive creation of new Random objects
> -
>
> Key: FLINK-24144
> URL: https://issues.apache.org/jira/browse/FLINK-24144
> Project: Flink
> Issue Type: Improvement
> Components: Documentation / Training / Exercises
>Affects Versions: 1.14.0, 1.13.2
>Reporter: Nico Kruber
>Priority: Not a Priority
>
> For a couple of methods in {{DataGenerator}}, new {{Random}} objects are
> created with a specific seed. Instead, we could create a single {{Random}}
> object and reset the seed when needed.
--
This message was sent by Atlassian Jira
(v8.20.7#820007)
[jira] [Updated] (FLINK-24144) Improve DataGenerator to prevent excessive creation of new Random objects
[
https://issues.apache.org/jira/browse/FLINK-24144?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Nico Kruber updated FLINK-24144:
Priority: Not a Priority (was: Major)
> Improve DataGenerator to prevent excessive creation of new Random objects
> -
>
> Key: FLINK-24144
> URL: https://issues.apache.org/jira/browse/FLINK-24144
> Project: Flink
> Issue Type: Improvement
> Components: Documentation / Training / Exercises
>Affects Versions: 1.14.0, 1.13.2
>Reporter: Nico Kruber
>Assignee: Nico Kruber
>Priority: Not a Priority
>
> For a couple of methods in {{DataGenerator}}, new {{Random}} objects are
> created with a specific seed. Instead, we could create a single {{Random}}
> object and reset the seed when needed.
--
This message was sent by Atlassian Jira
(v8.20.7#820007)
[jira] [Resolved] (FLINK-26382) Add Chinese documents for flink-training exercises
[ https://issues.apache.org/jira/browse/FLINK-26382?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Nico Kruber resolved FLINK-26382. - Fix Version/s: 1.15.0 1.14.5 (was: 1.14.3) Assignee: tonny Resolution: Fixed Fixed on - release-1.14: 0132dd7be8c881607f9a374613309493ade8c6dd, 18e6db2206ca4156e21276b14d35bebaf222c151 - master: aca6c47b79d486eb38969492c7e2dc8cb200d146 > Add Chinese documents for flink-training exercises > -- > > Key: FLINK-26382 > URL: https://issues.apache.org/jira/browse/FLINK-26382 > Project: Flink > Issue Type: New Feature > Components: Documentation / Training / Exercises >Affects Versions: 1.14.3 >Reporter: tonny >Assignee: tonny >Priority: Major > Labels: pull-request-available > Fix For: 1.15.0, 1.14.5 > > > Provide Chinese documents for all `README` and `DISCUSSION` accompanied by > Chinese documents of Flink -- This message was sent by Atlassian Jira (v8.20.7#820007)
[jira] [Commented] (FLINK-27215) JDBC sink transiently deleted a record because of -u message of that record
[
https://issues.apache.org/jira/browse/FLINK-27215?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17526283#comment-17526283
]
tim yu commented on FLINK-27215:
Hi [~fsk119], I have seen some classes that like
FlinkChangelogModeInferenceProgram, StreamPhysicalDropUpdateBefore and
StreamExecDropUpdateBefore. Thank you.
> JDBC sink transiently deleted a record because of -u message of that record
> ---
>
> Key: FLINK-27215
> URL: https://issues.apache.org/jira/browse/FLINK-27215
> Project: Flink
> Issue Type: Bug
> Components: Connectors / JDBC
>Affects Versions: 1.12.7, 1.13.5, 1.14.3
>Reporter: tim yu
>Priority: Major
>
> A record is deleted transiently when using JDBC sink in upsert mode.
> The -U message is processed as delete operation in class
> TableBufferReducedStatementExecutor.
> The following codes show how to process -U message:
> {code:java}
> /**
> * Returns true if the row kind is INSERT or UPDATE_AFTER, returns false
> if the row kind is
> * DELETE or UPDATE_BEFORE.
> */
> private boolean changeFlag(RowKind rowKind) {
> switch (rowKind) {
> case INSERT:
> case UPDATE_AFTER:
> return true;
> case DELETE:
> case UPDATE_BEFORE:
> return false;
> default:
> throw new UnsupportedOperationException(
> String.format(
> "Unknown row kind, the supported row kinds
> is: INSERT, UPDATE_BEFORE, UPDATE_AFTER,"
> + " DELETE, but get: %s.",
> rowKind));
> }
> }
> @Override
> public void executeBatch() throws SQLException {
> for (Map.Entry> entry :
> reduceBuffer.entrySet()) {
> if (entry.getValue().f0) {
> upsertExecutor.addToBatch(entry.getValue().f1);
> } else {
> // delete by key
> deleteExecutor.addToBatch(entry.getKey());
> }
> }
> upsertExecutor.executeBatch();
> deleteExecutor.executeBatch();
> reduceBuffer.clear();
> }
> {code}
> If -U and +U messages of one record are executed separately in different JDBC
> batches, that record will be deleted transiently in external database and
> then insert a new updated record to it. In fact, this record should be merely
> updated once in the external database.
>
--
This message was sent by Atlassian Jira
(v8.20.7#820007)
