Re: confirm subscribe to user@flink.apache.org
b1NsQZU9Bc0p2ap/+LMG/VXlEtQ7GMf80DfxiQgBoGgs0rpclCFWn0nk23lU7p7Xgvq67aYV71OWW1ZzilI0/PRxQPpk1go+7px82YlBMe5ujw8Z6owk/1cU4359/bVjKVYJKACiPg8CQLuWvDj6JLfIEPjHDVAEb6NYF86iCHiAj7RNRpA== X-SONIC-DKIM-SIGN: v=1; a=rsa-sha256; c=relaxed/relaxed; d=yahoo.com; s=s2048; t=1626924125; bh=Jp5jaP1tt2DyGF6+aoNQDoCsd5Me/3VMlbYVaaFIjD5=; h=X-Sonic-MF:Date:From:To:Subject:From:Subject; b=kpDB3/OZeaq1QtkihINJ1zvI1BjIFDav+st9g2d/2h4Rt8WyZR0YOzJWk45LKDhu1APTg0/Ai/orraZkA9YDx6dKswZnqpI/mbyhxlV1/3EOPdLxj31YXIx8iGVWA7AfdDsI9sgknvKvAHN4H645n94Mx5NWqhOGTvd3f2zS2yHr1FzoFzlNdDlM0/zZW9lGS9YMUkFNGzug+fYzRgDmB3Cw3BDr7TID3OoASoUvjr2Qr9eCkya9Jr7ZjbnZZ7JaLixKLnKBhnR4hckAcXpwgeUgwdHgysM3yqCqV3rByf970pXv3SB6KfzSs0FCJzZAkyV9L938gGfxQ3hoRYKtZQ== X-YMail-OSG: ob9eYU4VM1k3B6JeaVLTIDQ_6AfcF5jSQUqt7HvVTJrUmIWeqoyWpunVxeUf7e6 dpVR1cqcca2DZVG.TpQd0UkuB7mOv1AXV4JhO4rOf4TWfTbQmdGr_6bfRgvf2ekqu57ZD.dCj6EE CfRoVoY5Skk04MCSOzeyOY5WWfNuXl2V67WL_Oq48.Bc2kZL8y.DWCUznabyiVVn_jiEqYJr6Rg9 ikSF.BUJjJbFy6QR5etzRLsU9VallqYoVXemWxmjpk5NpIixQcKmc6WUH1G4mblRmdMdT_Q9f2rx _UltuU8lGUAvzSW0j0WNkbqRlt_x.rQNwF156Ke3ZzelXM6eQB9wzdTdeNmQutvzvSwK.AsNLPCD Va9UJ5ecVv2gIjWin28fOFQyIOv2QCxE5G2vG6PaHDZcAigcCa5NjFTk4m_.2T_EGixMrq8Nk4Tj PgQBG1kV2IgKQHR1AnupHekdedi5.y4fUmpiWJWf0Rc2jZQ3gSB64y8xVTJIVJsML4M2Oj6aVRfZ YP..LGLKyyL0jiewtwXMtXUCWbQVOyCLY3mtN5hROTG64PcWoc0dIYpYps1ZaHtAhcj6z3kI_g5t yVNnZ92C7eg7odl4HqfHhPda0PQ3dii57FWoXnk2e0YHZxs3AIzlmy7S._ZgdPX5iVZPeD9L24TB Wqiu3mYQ75H_sNRFYugGqL74w6yrj5NJrsrbd4g49tKYHSx6f68x1rVlHzzSCFvbt.Fgm7jq_yB2 LDb0u_vjFpbUw9_v2UnxenmnAdtrjhpNVyV.Pd3FJ8HzbKkbGz1siyq_bt03U0Pys4eSKsh5mSXt lpBKMC6o7VynjSb4N02kpcs5dYUXFSaN5AKVw6zzBRwchyTcNKJnhBJe2LUguCP1T2Cr0WrcoXEl uUfcVcDEzZ0K_r5QaOBAu4TfbA_1lA8pEcFvT76l0JwbKacbGZywE.Xwl4KrTP6PmUHy2hYJnQKx NT7KFd8BUVbNPz7PVlYZJvccvDkRZtvCvdVeVmxIl9aoQDzzN55uDGihYvygURQ__V8bdzb0Vy5k OcE4jU0LI9P9NwH6YwJjeJR9xD_IxkTspGeX2lQyS8dMJ54vwyHeGNh4zfJIfnYZLPVlsyjc_nbr LSCAHi5B2c93i5gtTeBtTP9t7gbnuh7DQlZcZHcTVs5uNAyxMB7tg4r.SFRawD_m_Gg3v2sWBWpf rMCVOITFjWd42sBFCrsqI53wvWkAX7hu9DJ39PT79OiphtaaIZuGb2dhb0wb2GYe2wkCn1k9E32J yozBlJWEUSHkhLm3peTc3CjTPcxLsRUPs2efHYBGlf00ZX_aZqWsnwJr3CTO7xiXvtXwBgL_Zpcf MDRHBHprsWyuBcm.ezhrPKF.n_O3tdgq7bFditXWtORNEzwB3ETNHChdHd17EPgk8HtLxYfSCoWA X4aLft8ZUK.sa9ctP7X0GHB29cr23hul9rJ8xyFq1kHA4pULqFo2tQIJVfMSDr0mtraculkPQFh6 qRGdWdzMFzc2_zTxAh0riAcDdR7kJWj.DM25MIT1DJYQERbXxJCY_r45wkvUukLjykqWaZPfEHd. TbdWM13IPNk7f1XNAAosyjanrWifQZe0f8RW9Vi368cyz2lyQvcChOAp6Qg7Z6C8D16U.kAgy9ET 9HXvvtbHguUzIVniW4FOdylyT7mhsjXUkjTrmpACfyDN8I4tupjlK.wnU5yxfXLD1H1mV0L_XZnu guMz3sjbHjpRd7Fo3u6dZq0IkaCAF0lK8LQZRwXnO8YaoI8Qdf3bmw3pv2TPVllWItFwUNEvHLwo Aau48CoWRad4br4RaYLDq9oReSxa3BuU1DCrZduRnjrp43Vmh.g2gSPUURlTIjkQ8Oi0cYqg.up8 k7spGOOnrPWnq3GIWdVFcGZClM8sZ0nvVxuaCBBrN__ojGQZG7tueYo_m4LESfy_mP3gXsT5eFio M.8WZZSXgCI_OWz0x7bSQ0X023Z_fRjXi8hFIcjmK8RvNf.nx6IA2XDocREH9SwnFKlVqRYp6JRa xq.yIclzi5o0u6s1CWiBt7cTKYhSA65HDKDeePYO1dQgeR8YxIYT3uigt8lUM9DBv.S6bnZ99Qaw iTdwAubk3jbLp2v3I9ndtn1qZoXd9JTyEIe.WA14GzAS.qZTlejDLFKSIKPCA6ski1HcKVjEqVdl x6TxeU4Tp.EBkBXZCV3zIW7WBeEd11YeNdkseY36BxaOVsj3gf.mPfyLy6q5lRPzfw1hW1f6acEh VS3Z_upYfVCWGVjZZRcC064SJTD_487pAaG4CvXtaUDiRFiSbG3auOeVuixKFTpcYPpUSYCxq2im Aj0odzraqWgbDCBGqBVNW8OMFQkGZL1bbFyG.2JnTPOKiCuZwZ3BwGrHy8kZKGf5WlJl7ZHonGqE YV.S7aXHuMLfRQhK8MePknjjs5NKjC8gTRTkg6qJpviDTIWE82YAE4q153z5ZVB.SlKap5uNKX_f BHzDg6cfjOy_KJn.LWyeJdPiP31xE9qVe6DsqAlbk2n43HzdFjfFYwOqvFeGYODOPGg-- X-Sonic-MF: Received: from sonic.gate.mail.ne1.yahoo.com by sonic309.consmr.mail.bf2.yahoo.com with HTTP; Thu, 22 Jul 2021 03:22:05 + Date: Thu, 22 Jul 2021 03:22:03 + (UTC) From: Dhiru To: "user-subscr...@flink.apache.org" Message-ID: <362567283.31384.1626924123...@mail.yahoo.com> Subject: subscribe to flink mailing list MIME-Version: 1.0 Content-Type: multipart/alternative; boundary="=_Part_31383_951932538.1626924123565" References: <362567283.31384.1626924123566@mail.yahoo.com> X-Mailer: WebService/1.1.18736 YMailNorrin
Need help of deploying Flink HA on kubernetes cluster
hi , I am very new to flink , I am planning to install Flink HA setup on eks cluster with 5 worker nodes . Please can some one point me to right materials or direction how to install as well as any sample job which I can run only for testing and confirm all things are working as expected . --Dhirendra
Flink installation using flinkoperator
hello all , I read some article , I think many company using flink operator is using separate cluster for each job and this can be achieved using flinkk8soperator ? Please can you help me sharing some pointer video /git link which can help me for installing on AWS- EKS and I have zookeeper/kafka cluster all-ready running . How Can I use both of them ? --kumar
submit new job is not working
hello all , I was able to run sample example and was able to upload jar using UI, cluster which I have deployed on k8s Today I had to reboot jobmanager after that I am not able to upload any jar to my cluster. Do not see any log as well to debug any help --kumar
Re: submit new job is not working
thanks @chesney its later On Friday, August 20, 2021, 03:05:26 AM AST, Chesnay Schepler wrote: Is the problem that previously uploaded jars are no longer available (which would be expected behavior), or that you cannot upload new jars? If it is the latter, could you use the developer tools of you browser to check what response the UI receives when attempting to upload the jar? On 20/08/2021 07:55, Dhiru wrote: hello all , I was able to run sample example and was able to upload jar using UI, cluster which I have deployed on k8s Today I had to reboot jobmanager after that I am not able to upload any jar to my cluster. Do not see any log as well to debug any help --kumar
aws s3 configuring error for flink image
Need to configure aws S3 getting this error org.apache.flink.core.fs.UnsupportedFileSystemSchemeException: Could not find a file system implementation for scheme 's3'. The scheme is directly supported by Flink through the following plugins: flink-s3-fs-hadoop, flink-s3-fs-presto. Please ensure that each plugin resides within its own subfolder within the plugins directory. See https://ci.apache.org/projects/flink/flink-docs-stable/ops/plugins.html for more information. If you want to use a Hadoop file system for that scheme, please add the scheme to the configuration fs.allowed-fallback-filesystems. For a full list of supported file systems, please see https://ci.apache.org/projects/flink/flink-docs-stable/ops/filesystems/.at org.apache.flink.core.fs.FileSystem.getUnguardedFileSystem(FileSystem.java:473) at org.apache.flink.core.fs.FileSystem.get(FileSystem.java:389)at org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSink$RowFormatBuilder.createBuckets(StreamingFileSink.java:260) at org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSink.initializeState(StreamingFileSink.java:396) at org.apache.flink.streaming.util.functions.StreamingFunctionUtils.tryRestoreFunction(StreamingFunctionUtils.java:185) at org.apache.flink.streaming.util.functions.StreamingFunctionUtils.restoreFunctionState(StreamingFunctionUtils.java:167) at org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator.initializeState(AbstractUdfStreamOperator.java:96) at org.apache.flink.streaming.api.operators.StreamOperatorStateHandler.initializeOperatorState(StreamOperatorStateHandler.java:106) at org.apache.flink.streaming.api.operators.AbstractStreamOperator.initializeState(AbstractStreamOperator.java:260) at org.apache.flink.streaming.runtime.tasks.OperatorChain.initializeStateAndOpenOperators(OperatorChain.java:290) at org.apache.flink.streaming.runtime.tasks.StreamTask.lambda$beforeInvoke$1(StreamTask.java:506) at org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$SynchronizedStreamTaskActionExecutor.runThrowing(StreamTaskActionExecutor.java:92) at org.apache.flink.streaming.runtime.tasks.StreamTask.beforeInvoke(StreamTask.java:475) at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:526) at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:721)at org.apache.flink.runtime.taskmanager.Task.run(Task.java:546)at java.base/java.lang.Thread.run(Unknown Source) image used flink:1.11.3-scala_2.12-java11 some of the part which I added for flink-conf.yaml # REQUIRED: set storage location for job metadata in remote storage high-availability.storageDir: s3://msc-sandbox-flink/test state.backend: filesystem state.checkpoints.dir: s3://msc-sandbox-flink/test state.backend.fs.checkpointdir: s3://msc-sandbox-flink/test s3.path.style.access: true I am trying to deploy jobmanager and taskmanager using eks , please let me know if I need to do any more support for adding s3
Re: aws s3 configuring error for flink image
yes I copied to plugin folder but not sure same jar I see in /opt as well by default root@d852f125da1f:/opt/flink/plugins# lsREADME.txt flink-s3-fs-hadoop-1.13.1.jar metrics-datadog metrics-influx metrics-prometheus metrics-statsdexternal-resource-gpu flink-s3-fs-presto-1.13.1.jar metrics-graphite metrics-jmx metrics-slf4j I need help sooner on this On Wednesday, September 8, 2021, 09:26:46 AM EDT, Dhiru wrote: yes I copied to plugin folder but not sure same jar I see in /opt as well by default root@d852f125da1f:/opt/flink/plugins# lsREADME.txt flink-s3-fs-hadoop-1.13.1.jar metrics-datadog metrics-influx metrics-prometheus metrics-statsdexternal-resource-gpu flink-s3-fs-presto-1.13.1.jar metrics-graphite metrics-jmx metrics-slf4j On Wednesday, September 8, 2021, 02:58:38 AM EDT, Martijn Visser wrote: Hi, Have you copied the correct JAR [1] to the plugins directory? Best regards, Martijn [1] https://ci.apache.org/projects/flink/flink-docs-release-1.11/ops/filesystems/s3.html On Wed, 8 Sept 2021 at 04:27, Dhiru wrote: Need to configure aws S3 getting this error org.apache.flink.core.fs.UnsupportedFileSystemSchemeException: Could not find a file system implementation for scheme 's3'. The scheme is directly supported by Flink through the following plugins: flink-s3-fs-hadoop, flink-s3-fs-presto. Please ensure that each plugin resides within its own subfolder within the plugins directory. See https://ci.apache.org/projects/flink/flink-docs-stable/ops/plugins.html for more information. If you want to use a Hadoop file system for that scheme, please add the scheme to the configuration fs.allowed-fallback-filesystems. For a full list of supported file systems, please see https://ci.apache.org/projects/flink/flink-docs-stable/ops/filesystems/.at org.apache.flink.core.fs.FileSystem.getUnguardedFileSystem(FileSystem.java:473) at org.apache.flink.core.fs.FileSystem.get(FileSystem.java:389)at org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSink$RowFormatBuilder.createBuckets(StreamingFileSink.java:260) at org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSink.initializeState(StreamingFileSink.java:396) at org.apache.flink.streaming.util.functions.StreamingFunctionUtils.tryRestoreFunction(StreamingFunctionUtils.java:185) at org.apache.flink.streaming.util.functions.StreamingFunctionUtils.restoreFunctionState(StreamingFunctionUtils.java:167) at org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator.initializeState(AbstractUdfStreamOperator.java:96) at org.apache.flink.streaming.api.operators.StreamOperatorStateHandler.initializeOperatorState(StreamOperatorStateHandler.java:106) at org.apache.flink.streaming.api.operators.AbstractStreamOperator.initializeState(AbstractStreamOperator.java:260) at org.apache.flink.streaming.runtime.tasks.OperatorChain.initializeStateAndOpenOperators(OperatorChain.java:290) at org.apache.flink.streaming.runtime.tasks.StreamTask.lambda$beforeInvoke$1(StreamTask.java:506) at org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$SynchronizedStreamTaskActionExecutor.runThrowing(StreamTaskActionExecutor.java:92) at org.apache.flink.streaming.runtime.tasks.StreamTask.beforeInvoke(StreamTask.java:475) at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:526) at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:721)at org.apache.flink.runtime.taskmanager.Task.run(Task.java:546)at java.base/java.lang.Thread.run(Unknown Source) image used flink:1.11.3-scala_2.12-java11 some of the part which I added for flink-conf.yaml # REQUIRED: set storage location for job metadata in remote storage high-availability.storageDir: s3://msc-sandbox-flink/test state.backend: filesystem state.checkpoints.dir: s3://msc-sandbox-flink/test state.backend.fs.checkpointdir: s3://msc-sandbox-flink/test s3.path.style.access: true I am trying to deploy jobmanager and taskmanager using eks , please let me know if I need to do any more support for adding s3
Re: aws s3 configuring error for flink image
I copied FROM flink:1.11.3-scala_2.12-java11 RUN mkdir ./plugins/flink-s3-fs-presto RUN cp ./opt/flink-s3-fs-presto-1.11.3.jar ./plugins/flink-s3-fs-presto/ then started getting this error , trying to run on aws eks and trying to access s3 bucket 2021-09-08 14:38:10java.lang.UnsupportedOperationException: This s3 file system implementation does not support recoverable writers.at org.apache.flink.fs.s3.common.FlinkS3FileSystem.createRecoverableWriter(FlinkS3FileSystem.java:136) at org.apache.flink.core.fs.PluginFileSystemFactory$ClassLoaderFixingFileSystem.createRecoverableWriter(PluginFileSystemFactory.java:134) at org.apache.flink.core.fs.SafetyNetWrapperFileSystem.createRecoverableWriter(SafetyNetWrapperFileSystem.java:69) at org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSink$RowFormatBuilder.createBuckets(StreamingFileSink.java:260) at org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSink.initializeState(StreamingFileSink.java:396) at org.apache.flink.streaming.util.functions.StreamingFunctionUtils.tryRestoreFunction(StreamingFunctionUtils.java:185) at org.apache.flink.streaming.util.functions.StreamingFunctionUtils.restoreFunctionState(StreamingFunctionUtils.java:167) at org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator.initializeState(AbstractUdfStreamOperator.java:96) at org.apache.flink.streaming.api.operators.StreamOperatorStateHandler.initializeOperatorState(StreamOperatorStateHandler.java:106) at org.apache.flink.streaming.api.operators.AbstractStreamOperator.initializeState(AbstractStreamOperator.java:260) at org.apache.flink.streaming.runtime.tasks.OperatorChain.initializeStateAndOpenOperators(OperatorChain.java:290) at org.apache.flink.streaming.runtime.tasks.StreamTask.lambda$beforeInvoke$1(StreamTask.java:506) at org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$SynchronizedStreamTaskActionExecutor.runThrowing(StreamTaskActionExecutor.java:92) at org.apache.flink.streaming.runtime.tasks.StreamTask.beforeInvoke(StreamTask.java:475) at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:526) at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:721)at org.apache.flink.runtime.taskmanager.Task.run(Task.java:546)at java.base/java.lang.Thread.run(Unknown Source) On Wednesday, September 8, 2021, 12:47:10 PM EDT, Chesnay Schepler wrote: you need to put the flink-s3-fs-hadoop/presto jar into a directory within the plugins directory, for example the final path should look like this: /opt/flink/plugins/flink-s3-fs-hadoop/flink-s3-fs-hadoop-1.13.1.jar Furthermore, you only need either the hadoop or presto jar, _not_ both of them. See also: https://nightlies.apache.org/flink/flink-docs-release-1.13/docs/deployment/filesystems/s3/#hadooppresto-s3-file-systems-plugins https://nightlies.apache.org/flink/flink-docs-release-1.13/docs/deployment/filesystems/plugins/ On 08/09/2021 17:10, Dhiru wrote: yes I copied to plugin folder but not sure same jar I see in /opt as well by default root@d852f125da1f:/opt/flink/plugins# ls README.txt flink-s3-fs-hadoop-1.13.1.jar metrics-datadog metrics-influx metrics-prometheus metrics-statsd external-resource-gpu flink-s3-fs-presto-1.13.1.jar metrics-graphite metrics-jmx metrics-slf4j I need help sooner on this On Wednesday, September 8, 2021, 09:26:46 AM EDT, Dhiru wrote: yes I copied to plugin folder but not sure same jar I see in /opt as well by default root@d852f125da1f:/opt/flink/plugins# ls README.txt flink-s3-fs-hadoop-1.13.1.jar metrics-datadog metrics-influx metrics-prometheus metrics-statsd external-resource-gpu flink-s3-fs-presto-1.13.1.jar metrics-graphite metrics-jmx metrics-slf4j On Wednesday, September 8, 2021, 02:58:38 AM EDT, Martijn Visser wrote: Hi, Have you copied the correct JAR [1] to the plugins directory? Best regards, Martijn [1] https://ci.apache.org/projects/flink/flink-docs-release-1.11/ops/filesystems/s3.html On Wed, 8 Sept 2021 at 04:27, Dhiru wrote: Need to configure aws S3 getting this error org.apache.flink.core.fs.UnsupportedFileSystemSchemeException: Could not find a file system implementation for scheme 's3'. The scheme is directly supported by Flink through the following plugins: flink-s3-fs-hadoop, flink-s3-fs-presto. Please ensure that each plugin resides within its own subfolder within the plugins directory. See https://ci.apache.org/projects/flink/flink-docs-stable/ops/plugins.html for more information. If you want to use a Hadoop file system for that scheme, please add the scheme to the configuration fs.allowed-fallback-filesystems. For a full list of supported file systems, please see https://ci.apache.org/projects/f
Error while fetching data from Apache Kafka
I am not sure when we try to receive data from Apache Kafka I get this error , but works good for me when I try to run via Conflunece kafka java.lang.ClassCastException: class java.lang.String cannot be cast to class scala.Product (java.lang.String is in module java.base of loader 'bootstrap'; scala.Product is in unnamed module of loader 'app') at org.apache.flink.api.scala.typeutils.CaseClassSerializer.copy(CaseClassSerializer.scala:32) at org.apache.flink.api.scala.typeutils.CaseClassSerializer.copy(CaseClassSerializer.scala:96) at org.apache.flink.api.scala.typeutils.CaseClassSerializer.copy(CaseClassSerializer.scala:32) at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.pushToOperator(OperatorChain.java:715) at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.collect(OperatorChain.java:692) at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.collect(OperatorChain.java:672) at org.apache.flink.streaming.api.operators.CountingOutput.collect(CountingOutput.java:52) at org.apache.flink.streaming.api.operators.CountingOutput.collect(CountingOutput.java:30) at org.apache.flink.streaming.api.operators.StreamSourceContexts$ManualWatermarkContext.processAndCollectWithTimestamp(StreamSourceContexts.java:310) at org.apache.flink.streaming.api.operators.StreamSourceContexts$WatermarkContext.collectWithTimestamp(StreamSourceContexts.java:409) at org.apache.flink.streaming.connectors.kafka.internals.AbstractFetcher.emitRecordsWithTimestamps(AbstractFetcher.java:352) at org.apache.flink.streaming.connectors.kafka.internal.KafkaFetcher.partitionConsumerRecordsHandler(KafkaFetcher.java:185) at org.apache.flink.streaming.connectors.kafka.internal.KafkaFetcher.runFetchLoop(KafkaFetcher.java:141) at org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumerBase.run(FlinkKafkaConsumerBase.java:755) at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:100) at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:63) at org.apache.flink.streaming.runtime.tasks.SourceStreamTask$LegacySourceFunctionThread.run(SourceStreamTask.java:213)
Re: Error while fetching data from Apache Kafka
Thanks @caizhi , let me look this On Monday, September 13, 2021, 10:24:03 PM EDT, Caizhi Weng wrote: Hi! This seems to be caused by some mismatching types in your source definition and your workflow. If possible could you describe the schema of your Kafka source and paste your datastream / Table / SQL code here? Dhiru 于2021年9月14日周二 上午3:49写道： I am not sure when we try to receive data from Apache Kafka I get this error , but works good for me when I try to run via Conflunece kafka java.lang.ClassCastException: class java.lang.String cannot be cast to class scala.Product (java.lang.String is in module java.base of loader 'bootstrap'; scala.Product is in unnamed module of loader 'app') at org.apache.flink.api.scala.typeutils.CaseClassSerializer.copy(CaseClassSerializer.scala:32) at org.apache.flink.api.scala.typeutils.CaseClassSerializer.copy(CaseClassSerializer.scala:96) at org.apache.flink.api.scala.typeutils.CaseClassSerializer.copy(CaseClassSerializer.scala:32) at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.pushToOperator(OperatorChain.java:715) at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.collect(OperatorChain.java:692) at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.collect(OperatorChain.java:672) at org.apache.flink.streaming.api.operators.CountingOutput.collect(CountingOutput.java:52) at org.apache.flink.streaming.api.operators.CountingOutput.collect(CountingOutput.java:30) at org.apache.flink.streaming.api.operators.StreamSourceContexts$ManualWatermarkContext.processAndCollectWithTimestamp(StreamSourceContexts.java:310) at org.apache.flink.streaming.api.operators.StreamSourceContexts$WatermarkContext.collectWithTimestamp(StreamSourceContexts.java:409) at org.apache.flink.streaming.connectors.kafka.internals.AbstractFetcher.emitRecordsWithTimestamps(AbstractFetcher.java:352) at org.apache.flink.streaming.connectors.kafka.internal.KafkaFetcher.partitionConsumerRecordsHandler(KafkaFetcher.java:185) at org.apache.flink.streaming.connectors.kafka.internal.KafkaFetcher.runFetchLoop(KafkaFetcher.java:141) at org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumerBase.run(FlinkKafkaConsumerBase.java:755) at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:100) at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:63) at org.apache.flink.streaming.runtime.tasks.SourceStreamTask$LegacySourceFunctionThread.run(SourceStreamTask.java:213)
S3 access permission error
flink image I have added both s3 plugin FROM flink:1.11.3-scala_2.12-java11RUN mkdir ./plugins/flink-s3-fs-prestoRUN cp ./opt/flink-s3-fs-presto-1.11.3.jar ./plugins/flink-s3-fs-presto/RUN mkdir ./plugins/flink-s3-fs-hadoopRUN cp ./opt/flink-s3-fs-hadoop-1.11.3.jar ./plugins/flink-s3-fs-hadoop/ some part of flink-conf.yaml ( I tried with both s3a and s3 ) # REQUIRED: set storage location for job metadata in remote storage state.backend: filesystem state.backend.fs.checkpointdir: s3a://msc-actigraph-test-bucket/flink-checkpointing/checkpoints state.checkpoints.dir: s3a://msc-actigraph-test-bucket/flink-checkpointing/externalized-checkpoints state.savepoints.dir: s3a://msc-actigraph-test-bucket/flink-checkpointing/savepoints high-availability.storageDir: s3a://msc-actigraph-test-bucket/flink-checkpointing/storagedir s3.path.style.access: true org.apache.flink.runtime.rest.handler.RestHandlerException: Could not execute application. at org.apache.flink.runtime.webmonitor.handlers.JarRunHandler.lambda$handleRequest$1(JarRunHandler.java:103) at java.base/java.util.concurrent.CompletableFuture.uniHandle(Unknown Source) at java.base/java.util.concurrent.CompletableFuture$UniHandle.tryFire(Unknown Source) at java.base/java.util.concurrent.CompletableFuture.postComplete(Unknown Source) at java.base/java.util.concurrent.CompletableFuture$AsyncSupply.run(Unknown Source) at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Unknown Source) at java.base/java.util.concurrent.FutureTask.run(Unknown Source) at java.base/java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(Unknown Source) 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.util.concurrent.CompletionException: org.apache.flink.util.FlinkRuntimeException: Could not execute application. at java.base/java.util.concurrent.CompletableFuture.encodeThrowable(Unknown Source) at java.base/java.util.concurrent.CompletableFuture.completeThrowable(Unknown Source) ... 7 more Caused by: org.apache.flink.util.FlinkRuntimeException: Could not execute application. at org.apache.flink.client.deployment.application.DetachedApplicationRunner.tryExecuteJobs(DetachedApplicationRunner.java:81) at org.apache.flink.client.deployment.application.DetachedApplicationRunner.run(DetachedApplicationRunner.java:67) at org.apache.flink.runtime.webmonitor.handlers.JarRunHandler.lambda$handleRequest$0(JarRunHandler.java:100) ... 7 more Caused by: org.apache.flink.client.program.ProgramInvocationException: The main method caused an error: Failed to execute job 'DeduplicationJob'. at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:302) at org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:198) at org.apache.flink.client.ClientUtils.executeProgram(ClientUtils.java:149) at org.apache.flink.client.deployment.application.DetachedApplicationRunner.tryExecuteJobs(DetachedApplicationRunner.java:78) ... 9 more Caused by: org.apache.flink.util.FlinkException: Failed to execute job 'DeduplicationJob'. at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.executeAsync(StreamExecutionEnvironment.java:1829) at org.apache.flink.client.program.StreamContextEnvironment.executeAsync(StreamContextEnvironment.java:128) at org.apache.flink.client.program.StreamContextEnvironment.execute(StreamContextEnvironment.java:76) at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.execute(StreamExecutionEnvironment.java:1700) at org.apache.flink.streaming.api.scala.StreamExecutionEnvironment.execute(StreamExecutionEnvironment.scala:699) at io.epiphanous.flinkrunner.flink.BaseFlinkJob.run(BaseFlinkJob.scala:45) at io.epiphanous.flinkrunner.FlinkRunner.process1(FlinkRunner.scala:56) at io.epiphanous.flinkrunner.FlinkRunner.process(FlinkRunner.scala:33) at com.mdsol.flink.delivery_streams.Runner$.run(Runner.scala:25) at com.mdsol.flink.delivery_streams.Runner$.main(Runner.scala:7) at com.mdsol.flink.delivery_streams.Runner.main(Runner.scala) at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method) at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(Unknown Source) at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(Unknown Source) at java.base/java.lang.reflect.Method.invoke(Unknown Source) at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:288) ... 12 more Caused by: org.apache.flink.runtime.client.JobSubmissionException: Failed to submit job. at org.apache.flink.runtime.dispatcher.Dispatcher.lambda$internalSubmitJob$3(Dispatcher.java:362) at java.base/java.util.concurrent.CompletableFuture.uniHandle(Unk
Re: S3 access permission error
i see org.apache.hadoop.fs.FileSystem.mkdirs(FileSystem.java:2326) plugin is not able to create folder , not sure if I need to change something Whereas when We are trying to pass from the local laptop and passing aws credentails its able to create a folder and running as expected On Wednesday, September 22, 2021, 01:39:04 AM EDT, Dhiru wrote: flink image I have added both s3 plugin FROM flink:1.11.3-scala_2.12-java11RUN mkdir ./plugins/flink-s3-fs-prestoRUN cp ./opt/flink-s3-fs-presto-1.11.3.jar ./plugins/flink-s3-fs-presto/RUN mkdir ./plugins/flink-s3-fs-hadoopRUN cp ./opt/flink-s3-fs-hadoop-1.11.3.jar ./plugins/flink-s3-fs-hadoop/ some part of flink-conf.yaml ( I tried with both s3a and s3 ) # REQUIRED: set storage location for job metadata in remote storage state.backend: filesystem state.backend.fs.checkpointdir: s3a://msc-actigraph-test-bucket/flink-checkpointing/checkpoints state.checkpoints.dir: s3a://msc-actigraph-test-bucket/flink-checkpointing/externalized-checkpoints state.savepoints.dir: s3a://msc-actigraph-test-bucket/flink-checkpointing/savepoints high-availability.storageDir: s3a://msc-actigraph-test-bucket/flink-checkpointing/storagedir s3.path.style.access: true org.apache.flink.runtime.rest.handler.RestHandlerException: Could not execute application. at org.apache.flink.runtime.webmonitor.handlers.JarRunHandler.lambda$handleRequest$1(JarRunHandler.java:103) at java.base/java.util.concurrent.CompletableFuture.uniHandle(Unknown Source) at java.base/java.util.concurrent.CompletableFuture$UniHandle.tryFire(Unknown Source) at java.base/java.util.concurrent.CompletableFuture.postComplete(Unknown Source) at java.base/java.util.concurrent.CompletableFuture$AsyncSupply.run(Unknown Source) at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Unknown Source) at java.base/java.util.concurrent.FutureTask.run(Unknown Source) at java.base/java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(Unknown Source) 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.util.concurrent.CompletionException: org.apache.flink.util.FlinkRuntimeException: Could not execute application. at java.base/java.util.concurrent.CompletableFuture.encodeThrowable(Unknown Source) at java.base/java.util.concurrent.CompletableFuture.completeThrowable(Unknown Source) ... 7 more Caused by: org.apache.flink.util.FlinkRuntimeException: Could not execute application. at org.apache.flink.client.deployment.application.DetachedApplicationRunner.tryExecuteJobs(DetachedApplicationRunner.java:81) at org.apache.flink.client.deployment.application.DetachedApplicationRunner.run(DetachedApplicationRunner.java:67) at org.apache.flink.runtime.webmonitor.handlers.JarRunHandler.lambda$handleRequest$0(JarRunHandler.java:100) ... 7 more Caused by: org.apache.flink.client.program.ProgramInvocationException: The main method caused an error: Failed to execute job 'DeduplicationJob'. at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:302) at org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:198) at org.apache.flink.client.ClientUtils.executeProgram(ClientUtils.java:149) at org.apache.flink.client.deployment.application.DetachedApplicationRunner.tryExecuteJobs(DetachedApplicationRunner.java:78) ... 9 more Caused by: org.apache.flink.util.FlinkException: Failed to execute job 'DeduplicationJob'. at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.executeAsync(StreamExecutionEnvironment.java:1829) at org.apache.flink.client.program.StreamContextEnvironment.executeAsync(StreamContextEnvironment.java:128) at org.apache.flink.client.program.StreamContextEnvironment.execute(StreamContextEnvironment.java:76) at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.execute(StreamExecutionEnvironment.java:1700) at org.apache.flink.streaming.api.scala.StreamExecutionEnvironment.execute(StreamExecutionEnvironment.scala:699) at io.epiphanous.flinkrunner.flink.BaseFlinkJob.run(BaseFlinkJob.scala:45) at io.epiphanous.flinkrunner.FlinkRunner.process1(FlinkRunner.scala:56) at io.epiphanous.flinkrunner.FlinkRunner.process(FlinkRunner.scala:33) at com.mdsol.flink.delivery_streams.Runner$.run(Runner.scala:25) at com.mdsol.flink.delivery_streams.Runner$.main(Runner.scala:7) at com.mdsol.flink.delivery_streams.Runner.main(Runner.scala) at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method) at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(Unknown Source) at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(Unknown Source) at java.base/java.lang.reflect.M
Re: S3 access permission error
Not sure @yangze ... but other services which are deployed in same places we are able to access s3 bucket, the link which you share are recommended way, if we have access to s3 then we should not pass credentials ? On Wednesday, September 22, 2021, 02:59:05 AM EDT, Yangze Guo wrote: You might need to configure the access credential. [1] [1] https://ci.apache.org/projects/flink/flink-docs-master/docs/deployment/filesystems/s3/#configure-access-credentials Best, Yangze Guo On Wed, Sep 22, 2021 at 2:17 PM Dhiru wrote: > > > i see org.apache.hadoop.fs.FileSystem.mkdirs(FileSystem.java:2326) plugin is > not able to create folder , not sure if I need to change something > Whereas when We are trying to pass from the local laptop and passing aws > credentails its able to create a folder and running as expected > On Wednesday, September 22, 2021, 01:39:04 AM EDT, Dhiru > wrote: > > > flink image I have added both s3 plugin > FROM flink:1.11.3-scala_2.12-java11 > RUN mkdir ./plugins/flink-s3-fs-presto > RUN cp ./opt/flink-s3-fs-presto-1.11.3.jar ./plugins/flink-s3-fs-presto/ > RUN mkdir ./plugins/flink-s3-fs-hadoop > RUN cp ./opt/flink-s3-fs-hadoop-1.11.3.jar ./plugins/flink-s3-fs-hadoop/ > > some part of flink-conf.yaml ( I tried with both s3a and s3 ) > # REQUIRED: set storage location for job metadata in remote storage > state.backend: filesystem > state.backend.fs.checkpointdir: >s3a://msc-actigraph-test-bucket/flink-checkpointing/checkpoints > state.checkpoints.dir: >s3a://msc-actigraph-test-bucket/flink-checkpointing/externalized-checkpoints > state.savepoints.dir: >s3a://msc-actigraph-test-bucket/flink-checkpointing/savepoints > high-availability.storageDir: >s3a://msc-actigraph-test-bucket/flink-checkpointing/storagedir > s3.path.style.access: true > > org.apache.flink.runtime.rest.handler.RestHandlerException: Could not execute > application. at > org.apache.flink.runtime.webmonitor.handlers.JarRunHandler.lambda$handleRequest$1(JarRunHandler.java:103) > at java.base/java.util.concurrent.CompletableFuture.uniHandle(Unknown > Source) at > java.base/java.util.concurrent.CompletableFuture$UniHandle.tryFire(Unknown > Source) at > java.base/java.util.concurrent.CompletableFuture.postComplete(Unknown Source) > at java.base/java.util.concurrent.CompletableFuture$AsyncSupply.run(Unknown > Source) at > java.base/java.util.concurrent.Executors$RunnableAdapter.call(Unknown Source) > at java.base/java.util.concurrent.FutureTask.run(Unknown Source) at > java.base/java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(Unknown > Source) 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.util.concurrent.CompletionException: > org.apache.flink.util.FlinkRuntimeException: Could not execute application. > at java.base/java.util.concurrent.CompletableFuture.encodeThrowable(Unknown > Source) at > java.base/java.util.concurrent.CompletableFuture.completeThrowable(Unknown > Source) ... 7 more Caused by: org.apache.flink.util.FlinkRuntimeException: > Could not execute application. at > org.apache.flink.client.deployment.application.DetachedApplicationRunner.tryExecuteJobs(DetachedApplicationRunner.java:81) > at > org.apache.flink.client.deployment.application.DetachedApplicationRunner.run(DetachedApplicationRunner.java:67) > at > org.apache.flink.runtime.webmonitor.handlers.JarRunHandler.lambda$handleRequest$0(JarRunHandler.java:100) > ... 7 more Caused by: > org.apache.flink.client.program.ProgramInvocationException: The main method > caused an error: Failed to execute job 'DeduplicationJob'. at > org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:302) > at > org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:198) > at org.apache.flink.client.ClientUtils.executeProgram(ClientUtils.java:149) > at > org.apache.flink.client.deployment.application.DetachedApplicationRunner.tryExecuteJobs(DetachedApplicationRunner.java:78) > ... 9 more Caused by: org.apache.flink.util.FlinkException: Failed to > execute job 'DeduplicationJob'. at > org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.executeAsync(StreamExecutionEnvironment.java:1829) > at > org.apache.flink.client.program.StreamContextEnvironment.executeAsync(StreamContextEnvironment.java:128) > at > org.apache.flink.client.program.StreamContextEnvironment.execute(StreamContextEnvironment.java:76) >
how to run job and make jobmanager HA
spec: replicas: 1 selector: matchLabels: app: flink component: jobmanager template: metadata: labels: app: flink component: jobmanager spec: serviceAccountName: msc-s3-shared-content containers: - name: jobmanager image: test:latest ( flink:1.11.3-scala_2.12-java11 image has DeliveryStreams-0.0.1_3.1.0.jar jar copied to ./bin/flink) args: ["jobmanager"] command: ['./bin/flink', 'run', './bin/DeliveryStreams-0.0.1_3.1.0.jar', 'DeduplicationJob'] . ( I am planning to run job ... Please let me know if this is right way) ports: - containerPort: 6123 name: rpc - containerPort: 6124 name: blob-server - containerPort: 8081 name: webui env: - name: JOB_MANAGER_RPC_ADDRESS value: flink-jobmanager - name: KAFKA_BROKERS value: kafka:29092 livenessProbe: tcpSocket: port: 6123 initialDelaySeconds: 30 periodSeconds: 60 volumeMounts: - name: flink-config-volume mountPath: /opt/flink/conf securityContext: runAsUser: # refers to user _flink_ from official flink image, change if necessary volumes: - name: flink-config-volume configMap: name: flink-config items: - key: flink-conf.yaml path: flink-conf.yaml - key: log4j-console.properties path: log4j-console.properties"msc-jobmanager-deployment.yaml" 54L, 1640C a) Do not want to run my job from UI , want to run my job from jobmanager image during boot time , when I am trying to run its throwing me error b) How to make sure my jobManager and taskmanager is HA configured ( so that if jobmanager goes off do not loose data) Thanks Kumar
Re: how to run job and make jobmanager HA
please let me know if anyone can help me on this On Friday, September 24, 2021, 01:45:39 PM EDT, Dhiru wrote: spec: replicas: 1 selector: matchLabels: app: flink component: jobmanager template: metadata: labels: app: flink component: jobmanager spec: serviceAccountName: msc-s3-shared-content containers: - name: jobmanager image: test:latest ( flink:1.11.3-scala_2.12-java11 image has DeliveryStreams-0.0.1_3.1.0.jar jar copied to ./bin/flink) args: ["jobmanager"] command: ['./bin/flink', 'run', './bin/DeliveryStreams-0.0.1_3.1.0.jar', 'DeduplicationJob'] . ( I am planning to run job ... Please let me know if this is right way) ports: - containerPort: 6123 name: rpc - containerPort: 6124 name: blob-server - containerPort: 8081 name: webui env: - name: JOB_MANAGER_RPC_ADDRESS value: flink-jobmanager - name: KAFKA_BROKERS value: kafka:29092 livenessProbe: tcpSocket: port: 6123 initialDelaySeconds: 30 periodSeconds: 60 volumeMounts: - name: flink-config-volume mountPath: /opt/flink/conf securityContext: runAsUser: # refers to user _flink_ from official flink image, change if necessary volumes: - name: flink-config-volume configMap: name: flink-config items: - key: flink-conf.yaml path: flink-conf.yaml - key: log4j-console.properties path: log4j-console.properties"msc-jobmanager-deployment.yaml" 54L, 1640C a) Do not want to run my job from UI , want to run my job from jobmanager image during boot time , when I am trying to run its throwing me error b) How to make sure my jobManager and taskmanager is HA configured ( so that if jobmanager goes off do not loose data) Thanks Kumar
Re: Potential bug when assuming roles from AWS EKS when using S3 as RocksDb checkpoint backend?
We need to overwrite using WebIdentityTokenFileCredentialsProviderhttps://github.com/aws/aws-sdk-java-v2/issues/1470#issuecomment-543601232. otherwise java takes presidency to secret key and access keys than SA On Saturday, September 25, 2021, 04:37:22 PM EDT, Xiangyu Su wrote: Hi Thomas,did you try to login to EKS node and run some aws command like : aws s3 ls ?It sounds like EKS issue, but not 100% sure.Best On Sat, 25 Sept 2021 at 22:12, Ingo Bürk wrote: Hi Thomas, I think you might be looking for this: https://github.com/apache/flink/pull/16717 BestIngo On Sat, Sep 25, 2021, 20:46 Thomas Wang wrote: Hi, I'm using the official docker image: apache/flink:1.12.1-scala_2.11-java11 I'm trying to run a Flink job on an EKS cluster. The job is running under a k8s service account that is tied to an IAM role. If I'm not using s3 as RocksDB checkpoint backend, everything works just fine. However, when I enabled s3 as RocksDB checkpoint backend, I got permission denied. The IAM role tied to the service account has the appropriate permissions to s3. However the underlying role tied to the EKS node doesn't. After debugging with AWS support, it looks like the request to s3 was made under the EKS node role, not the role tied to the service account. Thus the permission denial. With the same Flink application, I'm also making requests to AWS Secrets Manager to get some sensitive information and those requests were made explicitly with AWS Java SDK 2.x bundled in the same application Jar file. Those requests were made correctly with the IAM role tied to the service account. Based on the info above, I suspect Flink may be using an older version of the AWS SDK that doesn't support assuming an IAM role via an IODC web identity token file. Please see AWS doc here: https://docs.aws.amazon.com/eks/latest/userguide/iam-roles-for-service-accounts-minimum-sdk.html Could someone help me confirm this bug and maybe have it fixed some time? Thanks. Thomas -- Xiangyu Su Java Developer xian...@smaato.com Smaato Inc. San Francisco - New York - Hamburg - Singapore www.smaato.com Germany: Barcastraße 5 22087 Hamburg Germany M 0049(176)43330282 The information contained in this communication may be CONFIDENTIAL and is intended only for the use of the recipient(s) named above. If you are not the intended recipient, you are hereby notified that any dissemination, distribution, or copying of this communication, or any of its contents, is strictly prohibited. If you have received this communication in error, please notify the sender and delete/destroy the original message and any copy of it from your computer or paper files.
HA setup Flink version flink:1.13.2-scala_2.12-java11
I am following this link for setting up HA configuration ZooKeeper HA Services | | | | ZooKeeper HA Services ZooKeeper HA Services # Flink’s ZooKeeper HA services use ZooKeeper for high availability services. Flink lever... | | | zookeeper version which I am using is 3.4.10 high-availability.storageDir: s3a:// kubernetes.cluster-id: cluster1337 high-availability: zookeeper high-availability.zookeeper.quorum: zookeeper:2181 high-availability.zookeeper.path.root: /flink logs I see zookeeper has this error org.apache.flink.shaded.curator4.org.apache.curator.ConnectionState [] - Authentication failed tried searching online, but didn't any solution, Please let me know if anyone has fixed this issue
Re: HA setup Flink version flink:1.13.2-scala_2.12-java11
Thanks I got this working On Wednesday, September 29, 2021, 12:12:17 AM EDT, Dhiru wrote: I am following this link for setting up HA configuration ZooKeeper HA Services | | | | ZooKeeper HA Services ZooKeeper HA Services # Flink’s ZooKeeper HA services use ZooKeeper for high availability services. Flink lever... | | | zookeeper version which I am using is 3.4.10 high-availability.storageDir: s3a:// kubernetes.cluster-id: cluster1337 high-availability: zookeeper high-availability.zookeeper.quorum: zookeeper:2181 high-availability.zookeeper.path.root: /flink logs I see zookeeper has this error org.apache.flink.shaded.curator4.org.apache.curator.ConnectionState [] - Authentication failed tried searching online, but didn't any solution, Please let me know if anyone has fixed this issue
Flink application mode with no ui , how to start job using k8s ?
Hi , My requirement is to create Flink cluster application Mode on k8s and do not want to expose UI, my requirement is to start the long-running job which can be instantiated at boot time of flink and keep running use these resource files from jobmanager-application-ha.yaml and taskmanager-job-deployment.yaml for creating cluster (https://ci.apache.org/projects/flink/flink-docs-master/docs/deployment/resource-providers/standalone/kubernetes/#application-cluster-resource-definitions) a) I need to start job during run time , I can bundle my jar with Flink image, so that can instantiate jar b) Can I apply HPA (horizontal pod autoscalar) for task manager, will this work so that according to workload instance of taskmanager goes up and down. --kumar
Re: Flink application mode with no ui , how to start job using k8s ?
Thanks Dawid, if I am not exposing UI , how I am going to run my job we need to submit jar someway, I do not want my flink image tightly coupled with my jar On Monday, October 4, 2021, 09:52:31 AM EDT, Dawid Wysakowicz wrote: Hi Dhiru, For the question about auto scaling I'd recommend you this[1] blogpost from my colleague. I believe he explains it quite well how to do it. Besides that I am not sure what is your other question. Are you asking how to start the jobmanager without the UI? Can't you just simply not expose the port? Best, Dawid [1] https://flink.apache.org/2021/05/06/reactive-mode.html On 30/09/2021 02:41, Dhiru wrote: Hi , My requirement is to create Flink cluster application Mode on k8s and do not want to expose UI, my requirement is to start the long-running job which can be instantiated at boot time of flink and keep running use these resource files from jobmanager-application-ha.yaml and taskmanager-job-deployment.yaml for creating cluster (https://ci.apache.org/projects/flink/flink-docs-master/docs/deployment/resource-providers/standalone/kubernetes/#application-cluster-resource-definitions) a) I need to start job during run time , I can bundle my jar with Flink image, so that can instantiate jar b) Can I apply HPA (horizontal pod autoscalar) for task manager, will this work so that according to workload instance of taskmanager goes up and down. --kumar
k8s not able to submit job from jobmanager
My DockerFile FROM flink:1.13.2-scala_2.12-java11 RUN mkdir -p /opt/flink/plugins/flink-s3-fs-hadoopRUN ln -fs /opt/flink/opt/flink-s3-fs-hadoop-*.jar /opt/flink/plugins/flink-s3-fs-hadoop/. RUN mkdir -p /opt/flink/plugins/flink-s3-fs-prestoRUN ln -fs /opt/flink/opt/flink-s3-fs-presto-*.jar /opt/flink/plugins/flink-s3-fs-presto/. COPY WordCount.jar /opt/flink/bin/ I am trying to run the job manually from my local laptop, able to run job successfullydocker run -it images sh ./start-cluster.sh flink run WordCount.sh for Kubernetes Now using the same image am trying to create a cluster in Application mode (https://nightlies.apache.org/flink/flink-docs-release-1.14/docs/deployment/resource-providers/native_kubernetes/) Trying to use the same image and run from Kubernetes jobmanager, getting this error k exec -it flink-jobmanager-kzgwk sh -n ha $ flink run WordCount.jar Executing WordCount example with default input data set.Use --input to specify file input.Printing result to stdout. Use --output to specify output path.WARNING: An illegal reflective access operation has occurredWARNING: Illegal reflective access by org.apache.flink.api.java.ClosureCleaner (file:/opt/flink/lib/flink-dist_2.12-1.13.2.jar) to field java.lang.String.valueWARNING: Please consider reporting this to the maintainers of org.apache.flink.api.java.ClosureCleanerWARNING: Use --illegal-access=warn to enable warnings of further illegal reflective access operationsWARNING: All illegal access operations will be denied in a future release09:30:38.530 [main] ERROR org.apache.flink.client.cli.CliFrontend - Error while running the command.org.apache.flink.client.program.ProgramInvocationException: The main method caused an error: Failed to execute job 'Streaming WordCount'. at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:372) ~[flink-dist_2.12-1.13.2.jar:1.13.2] at org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:222) ~[flink-dist_2.12-1.13.2.jar:1.13.2] at org.apache.flink.client.ClientUtils.executeProgram(ClientUtils.java:114) ~[flink-dist_2.12-1.13.2.jar:1.13.2] at org.apache.flink.client.cli.CliFrontend.executeProgram(CliFrontend.java:812) ~[flink-dist_2.12-1.13.2.jar:1.13.2] at org.apache.flink.client.cli.CliFrontend.run(CliFrontend.java:246) ~[flink-dist_2.12-1.13.2.jar:1.13.2] at org.apache.flink.client.cli.CliFrontend.parseAndRun(CliFrontend.java:1054) ~[flink-dist_2.12-1.13.2.jar:1.13.2] at org.apache.flink.client.cli.CliFrontend.lambda$main$10(CliFrontend.java:1132) ~[flink-dist_2.12-1.13.2.jar:1.13.2] at org.apache.flink.runtime.security.contexts.NoOpSecurityContext.runSecured(NoOpSecurityContext.java:28) [flink-dist_2.12-1.13.2.jar:1.13.2] at org.apache.flink.client.cli.CliFrontend.main(CliFrontend.java:1132) [flink-dist_2.12-1.13.2.jar:1.13.2]Caused by: org.apache.flink.util.FlinkException: Failed to execute job 'Streaming WordCount'. anything which I am missing ?
Re: k8s not able to submit job from jobmanager
I think , I got the answer ( application mode cluster doesn't support running job using cli ) On Tuesday, October 5, 2021, 08:38:24 AM EDT, Israel Ekpo wrote: Your Flink versions are different Your Docker container has version 1.13.2 but it seems your job is attempting to submit with 1.14 in application mode That is the first obvious observation On Tue, Oct 5, 2021 at 5:35 AM Dhiru wrote: My DockerFile FROM flink:1.13.2-scala_2.12-java11 RUN mkdir -p /opt/flink/plugins/flink-s3-fs-hadoopRUN ln -fs /opt/flink/opt/flink-s3-fs-hadoop-*.jar /opt/flink/plugins/flink-s3-fs-hadoop/. RUN mkdir -p /opt/flink/plugins/flink-s3-fs-prestoRUN ln -fs /opt/flink/opt/flink-s3-fs-presto-*.jar /opt/flink/plugins/flink-s3-fs-presto/. COPY WordCount.jar /opt/flink/bin/ I am trying to run the job manually from my local laptop, able to run job successfullydocker run -it images sh ./start-cluster.sh flink run WordCount.sh for Kubernetes Now using the same image am trying to create a cluster in Application mode (https://nightlies.apache.org/flink/flink-docs-release-1.14/docs/deployment/resource-providers/native_kubernetes/) Trying to use the same image and run from Kubernetes jobmanager, getting this error k exec -it flink-jobmanager-kzgwk sh -n ha $ flink run WordCount.jar Executing WordCount example with default input data set.Use --input to specify file input.Printing result to stdout. Use --output to specify output path.WARNING: An illegal reflective access operation has occurredWARNING: Illegal reflective access by org.apache.flink.api.java.ClosureCleaner (file:/opt/flink/lib/flink-dist_2.12-1.13.2.jar) to field java.lang.String.valueWARNING: Please consider reporting this to the maintainers of org.apache.flink.api.java.ClosureCleanerWARNING: Use --illegal-access=warn to enable warnings of further illegal reflective access operationsWARNING: All illegal access operations will be denied in a future release09:30:38.530 [main] ERROR org.apache.flink.client.cli.CliFrontend - Error while running the command.org.apache.flink.client.program.ProgramInvocationException: The main method caused an error: Failed to execute job 'Streaming WordCount'. at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:372) ~[flink-dist_2.12-1.13.2.jar:1.13.2] at org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:222) ~[flink-dist_2.12-1.13.2.jar:1.13.2] at org.apache.flink.client.ClientUtils.executeProgram(ClientUtils.java:114) ~[flink-dist_2.12-1.13.2.jar:1.13.2] at org.apache.flink.client.cli.CliFrontend.executeProgram(CliFrontend.java:812) ~[flink-dist_2.12-1.13.2.jar:1.13.2] at org.apache.flink.client.cli.CliFrontend.run(CliFrontend.java:246) ~[flink-dist_2.12-1.13.2.jar:1.13.2] at org.apache.flink.client.cli.CliFrontend.parseAndRun(CliFrontend.java:1054) ~[flink-dist_2.12-1.13.2.jar:1.13.2] at org.apache.flink.client.cli.CliFrontend.lambda$main$10(CliFrontend.java:1132) ~[flink-dist_2.12-1.13.2.jar:1.13.2] at org.apache.flink.runtime.security.contexts.NoOpSecurityContext.runSecured(NoOpSecurityContext.java:28) [flink-dist_2.12-1.13.2.jar:1.13.2] at org.apache.flink.client.cli.CliFrontend.main(CliFrontend.java:1132) [flink-dist_2.12-1.13.2.jar:1.13.2]Caused by: org.apache.flink.util.FlinkException: Failed to execute job 'Streaming WordCount'. anything which I am missing ?
After Receiving Kafka Data ( getting this error) for s3 bucket access
We have configured s3 bucket s3a://msc-sandbox-test-bucketI am not sure how come some extra characters get added for a bucket? java.lang.IllegalArgumentException: java.net.URISyntaxException: Relative path in absolute URI: s3a://msc-sandbox-test-bucket3TjIvqnUkP1YBpoy.3MxSF/3TjIwLWrI71fbMZmGYK7rV/31-07-2017 at org.apache.flink.core.fs.Path.initialize(Path.java:230)at org.apache.flink.core.fs.Path.(Path.java:139)at org.apache.flink.core.fs.Path.(Path.java:93)at org.apache.flink.streaming.api.functions.sink.filesystem.Buckets.assembleBucketPath(Buckets.java:353) at org.apache.flink.streaming.api.functions.sink.filesystem.Buckets.getOrCreateBucketForBucketId(Buckets.java:319) at org.apache.flink.streaming.api.functions.sink.filesystem.Buckets.onElement(Buckets.java:304) at org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSinkHelper.onElement(StreamingFileSinkHelper.java:103) at org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSink.invoke(StreamingFileSink.java:492) at org.apache.flink.streaming.api.operators.StreamSink.processElement(StreamSink.java:54) at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.pushToOperator(CopyingChainingOutput.java:71) at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:46) at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:26) at org.apache.flink.streaming.runtime.tasks.BroadcastingOutputCollector.collect(BroadcastingOutputCollector.java:75) at org.apache.flink.streaming.runtime.tasks.BroadcastingOutputCollector.collect(BroadcastingOutputCollector.java:32) at org.apache.flink.streaming.api.operators.CountingOutput.collect(CountingOutput.java:50) at org.apache.flink.streaming.api.operators.CountingOutput.collect(CountingOutput.java:28) at org.apache.flink.streaming.api.operators.StreamFilter.processElement(StreamFilter.java:39) at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.pushToOperator(CopyingChainingOutput.java:71) at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:46) at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:26) at org.apache.flink.streaming.api.operators.CountingOutput.collect(CountingOutput.java:50) at org.apache.flink.streaming.api.operators.CountingOutput.collect(CountingOutput.java:28) at org.apache.flink.streaming.runtime.operators.TimestampsAndWatermarksOperator.processElement(TimestampsAndWatermarksOperator.java:104) at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.pushToOperator(CopyingChainingOutput.java:71) at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:46) at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:26) at org.apache.flink.streaming.api.operators.CountingOutput.collect(CountingOutput.java:50) at org.apache.flink.streaming.api.operators.CountingOutput.collect(CountingOutput.java:28) at org.apache.flink.streaming.api.operators.StreamSourceContexts$ManualWatermarkContext.processAndCollectWithTimestamp(StreamSourceContexts.java:322) at org.apache.flink.streaming.api.operators.StreamSourceContexts$WatermarkContext.collectWithTimestamp(StreamSourceContexts.java:426) at org.apache.flink.streaming.connectors.kafka.internals.AbstractFetcher.emitRecordsWithTimestamps(AbstractFetcher.java:365) at org.apache.flink.streaming.connectors.kafka.internals.KafkaFetcher.partitionConsumerRecordsHandler(KafkaFetcher.java:183) at org.apache.flink.streaming.connectors.kafka.internals.KafkaFetcher.runFetchLoop(KafkaFetcher.java:142) at org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumerBase.run(FlinkKafkaConsumerBase.java:826) at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:110) at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:66) at org.apache.flink.streaming.runtime.tasks.SourceStreamTask$LegacySourceFunctionThread.run(SourceStreamTask.java:269)Caused by: java.net.URISyntaxException: Relative path in absolute URI: s3a://msc-sandbox-test-bucket3TjIvqnUkP1YBpoy.3MxSF/3TjIwLWrI71fbMZmGYK7rV/31-07-2017 at java.base/java.net.URI.checkPath(Unknown Source)at java.base/java.net.URI.(Unknown Source)at org.apache.flink.core.fs.Path.initialize(Path.java:228)... 36 more
Re: After Receiving Kafka Data ( getting this error) for s3 bucket access
sorry , there was issue with path of s3 bucket, Got this fixed .. Sorry for troubling you guys On Sunday, October 10, 2021, 12:33:16 PM EDT, Dhiru wrote: We have configured s3 bucket s3a://msc-sandbox-test-bucketI am not sure how come some extra characters get added for a bucket? java.lang.IllegalArgumentException: java.net.URISyntaxException: Relative path in absolute URI: s3a://msc-sandbox-test-bucket3TjIvqnUkP1YBpoy.3MxSF/3TjIwLWrI71fbMZmGYK7rV/31-07-2017 at org.apache.flink.core.fs.Path.initialize(Path.java:230)at org.apache.flink.core.fs.Path.(Path.java:139)at org.apache.flink.core.fs.Path.(Path.java:93)at org.apache.flink.streaming.api.functions.sink.filesystem.Buckets.assembleBucketPath(Buckets.java:353) at org.apache.flink.streaming.api.functions.sink.filesystem.Buckets.getOrCreateBucketForBucketId(Buckets.java:319) at org.apache.flink.streaming.api.functions.sink.filesystem.Buckets.onElement(Buckets.java:304) at org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSinkHelper.onElement(StreamingFileSinkHelper.java:103) at org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSink.invoke(StreamingFileSink.java:492) at org.apache.flink.streaming.api.operators.StreamSink.processElement(StreamSink.java:54) at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.pushToOperator(CopyingChainingOutput.java:71) at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:46) at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:26) at org.apache.flink.streaming.runtime.tasks.BroadcastingOutputCollector.collect(BroadcastingOutputCollector.java:75) at org.apache.flink.streaming.runtime.tasks.BroadcastingOutputCollector.collect(BroadcastingOutputCollector.java:32) at org.apache.flink.streaming.api.operators.CountingOutput.collect(CountingOutput.java:50) at org.apache.flink.streaming.api.operators.CountingOutput.collect(CountingOutput.java:28) at org.apache.flink.streaming.api.operators.StreamFilter.processElement(StreamFilter.java:39) at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.pushToOperator(CopyingChainingOutput.java:71) at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:46) at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:26) at org.apache.flink.streaming.api.operators.CountingOutput.collect(CountingOutput.java:50) at org.apache.flink.streaming.api.operators.CountingOutput.collect(CountingOutput.java:28) at org.apache.flink.streaming.runtime.operators.TimestampsAndWatermarksOperator.processElement(TimestampsAndWatermarksOperator.java:104) at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.pushToOperator(CopyingChainingOutput.java:71) at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:46) at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:26) at org.apache.flink.streaming.api.operators.CountingOutput.collect(CountingOutput.java:50) at org.apache.flink.streaming.api.operators.CountingOutput.collect(CountingOutput.java:28) at org.apache.flink.streaming.api.operators.StreamSourceContexts$ManualWatermarkContext.processAndCollectWithTimestamp(StreamSourceContexts.java:322) at org.apache.flink.streaming.api.operators.StreamSourceContexts$WatermarkContext.collectWithTimestamp(StreamSourceContexts.java:426) at org.apache.flink.streaming.connectors.kafka.internals.AbstractFetcher.emitRecordsWithTimestamps(AbstractFetcher.java:365) at org.apache.flink.streaming.connectors.kafka.internals.KafkaFetcher.partitionConsumerRecordsHandler(KafkaFetcher.java:183) at org.apache.flink.streaming.connectors.kafka.internals.KafkaFetcher.runFetchLoop(KafkaFetcher.java:142) at org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumerBase.run(FlinkKafkaConsumerBase.java:826) at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:110) at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:66) at org.apache.flink.streaming.runtime.tasks.SourceStreamTask$LegacySourceFunctionThread.run(SourceStreamTask.java:269)Caused by: java.net.URISyntaxException: Relative path in absolute URI: s3a://msc-sandbox-test-bucket3TjIvqnUkP1YBpoy.3MxSF/3TjIwLWrI71fbMZmGYK7rV/31-07-2017 at java.base/java.net.URI.checkPath(Unknown Source)at java.base/java.net.URI.(Unknown Source)at org.apache.flink.core.fs.Path.initialize(Path.java:228)... 36 more
EKs FlinkK8sOperator for 1.20
hi , I was planning to install Flink using k8sOperator for EKS version 1.20GitHub - GoogleCloudPlatform/flink-on-k8s-operator: Kubernetes operator for managing the lifecycle of Apache Flink and Beam applications. | | | | | | | | | | | GitHub - GoogleCloudPlatform/flink-on-k8s-operator: Kubernetes operator ... Kubernetes operator for managing the lifecycle of Apache Flink and Beam applications. - GitHub - GoogleCloudPlat... | | | When I googled, its says still not supported, did anyone found a similar issue, and if we need to have a support need to move k8s support to < 1.18
Re: EKs FlinkK8sOperator for 1.20
Thanks Kim, I got solution, we need to downgrade controller-gen@v0.2.4 to make this working But thanks a lot On Monday, October 18, 2021, 11:46:23 PM EDT, Youngwoo Kim (김영우) wrote: Hi Dhiru, Take a look at this flink operator, https://github.com/spotify/flink-on-k8s-operatorThe operator is forked and even enhanced by Soptify devs and contributors. Looks like it works on k8s 0.20+ Thanks,Youngwoo On Mon, Oct 18, 2021 at 12:05 AM Dhiru wrote: hi , I was planning to install Flink using k8sOperator for EKS version 1.20GitHub - GoogleCloudPlatform/flink-on-k8s-operator: Kubernetes operator for managing the lifecycle of Apache Flink and Beam applications. | | | | | | | | | | | GitHub - GoogleCloudPlatform/flink-on-k8s-operator: Kubernetes operator ... Kubernetes operator for managing the lifecycle of Apache Flink and Beam applications. - GitHub - GoogleCloudPlat... | | | When I googled, its says still not supported, did anyone found a similar issue, and if we need to have a support need to move k8s support to < 1.18
