How to set UserGroupInformation?

[Xinyu Zhang]

Hi all

I'm trying to write data to HDFS in directory "/home/hbase/XXX". Only
"hbase" user can write to the directory.
The problem is: I submit a job to yarn. The job will be a "yarn" user to
write data, while "yarn" user is not allowed to write to  "/home/hbase/XXX".
Is there any method that I can set UserGroupInformation to "hbase"?

Thanks!

Xinyu Zhang

Re: Trigerring Savepoint for the Flink Job

[Anil]

Thanks for the reply Rong. We had updated Athenax to version 1.4. 

I had checked Flink 1.4, it's rest endpoint dose not support only creating
Savepoint. It has cancel With Savepoint. I think creating Savepoint is
supported in 1.5. Since we can't upgrade to 1.5 at the moment it would like
to find a workaround for the moment. 

Can you tell me how to reattaches to a running job in the cluster. 





--
Sent from: http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/

Re: Trigerring Savepoint for the Flink Job

[Rong Rong]

Hi Anil,

I am actually one of the engineers maintaining Uber's AthenaX open-source
platform. As for now, AthenaX is still running on Flink 1.3.2 so it might
be weird to follow the 1.4.2 release document. (We are working on upgrading
to latest 1.5 release)

For your question regarding savepoints, AthenaX does not support savepoint
natively at this moment. But we have a separated API WatchdogPolicy that
you can customized. It supports monitoring / management APIs which will be
called periodically. You can utilize that to trigger your daily savepoints.
In term of how to do savepoint, I think the REST API [1] might be a good
starting point in AthenaX case because we launch the job in detach mode.

Please let me know if this is helpful.

Thanks,
Rong

[1]
https://ci.apache.org/projects/flink/flink-docs-release-1.5/monitoring/rest_api.html#available-requests

On Thu, May 31, 2018 at 1:01 PM, Anil  wrote:

> I am using Flink 1.4.2. I have forker Uber's AthenaX  project
>   .
>
> The Flink jobs are deployed in Yarn cluster. I needed to save the Savepoint
> for all the jobs everyday.
>
> ClusterClient
>  flink-clients/src/main/java/org/apache/flink/client/
> program/ClusterClient.java#L672>
> gave an implementation for saving savepoint using Flink ID.
> YarnClusterClient
>  flink-yarn/src/main/java/org/apache/flink/yarn/YarnClusterClient.java>
> is an implementation of ClusteClient.
>
> Initial though was to use YarnClusterClient instance with Flink Id (I save
> this when the Flink Job is deployed to Yarn cluster) to trigger savepoint.
> So I created an instance of YarnClusterClient once and saved it so that I
> could use it anytime in the application. But this doesn't seems to work. It
> doesn't seems that it can cancel or trigger savepoint even with valid Flink
> ID. When I try to cancel a valid Flink Job it throws and error for invalid
> id.
>
> I would appreciate if someone could help me out here.
>
>
>
>
>
> --
> Sent from: http://apache-flink-user-mailing-list-archive.2336050.
> n4.nabble.com/
>

REMINDER: Apache EU Roadshow 2018 in Berlin is less than 2 weeks away!

[sharan]

Hello Apache Supporters and Enthusiasts

This is a reminder that our Apache EU Roadshow in Berlin is less than 
two weeks away and we need your help to spread the word. Please let your 
work colleagues, friends and anyone interested in any attending know 
about our Apache EU Roadshow event.


We have a great schedule including tracks on Apache Tomcat, Apache Http 
Server, Microservices, Internet of Things (IoT) and Cloud Technologies. 
You can find more details at the link below:


https://s.apache.org/0hnG

Ticket prices will be going up on 8^th June 2018, so please make sure 
that you register soon if you want to beat the price increase. 
https://foss-backstage.de/tickets


Remember that registering for the Apache EU Roadshow also gives you 
access to FOSS Backstage so you can attend any talks and workshops from 
both conferences. And don’t forget that our Apache Lounge will be open 
throughout the whole conference as a place to meet up, hack and relax.


We look forward to seeing you in Berlin!

Thanks
Sharan Foga,  VP Apache Community Development

http://apachecon.com/
@apachecon

PLEASE NOTE: You are receiving this message because you are subscribed 
to a user@ or dev@ list of one or more Apache Software Foundation projects.

Looking for a working POM file example for EMR cluster

[Sandybayev, Turar (CAI - Atlanta)]

Hi,

I'm looking for a sample POM file that works when running on EMR cluster. I'm 
new to Flink and EMR, so I'm simply following AWS EMR documentation on Flink 
and I am creating a Step and submitting my program JAR file. My program is just 
a slight modification of the Wikipedia example. 

I was trying to follow an example from AWS reference architecture for their 
Taxi events example: 
https://github.com/aws-samples/flink-stream-processing-refarch/blob/master/flink-taxi-stream-processor/pom.xml
 

However, I've been seeing various errors having to do with dependencies and 
ClassNotFoundExceptions for basic common Flink dependencies. I tried removing 
excludes from the maven-shade-plugin section of the POM file from the reference 
architecture, and now I'm seeing the following exception:

Exception in thread "main" 
com.typesafe.config.ConfigException$UnresolvedSubstitution: Could not resolve 
substitution to a value: ${akka.stream.materializer}

If I run a local Flink cluster and submit my JAR, I'm not seeing any issues 
with pretty much any way I modify the POM file. I would greatly appreciate if 
someone can point me to a working POM example.

Thanks!
Turar

 

Trigerring Savepoint for the Flink Job

[Anil]

I am using Flink 1.4.2. I have forker Uber's AthenaX  project
  . 

The Flink jobs are deployed in Yarn cluster. I needed to save the Savepoint
for all the jobs everyday.

ClusterClient

  
gave an implementation for saving savepoint using Flink ID. 
YarnClusterClient

  
is an implementation of ClusteClient.

Initial though was to use YarnClusterClient instance with Flink Id (I save
this when the Flink Job is deployed to Yarn cluster) to trigger savepoint. 
So I created an instance of YarnClusterClient once and saved it so that I
could use it anytime in the application. But this doesn't seems to work. It
doesn't seems that it can cancel or trigger savepoint even with valid Flink
ID. When I try to cancel a valid Flink Job it throws and error for invalid
id. 

I would appreciate if someone could help me out here.





--
Sent from: http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/

RE: Use element of the DataStream in parameter of RichMapFunction (open function not called)

[Robin, Isabelle]

Hello Fabian,

Thanks for this quick answer ! I understand I cannot mix the custom map 
function with the lambda definition. The workaround we found for the issue was 
to pass the function to get the key as a parameter to the CustomMapFunction 
instead of the key itself, and now it works !

Regards,

Isabelle

De : Fabian Hueske [mailto:fhue...@gmail.com]
Envoyé : jeudi 31 mai 2018 09:34
À : Robin, Isabelle 
Cc : user@flink.apache.org; Nejjar, Driss 
Objet : Re: Use element of the DataStream in parameter of RichMapFunction (open 
function not called)

Hi Isabelle,
Welcome to the Flink user mailing list!
You are mixing up the two ways to specify a function:
1. Defining a function as a class / object and passing an instance in the map() 
method. Given your CustomMapFunction class, this looks as follows:
stream.keyBy(...).map(new CustomMapFunction())
2. Defining a function as a lambda function. In this case, you don't have to 
define an extra class (CustomMapFunction in your case) but rather give the 
logic that you would put into the MapFunction.map() method as a lambda function.
Internally, the lambda function is "wrapped" in a MapFunction. This style looks 
as follows:
stream.keyBy(...).map(t => [directly apply your map() logic and return a 
Transaction])

In your code, you define a lambda function (by using "transaction => ...") that 
returns a CustomMapFunction, instead of a Transaction. Hence, the first program 
fails due to an invalid type.
I would recommend to use the 1. class/object style for complex logic and the 2. 
lambda function style for simple transformations, filters, key-extractions, ...
Btw. your key extractor (transaction => Feature.transaction).getKey) is not 
very efficient because it creates a new Feature object, for each transaction.
Best, Fabian


2018-05-30 17:38 GMT+02:00 Robin, Isabelle 
mailto:isabelle.ro...@ercom.fr>>:
Hello,

I’m working with Flink 1.4.2 (Scala API) and I’m having some trouble with my 
custom RichMapFunction as I want the element in my Datastream to also be used 
for a parameter of this custom class. My RichMapFunction is a simple counter 
based on a MapState

Let’s say I have those classes

-  case class Feature(transaction: Transaction) { override def getKey: 
(String, String) = … }

-  class CustomMapFunction(featureKey: (String, String)) extends 
RichMapFunction[Transaction, Transaction]

I implemented my custom map function with the needed functions but I 
encountered different issues as I tried several solutions for this. In the 
following chunks of code, stream is a DataStream[Transaction] and I expect a 
DataStream[Transaction] as output type too


• stream.keyBy(transaction => 
Feature(transaction).getKey).map(transaction => new 
CustomMapFunction(Feature(transaction).getKey))

o   this leads to a compilation error (“Expression of type CustomMapFunction 
doesn’t conform to expected type R_”), which, as far as I understand, should 
come from the fact I’m already using transaction for the Feature(transaction) 
part

• stream.keyBy(transaction => 
Feature(transaction).getKey).map(transaction => new 
CustomMapFunction(Feature(transaction).getKey).map(transaction))

o   compiles but fails with a NullPointerException at runtime as the MapState 
is not initialized. When running with debugger the open function was not used 
which leads the MapState to stay null (I don’t have this problem with a more 
simple version of my CustomMapFunction which does not need this parameter based 
on the transaction)

Do you have an idea of how I could solve this issue ?

Thanks in advance for any help and I hope I was clear enough (that’s my first 
question on the mailing list, don’t hesitate to say if I forgot some steps or 
elements ☺)

Best regards,

Isabelle

Re: Flink 1.2.1 - Job initial start time

[Bajaj, Abhinav]

Adding my team mate.

~ Abhinav Bajaj

From: "Bajaj, Abhinav" 
Date: Thursday, May 31, 2018 at 9:30 AM
To: "user@flink.apache.org" 
Subject: Flink 1.2.1 - Job initial start time

Hi,

Is there a way to know the initial start time of a job? I am looking for the 
first time job was started and not the time of the restart.

It seems the “start-time” reported by monitoring API under 
joboverview
 is reset after a restart.
Also, it seems the “start-time” & “timestamps” under job 
details
 are also reset when a job is restarted.
I expected the “CREATED” date under “timestamps” to be initial created time of 
the job but it is reset after a job is restarted.

We are using Flink 1.2.1 and trying to figure out a way to find the initial 
start time, if possible.

Thanks,
Abhinav Bajaj

Flink 1.2.1 - Job initial start time

[Bajaj, Abhinav]

Hi,

Is there a way to know the initial start time of a job? I am looking for the 
first time job was started and not the time of the restart.

It seems the “start-time” reported by monitoring API under 
joboverview
 is reset after a restart.
Also, it seems the “start-time” & “timestamps” under job 
details
 are also reset when a job is restarted.
I expected the “CREATED” date under “timestamps” to be initial created time of 
the job but it is reset after a job is restarted.

We are using Flink 1.2.1 and trying to figure out a way to find the initial 
start time, if possible.

Thanks,
Abhinav Bajaj

File does not exist prevent from Job manager to start .

[miki haiat]

Hi,

Im having some wierd issue with the JM recovery ,
I using HDFS and ZOOKEEPER for HA stand alone cluster .

Iv  stop the cluster change some parameters in the flink conf (Memory).
But now when i start the cluster again im having an error that preventing
from JM to start.
somehow the checkpoint file doesn't exists in HDOOP  and JM wont start .

full log JM log file



> 2018-05-31 11:57:05,568 ERROR
> org.apache.flink.runtime.entrypoint.ClusterEntrypoint - Fatal error
> occurred in the cluster entrypoint.

Caused by: java.lang.Exception: Cannot set up the user code libraries: File
does not exist:
/flink1.5/ha/default/blob/job_5c545fc3f43d69325fb9966b8dd4c8f3/blob_p-5d9f3be555d3b05f90b5e148235d25730eb65b3d-ae486e221962f7b96e36da18fe1c57ca
at
org.apache.hadoop.hdfs.server.namenode.INodeFile.valueOf(INodeFile.java:72)

Re: is it OK to PR just for code comment errors?

[Rong Rong]

Yes, Chesnay was right if the error code comment doesn't affect the clarity
of the message. Sorry for the confusion.

--
Rong

On Thu, May 31, 2018 at 7:54 AM, Chesnay Schepler 
wrote:

> For typos in the documentation or javadocs feel free to open a PR.
>
> I would leave code comments alone as long as the typo doesn't affect the
> clarity/semantics of the comment,
> to not fragment the git history unnecessarily.
>
>
> On 31.05.2018 11:40, makeyang wrote:
>
>> I am not sure if it is OK to PR just for code comment errors since I found
>> there is one when I am reading code
>>
>>
>>
>> --
>> Sent from: http://apache-flink-user-mailing-list-archive.2336050.n4.
>> nabble.com/
>>
>>
>

Re: is it OK to PR just for code comment errors?

[Shuyi Chen]

I think this is classified as "hotifx" in the contributor list, no JIRA
should be needed.

*" Typo fixes that have no associated JIRA issue should be named following
this pattern: `[hotfix] [docs] Fix typo in event time introduction` or
`[hotfix] [javadocs] Expand JavaDoc for PuncuatedWatermarkGenerator`."*


On Thu, May 31, 2018 at 7:54 AM, Chesnay Schepler 
wrote:

> For typos in the documentation or javadocs feel free to open a PR.
>
> I would leave code comments alone as long as the typo doesn't affect the
> clarity/semantics of the comment,
> to not fragment the git history unnecessarily.
>
>
> On 31.05.2018 11:40, makeyang wrote:
>
>> I am not sure if it is OK to PR just for code comment errors since I found
>> there is one when I am reading code
>>
>>
>>
>> --
>> Sent from: http://apache-flink-user-mailing-list-archive.2336050.n4.
>> nabble.com/
>>
>>
>


-- 
"So you have to trust that the dots will somehow connect in your future."

Re: is it OK to PR just for code comment errors?

[Chesnay Schepler]

For typos in the documentation or javadocs feel free to open a PR.

I would leave code comments alone as long as the typo doesn't affect the 
clarity/semantics of the comment,

to not fragment the git history unnecessarily.

On 31.05.2018 11:40, makeyang wrote:

I am not sure if it is OK to PR just for code comment errors since I found
there is one when I am reading code



--
Sent from: http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/

Re: is it OK to PR just for code comment errors?

[Rong Rong]

Hi Makeyang,

It is definitely OK to open a PR for correcting code comment errors.
Especially for the ones that are misleading or provide incorrect
information to users.

I am not sure if it is required if only for a one-liner change, but please
file a JIRA ticket for the code comment error as well.

Thanks,
Rong

On Thu, May 31, 2018 at 2:40 AM, makeyang  wrote:

> I am not sure if it is OK to PR just for code comment errors since I found
> there is one when I am reading code
>
>
>
> --
> Sent from: http://apache-flink-user-mailing-list-archive.2336050.
> n4.nabble.com/
>

Re: Ask about convert KafkaTableSource to table

[Shuyi Chen]

Please take a look at the table api/SQL doc [1] below on how to register a
table/tablesource. Thanks.

[1]
https://ci.apache.org/projects/flink/flink-docs-release-1.4/dev/table/common.html#register-a-table

On Thu, May 31, 2018 at 6:22 AM, Radhya Sahal 
wrote:

> Hi,
>
> Could you help me which is the integrated versions of Java/Kafka/ Flink
> which can work without errors.
>
>
> I have a problem when I read streamed data from  KafkaTableSource to table.
> I use this command
>
>  tableEnv.registerDataStream("sensors", myKafkaTableSource)
>
> But, there it doesn't run (i.e., compilation error). Since, these are the
> dependencies in pom.xml
>
> 
> 
> org.apache.flink
> flink-streaming-java_2.11
> 1.4.0
> 
> 
> org.apache.flink
> flink-clients_2.11
> 1.4.0
> 
> 
> org.apache.flink
> flink-connector-kafka-0.11_2.11
> 1.4.0
> 
> 
> org.apache.flink
> flink-table_2.11
> 1.4.0
> 
> 
> org.apache.flink
> flink-core
> 1.4.0
> 
> 
> org.apache.flink
> flink-streaming-scala_2.11
> 1.4.0
> 
> 
>
>
>
>
>
> Regards.
>
> Radhya Sahal
>
>
>
>
> --
> Sent from: http://apache-flink-user-mailing-list-archive.2336050.
> n4.nabble.com/
>



-- 
"So you have to trust that the dots will somehow connect in your future."

Ask about convert KafkaTableSource to table

[Radhya Sahal]

Hi, 

Could you help me which is the integrated versions of Java/Kafka/ Flink
which can work without errors. 


I have a problem when I read streamed data from  KafkaTableSource to table.
I use this command 

 tableEnv.registerDataStream("sensors", myKafkaTableSource)

But, there it doesn't run (i.e., compilation error). Since, these are the
dependencies in pom.xml



org.apache.flink
flink-streaming-java_2.11
1.4.0


org.apache.flink
flink-clients_2.11
1.4.0


org.apache.flink
flink-connector-kafka-0.11_2.11
1.4.0


org.apache.flink
flink-table_2.11
1.4.0


org.apache.flink
flink-core
1.4.0


org.apache.flink
flink-streaming-scala_2.11
1.4.0







Regards. 

Radhya Sahal




--
Sent from: http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/

Re: Multiple Task Slots support in Flink 1.5

[Kien Truong]

Hi,

We're using multiple slots per TaskManager with legacy mode, and 
everything works fine.


For the new default mode, it also seems to works for us, so I'm not sure 
what is not supported. May be someone from Flink team could clarify.



Best regards,

Kien


On 5/31/2018 4:26 AM, Abdul Qadeer wrote:

Hi!

I came across the following point in release notes 
 
of 1.5 version:


"The allocation of TaskManagers with multiple slots is not fully 
supported yet."


Does this mean the support for it will come as a patch for 1.5? or 
will it be in the next stable release?
If I use legacy mode, will that support multiple slots per 
TaskManager?, or is it only the deployment change that will get affected?

some default config questions

[makeyang]

why jobmanager.execution.failover-strategy default value is full not
Individual?
why taskmanager.jvm-exit-on-oom default value is false not true?
code is flink 1.5



--
Sent from: http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/

is it OK to PR just for code comment errors?

[makeyang]

I am not sure if it is OK to PR just for code comment errors since I found
there is one when I am reading code



--
Sent from: http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/

Re: Use element of the DataStream in parameter of RichMapFunction (open function not called)

[Fabian Hueske]

Hi Isabelle,

Welcome to the Flink user mailing list!

You are mixing up the two ways to specify a function:

1. Defining a function as a class / object and passing an instance in the
map() method. Given your CustomMapFunction class, this looks as follows:
stream.keyBy(...).map(new CustomMapFunction())

2. Defining a function as a lambda function. In this case, you don't have
to define an extra class (CustomMapFunction in your case) but rather give
the logic that you would put into the MapFunction.map() method as a lambda
function.
Internally, the lambda function is "wrapped" in a MapFunction. This style
looks as follows:
stream.keyBy(...).map(t => [directly apply your map() logic and return a
Transaction])

In your code, you define a lambda function (by using "transaction => ...")
that returns a CustomMapFunction, instead of a Transaction. Hence, the
first program fails due to an invalid type.
I would recommend to use the 1. class/object style for complex logic and
the 2. lambda function style for simple transformations, filters,
key-extractions, ...

Btw. your key extractor (transaction => Feature.transaction).getKey) is not
very efficient because it creates a new Feature object, for each
transaction.

Best, Fabian


2018-05-30 17:38 GMT+02:00 Robin, Isabelle :

> Hello,
>
>
>
> I’m working with Flink 1.4.2 (Scala API) and I’m having some trouble with
> my custom RichMapFunction as I want the element in my Datastream to also be
> used for a parameter of this custom class. My RichMapFunction is a simple
> counter based on a MapState
>
>
>
> Let’s say I have those classes
>
> -  case class Feature(transaction: Transaction) { override def
> getKey: (String, String) = … }
>
> -  class CustomMapFunction(featureKey: (String, String)) extends
> RichMapFunction[Transaction, Transaction]
>
>
>
> I implemented my custom map function with the needed functions but I
> encountered different issues as I tried several solutions for this. In the
> following chunks of code, stream is a DataStream[Transaction] and I expect
> a DataStream[Transaction] as output type too
>
>
>
> · stream.keyBy(transaction => 
> Feature(transaction).getKey).map(transaction
> => new CustomMapFunction(Feature(transaction).getKey))
>
> o   this leads to a compilation error (“Expression of type
> CustomMapFunction doesn’t conform to expected type R_”), which, as far as I
> understand, should come from the fact I’m already using transaction for the
> Feature(transaction) part
>
> · stream.keyBy(transaction => 
> Feature(transaction).getKey).map(transaction
> => new CustomMapFunction(Feature(transaction).getKey).map(transaction))
>
> o   compiles but fails with a NullPointerException at runtime as the
> MapState is not initialized. When running with debugger the open function
> was not used which leads the MapState to stay null (I don’t have this
> problem with a more simple version of my CustomMapFunction which does not
> need this parameter based on the transaction)
>
>
>
> Do you have an idea of how I could solve this issue ?
>
>
>
> Thanks in advance for any help and I hope I was clear enough (that’s my
> first question on the mailing list, don’t hesitate to say if I forgot some
> steps or elements J)
>
>
>
> Best regards,
>
>
>
> Isabelle
>

Re: Flink and AWS S3 integration: java.lang.NullPointerException: null uri host

[Fabian Wollert]

I'm running it in docker on EC2, cant use EMR ... yes i followed those
instructions.

Cheers

--


*Fabian WollertZalando SE*

E-Mail: fabian.woll...@zalando.de


Am Do., 31. Mai 2018 um 03:07 Uhr schrieb Bowen Li :

> Did you run Flink on AWS EMR or somewhere else? Have you read and followed
> instructions on
> https://ci.apache.org/projects/flink/flink-docs-release-1.5/ops/deployment/aws.html#amazon-web-services-aws
> ?
>
>
>
> On Wed, May 30, 2018 at 7:08 AM, Fabian Wollert  wrote:
>
>> Hi, I'm trying to set up Checkpoints for Flink Jobs with S3 as a
>> filesystem backend. I configured the following:
>>
>> state.backend=filesystem
>> state.backend.fs.checkpointdir=s3:///mybucket/
>> state.checkpoints.dir=s3:///mybucket/
>> state.checkpoints.num-retained=3
>>
>> I also copied the flink-s3-fs-hadoop-1.5.0.jar into the lib folder.
>>
>> I get now though the following error message:
>>
>> Caused by: java.lang.NullPointerException: null uri host.
>> at java.util.Objects.requireNonNull(Objects.java:228)
>> at
>> org.apache.flink.fs.s3hadoop.shaded.org.apache.hadoop.fs.s3native.S3xLoginHelper.buildFSURI(S3xLoginHelper.java:65)
>> at
>> org.apache.flink.fs.s3hadoop.shaded.org.apache.hadoop.fs.s3a.S3AFileSystem.initialize(S3AFileSystem.java:165)
>> at
>> org.apache.flink.fs.s3hadoop.S3FileSystemFactory.create(S3FileSystemFactory.java:133)
>>
>> I tried to dig deeper into the source code, but struggled to find
>>
>>- what is meant with this URI
>>- where to configure it
>>
>> Can anybody give some advice how to set up the S3 Backend with the new
>> shaded lib jar?
>>
>> Thanks in advance
>> --
>>
>>
>> *Fabian WollertZalando SE*
>>
>> E-Mail: fabian.woll...@zalando.de
>>
>> Tamara-Danz-Straße 1
>> 
>> 10243 Berlin
>> 
>> Fax: +49 (0)30 2759 46 93
>> E-mail: legalnot...@zalando.co.uk
>> Notifications of major holdings (Sec. 33, 38, 39 WpHG):  +49 (0)30
>> 2000889349
>>
>> Management Board:
>> Robert Gentz, David Schneider, Rubin Ritter
>>
>> Chairman of the Supervisory Board:
>> Lothar Lanz
>>
>> Person responsible for providing the contents of Zalando SE acc. to Art.
>> 55 RStV [Interstate Broadcasting Agreement]: Rubin Ritter
>> Registered at the Local Court Charlottenburg Berlin, HRB 158855 B
>> VAT registration number: DE 260543043
>>
>
>

Re: NPE in flink sql over-window

[Fabian Hueske]

Hi Yan,

Thanks for the details and for digging into the issue.
If I got it right, the NPE caused the job failure and recovery (instead of
being the result of a recovery), correct?

Best, Fabian

2018-05-31 7:00 GMT+02:00 Yan Zhou [FDS Science] :

> Thanks for the replay.
>
>
> Yes, it only happen if I config the idle state retention times. The error
> occurs the first time before the first recovery. I haven't run with
> proctime but rowtime in flink 1.4.x. I am not sure if it will cause
> problems with proctime in 1.4.x.
>
>
> I am adding some trace log for ProcTimeBoundedRangeOver. I will update
> with my test result and fire a JIRA after that.
>
>
> Best
>
> Yan
> --
> *From:* Fabian Hueske 
> *Sent:* Wednesday, May 30, 2018 1:43:01 AM
> *To:* Dawid Wysakowicz
> *Cc:* user
> *Subject:* Re: NPE in flink sql over-window
>
> Hi,
>
> Dawid's analysis is certainly correct, but looking at the code this should
> not happen.
>
> I have a few questions:
> - You said this only happens if you configure idle state retention times,
> right?
> - Does the error occur the first time without a previous recovery?
> - Did you run the same query on Flink 1.4.x without any problems?
>
> Thanks, Fabian
>
> 2018-05-30 9:25 GMT+02:00 Dawid Wysakowicz :
>
> Hi Yan,
>
>
> I think it is a bug in the ProcTimeBoundedRangeOver. It tries to access a
> list of elements that was already cleared and does not check against null.
> Could you please file a JIRA for that?
>
>
> Best,
>
> Dawid
>
> On 30/05/18 08:27, Yan Zhou [FDS Science] wrote:
>
> I also get warnning that CodeCache is full around that time. It's printed
> by JVM and doesn't have timestamp. But I suspect that it's because so
> many failure recoveries from checkpoint and the sql queries are dynamically
> compiled too many times.
>
>
>
> *Java HotSpot(TM) 64-Bit Server VM warning: CodeCache is full. Compiler
> has been disabled.*
> *Java HotSpot(TM) 64-Bit Server VM warning: Try increasing the code cache
> size using -XX:ReservedCodeCacheSize=*
> *CodeCache: size=245760Kb used=244114Kb max_used=244146Kb free=1645Kb*
> *bounds [0x7fa4fd00, 0x7fa50c00, 0x7fa50c00]*
> *total_blobs=54308 nmethods=53551 adapters=617*
> *compilation: disabled (not enough contiguous free space left)*
>
>
>
> --
> *From:* Yan Zhou [FDS Science]  
> *Sent:* Tuesday, May 29, 2018 10:52:18 PM
> *To:* user@flink.apache.org
> *Subject:* NPE in flink sql over-window
>
>
> Hi,
>
> I am using flink sql 1.5.0. My application throws NPE. And after it
> recover from checkpoint automatically, it throws NPE immediately from same
> line of code.
>
>
> My application read message from kafka, convert the datastream into a
> table, issue an Over-window aggregation and write the result into a sink.
> NPE throws from class ProcTimeBoundedRangeOver. Please see exception log
> at the bottom.
>
>
> The exceptions always happens after the application started for 
> *maxIdleStateRetentionTime
> *time.  What could be the possible causes?
>
>
> Best
>
> Yan
>
>
> *2018-05-27 11:03:37,656 INFO  org.apache.flink.runtime.taskmanager.Task
>- over: (PARTITION BY: uid, ORDER BY: proctime,
> RANGEBETWEEN 8640 PRECEDI*
> *NG AND CURRENT ROW, select: (id, uid, proctime, group_concat($7) AS
> w0$o0)) -> select: *
> *(id, uid, proctime, w0$o0 AS EXPR$3) -> to: Row -> Flat Map -> Filter ->
> Sink: Unnamed (3/15) (327*
> *efe96243bbfdf1f1e40a3372f64aa) switched from RUNNING to FAILED.*
> *TimerException{java.lang.NullPointerException}*
> *   at
> org.apache.flink.streaming.runtime.tasks.SystemProcessingTimeService$TriggerTask.run(SystemProcessingTimeService.java:284)*
> *   at
> java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)*
> *   at java.util.concurrent.FutureTask.run(FutureTask.java:266)*
> *   at
> java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180)*
> *   at
> java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293)*
> *   at
> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)*
> *   at
> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)*
> *   at java.lang.Thread.run(Thread.java:748)*
> *Caused by: java.lang.NullPointerException*
> *   at
> org.apache.flink.table.runtime.aggregate.ProcTimeBoundedRangeOverWithLog.onTimer(ProcTimeBoundedRangeOver.scala:181)*
> *   at
> org.apache.flink.streaming.api.operators.LegacyKeyedProcessOperator.invokeUserFunction(LegacyKeyedProcessOperator.java:97)*
> *   at
> org.apache.flink.streaming.api.operators.LegacyKeyedProcessOperator.onProcessingTime(LegacyKeyedProcessOperator.java:81)*
> *   at
> org.apache.flink.streaming.api.operators.HeapInternalTimerService.onProcessingTime(HeapInternalTimerService.java:266)*
> *   at
> 

Re: Is Flink:1.5 Docker image broken?

[Alexandru Gutan]

Well those are unofficial, so you might raise a correspoinding issue on
github (since the images are there) for that.

On 31 May 2018 at 08:09, Chirag Dewan  wrote:

> Hi,
>
> flink:latest docker image doesn't seem to work. I am not able to access
> the Flink Dashboard after deploying it on Kubernetes.
>
> Anyone else facing the issue?
>
> Thanks,
>
> Chirag
>