Re: Best approach to aggregate state with idle timeout and periodic output

hi, we have a similar use case, and from my experience it is not simple (if at all) to implement your logic when using a session window. Eventually we used the state + timers approach and we have full control of the cleanup and merge logic. Thanks Sigalit On Mon, May 12, 2025 at 1:35 PM Sachin

Re: Task manager fails to start in HA mode flink 1.19.1

uster-config-map&; *resourceVersion*=0 should be retried after 100 millis because of IOException thanks again On Thu, Jan 23, 2025 at 1:04 PM Sigalit Eliazov wrote: > I saw there was an upgrade in flink 1.19 to 6.9.2 where in flink 1.18 > 6.6.2 was used. > when running the same jar wit

Re: Task manager fails to start in HA mode flink 1.19.1

I saw there was an upgrade in flink 1.19 to 6.9.2 where in flink 1.18 6.6.2 was used. when running the same jar with flink 18 it works ok. Was there any additional configuration required for version 19 due to this upgrade? thanks again On Wed, Jan 22, 2025 at 3:24 PM Sigalit Eliazov wrote

Task manager fails to start in HA mode flink 1.19.1

Hi, I recently upgraded to *Flink 1.19.1* and am using the *Flink Kubernetes Operator 1.9* to deploy the Flink cluster. The checkpoints are defined using PersistentVolumeClaims (PVCs), and the service account is configured with the necessary permissions. However, when starting the pipeline in *HA

Clarification on Flink Pipeline and Cluster Termination with Operator v1.9

Hi, I am currently deploying a Flink pipeline using Flink Kubernetes Operator v1.9 alongside Flink version 1.19.1 as part of a comprehensive use case. When the use case is undeployed, I need to ensure that the Flink pipeline is properly canceled and the Flink cluster is taken down. My approach in

flink operator - restart a pipeline from a manually trigger savepoint

hi, We are trying to restart a pipeline from a save point we triggered manually via the job manager rest api. with the following configuration in the flinkdeployment crd: savepointTriggerNonce: 1 initialSavepointPath: upgradeMode: savepoint this always fails with the following error org.apach

flink pipeline canary upgrade question

Hi, We are exploring options to support canary upgrades for our Flink pipelines and have considered the following approach. 1. *Stateless Pipelines*: For stateless pipelines, we define two clusters, job-v1 and job-v2, using the same consumer group to avoid event duplication. To contr

custom metric reporter

hi we have upgraded from flink 1.16 to 1.18 and our custom metric stopped working. i saw in the release note of 1.17 that there was a change so i have defined the following metrics.reporters: otlp metrics.reporter.otlp.factory.class: xxx.flink.metrics.otlp.OpenTelemetryProtocolReporterFactory metr

Access to S3 - checkpoints

Hi, We are using Ceph buckets to store the checkpoints and savepoints, and the access is done via the S3 protocol. Since we don't have any integration with Hadoop, we added a dependency on flink-s3-fs-presto. Our Flink configuration looks like this: state.checkpoint-storage: filesystemstate.che

Re: Task Manager memory usage

t; On Thu, 23 May 2024 at 6:19 PM, Sigalit Eliazov > wrote: > >> Hi, >> >> I am trying to understand the following behavior in our Flink application >> cluster. Any assistance would be appreciated. >> >> We are running a Flink application c

Task Manager memory usage

Hi, I am trying to understand the following behavior in our Flink application cluster. Any assistance would be appreciated. We are running a Flink application cluster with 5 task managers, each with the following configuration: - jobManagerMemory: 12g - taskManagerMemory: 20g - taskMana

Re: [NOTICE] Switch docker image base to Eclipse Temurin

Hi all, We pulled the new image and we are facing an issue to start the job manager pod. we are using version 1.14.5-java11 and the cluster is started using flink operator the error is [ERROR] Could not get JVM parameters and dynamic configurations properly. [ERROR] Raw output from BashJavaUtil

Re: start flink-operator using "watchNamespaces"

wing namespaces: [JOSDK_ALL_NAMESPACES] > This means that the operator is not configured correctly for the watched > namespaces. After you change the helm values you might have to reinstall the > operator. > > Cheers, > Gyula > > > On Mon, Aug 22, 2022 at 5:16 PM Si

start flink-operator using "watchNamespaces"

Hi, we are trying to start a flink operator with a specific namespace. flink-operator version: 1.1.0 we set the following parameter : watchNamespaces: ["my-namespace"] The operator fails during installation with the following error *Caused by: io.fabric8.kubernetes.client.KubernetesClientExcept

Re: Flink Operator - delete flinkdeployments

hello we upgraded to version 1.1.0 and i am afraid the problem exists in that version as well. I would appreciate any additional ideas or guidelines on how to do the cleanup correctly. thanks Sigalit On Tue, Aug 2, 2022 at 3:39 PM Sigalit Eliazov wrote: > Will do, thanks! > > On T

Re: Flink Operator - delete flinkdeployments

Will do, thanks! On Tue, Aug 2, 2022 at 3:39 PM Gyula Fóra wrote: > Before trying to solve any already fixed problems please upgrade to 1.1.0 > :) > > > > On Tue, Aug 2, 2022 at 2:33 PM Sigalit Eliazov > wrote: > >> we are working with 1.0.0 >> >>

Re: Flink Operator - delete flinkdeployments

we are working with 1.0.0 On Tue, Aug 2, 2022 at 3:24 PM Gyula Fóra wrote: > Are you running the latest 1.1.0 version of the operator? > > Gyula > > On Tue, Aug 2, 2022 at 2:18 PM Sigalit Eliazov > wrote: > >> Hi, >> >> We are deploying a few flink cl

Flink Operator - delete flinkdeployments

Hi, We are deploying a few flink clusters via the flink operator in our CI. In each run we first do a clean-up where one of the first steps is to run 'kubectl delete flinkdeployments --all -n ' after that we also delete the flink operator pod and our all namespace. Lately we face issues where t

Re: Flink Kubernetes Operator with K8S + Istio + mTLS - port definitions

Hi, we have enabled HA as suggested, the task manager tries to reach the job manager via pod id as expected but the task manager is unable to connect to the job manager: 2022-06-19 22:14:45,101 INFO org.apache.flink.runtime.taskexecutor.TaskExecutor [] - Connecting to ResourceManager akk

Flink operator deletes the FlinkDeplyoment after a while

after few hours of running job manager and task manager generated using the operator i get the following message in the operator log There really wasn't any traffic and the flink deployment is being delete === Finished metrics report === Deleting Fli

exception while activating calculateThroughp

Hi all We are using the flink k8s operator latest version with flink 1.14 in order to deploy our pipelines in application mode (one job per cluster, one job manager + one task manager) Once in a few minutes I receive the following error in the job manager and all the tasks are being restarted.

multiple pipeline deployment using flink k8s operator

Hi all, we just started using the flink k8s operator to deploy our flink cluster. >From what we understand we are only able to start a flink cluster per job. So in our case when we have 2 jobs we have to create 2 different clusters. obviously we would prefer to deploy these 2 job which relate to th

sharing data between 2 pipelines

Hi all i have 2 pipelines: A. receives information from kafka and "holds" that info B. a pipeline which is triggered by a scheduler and every x minutes should send the info i received in pipeline A to another kafka topic As i understood i cannot use the flink state for this since these are differe

flink pipeline handles very small amount of messages in a second (only 500)

hi all I would appreciate some help to understand the pipeline behaviour... We deployed a standalone flink cluster. The pipelines are deployed via the jm rest api. We have 5 task managers with 1 slot each. In total i am deploying 5 pipelines which mainly read from kafka, a simple object conversi

Re: Task manager errors with Flink ZooKeeper High Availability

Thanks for the response on this issue. with the same configuration defined *high-availability: zookeeper* *high-availability.zookeeper.quorum: zk-noa-edge-infra:2181* *high-availability.zookeeper.path.root: /flink* *high-availability.storageDir: /flink_state* *high-availabi

Flink metrics via permethous or opentelemerty

Hello. I am looking for a way to expose flink metrics via opentelemerty to the gcp could monitoring dashboard. Does anyone has experience with that? If it is not directly possible we thought about using permethous as a middlewere. If you have experience with that i would appreciate any guidance.

Re: Fwd: How to get memory specific metrics for tasknodes

Hello. Related metric issue. I am looking for a way to expose flink metrics via opentelemerty to the gcp could monitoring dashboard. Does anyone has experience with that? Thanks בתאריך יום ו׳, 18 בפבר׳ 2022, 21:55, מאת Chesnay Schepler ‏< ches...@apache.org>: > As I said, this is not possible. I

pipeline are not started sporadically

Hello We have 5 different pipelines running on standalone flink cluster. 2 – integrated with DB using JDBC 2 – integrated with GCP – writes to big query 1 – reads from kafka writes to kafka We are running with 1 job manager 5 task managers – 2 slots on each Our problem is that only 4

Re: select records using JDBC with parameters

do not match. SubtaskStateMapper.DISCARD_EXTRA_STATE is removed > since Flink 1.14 so please make sure that your Flink client version is also > 1.14. > > Sigalit Eliazov 于2021年11月10日周三 上午5:46写道： > >> Hello >> >> i am creating new pipeline which >> >>1. r

select records using JDBC with parameters

Hello i am creating new pipeline which 1. receives info from kafka (mainly the key) 2. with this key select information from a D 3. writes to kafka the results Flink is running has a standalone cluster I am failing on the pipeline deployment when activating step 2 with the following

Flink production configuration

Hello New to flink and I am struggling with defining the correct configuration. Currently we decided we have only one job manger. We have 5 different pipelines. We are not sure how many task managers needs to be defined, how many slots. Can you please send reference or is there any calculator that