[jira] [Commented] (FLINK-4535) ResourceManager registration with TaskExecutor

[ASF GitHub Bot (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLINK-4535?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15469604#comment-15469604
 ] 

ASF GitHub Bot commented on FLINK-4535:
---

Github user beyond1920 commented on the issue:

https://github.com/apache/flink/pull/2451
  
@tillrohrmann , thanks for your review. I changed the pr based on your 
advice, including ：
1. Modify UnmatchedLeaderSessionIDException to LeaderSessionIDException, 
format code style and method comment, and add checkNotNull to constructor
2. Format code style and method comment of ResourceManager
3. Group the TaskExecutorGateway and the InstanceID into a 
TaskExecutorRegistration
4. Modify testcases which are expected to throw exception at 
ResourceManagerTest

Besides, I don't understand this comment of yours: "Should we fail or 
decline the registration here? So either sending an exception or a 
RegistrationResponse.Decline message.". Do you means you prefer to sending a 
Decline message?


> ResourceManager registration with TaskExecutor
> --
>
> Key: FLINK-4535
> URL: https://issues.apache.org/jira/browse/FLINK-4535
> Project: Flink
>  Issue Type: Sub-task
>  Components: Cluster Management
>Reporter: zhangjing
>Assignee: zhangjing
>
> When TaskExecutor register at ResourceManager, it takes the following 3 input 
> parameters:
> 1. resourceManagerLeaderId:  the fencing token for the ResourceManager leader 
> which is kept by taskExecutor who send the registration
> 2.  taskExecutorAddress: the address of taskExecutor
> 3. resourceID: The resource ID of the TaskExecutor that registers
> ResourceManager need to process the registration event based on the following 
> steps:
> 1. Check whether input resourceManagerLeaderId is as same as the current 
> leadershipSessionId of resourceManager. If not, it means that maybe two or 
> more resourceManager exists at the same time, and current resourceManager is 
> not the proper rm. so it  rejects or ignores the registration.
> 2. Check whether exists a valid taskExecutor at the giving address by 
> connecting to the address. Reject the registration from invalid address.
> 3. Check whether it is a duplicate registration by input resourceId, reject 
> the registration
> 4. Keep resourceID and taskExecutorGateway mapping relationships, And 
> optionally keep resourceID and container mapping relationships in yarn mode.
> 5. Create the connection between resourceManager and taskExecutor, and ensure 
> its healthy based on heartbeat rpc calls between rm and tm ?
> 6. Send registration successful ack to the taskExecutor.
> Discussion:
> Maybe we need import errorCode or several registration decline subclass to 
> distinguish the different causes of decline registration. 



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[GitHub] flink issue #2451: [FLINK-4535] [cluster management] resourceManager process...

[beyond1920]

Github user beyond1920 commented on the issue:

https://github.com/apache/flink/pull/2451
  
@tillrohrmann , thanks for your review. I changed the pr based on your 
advice, including ：
1. Modify UnmatchedLeaderSessionIDException to LeaderSessionIDException, 
format code style and method comment, and add checkNotNull to constructor
2. Format code style and method comment of ResourceManager
3. Group the TaskExecutorGateway and the InstanceID into a 
TaskExecutorRegistration
4. Modify testcases which are expected to throw exception at 
ResourceManagerTest

Besides, I don't understand this comment of yours: "Should we fail or 
decline the registration here? So either sending an exception or a 
RegistrationResponse.Decline message.". Do you means you prefer to sending a 
Decline message?


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[jira] [Assigned] (FLINK-4513) Kafka connector documentation refers to Flink 1.1-SNAPSHOT

[Neelesh Srinivas Salian (JIRA)]

 [ 
https://issues.apache.org/jira/browse/FLINK-4513?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Neelesh Srinivas Salian reassigned FLINK-4513:
--

Assignee: Neelesh Srinivas Salian

> Kafka connector documentation refers to Flink 1.1-SNAPSHOT
> --
>
> Key: FLINK-4513
> URL: https://issues.apache.org/jira/browse/FLINK-4513
> Project: Flink
>  Issue Type: Bug
>  Components: Documentation
>Affects Versions: 1.1.1
>Reporter: Fabian Hueske
>Assignee: Neelesh Srinivas Salian
>Priority: Trivial
> Fix For: 1.1.3
>
>
> The Kafka connector documentation: 
> https://ci.apache.org/projects/flink/flink-docs-release-1.1/apis/streaming/connectors/kafka.html
>  of Flink 1.1 refers to a Flink 1.1-SNAPSHOT Maven version. 



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Updated] (FLINK-4587) Yet another java.lang.NoSuchFieldError: INSTANCE

[Renkai Ge (JIRA)]

 [ 
https://issues.apache.org/jira/browse/FLINK-4587?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Renkai Ge updated FLINK-4587:
-
Attachment: flink-explore-src.zip

> Yet another java.lang.NoSuchFieldError: INSTANCE
> 
>
> Key: FLINK-4587
> URL: https://issues.apache.org/jira/browse/FLINK-4587
> Project: Flink
>  Issue Type: Bug
>  Components: Local Runtime
>Affects Versions: 1.2.0
> Environment: Latest SNAPSHOT
>Reporter: Renkai Ge
> Attachments: flink-explore-src.zip
>
>
> For some reason I need to use org.apache.httpcomponents:httpasyncclient:4.1.2 
> in flink.
> The source file is:
> {code}
> import org.apache.flink.streaming.api.scala._
> import org.apache.http.impl.nio.conn.ManagedNHttpClientConnectionFactory
> /**
>   * Created by renkai on 16/9/7.
>   */
> object Main {
>   def main(args: Array[String]): Unit = {
> val instance = ManagedNHttpClientConnectionFactory.INSTANCE
> println("instance = " + instance)
> val env = StreamExecutionEnvironment.getExecutionEnvironment
> val stream = env.fromCollection(1 to 100)
> val result = stream.map { x =>
>   x * 2
> }
> result.print()
> env.execute("xixi")
>   }
> }
> {code}
> and 
> {code}
> name := "flink-explore"
> version := "1.0"
> scalaVersion := "2.11.8"
> crossPaths := false
> libraryDependencies ++= Seq(
>   "org.apache.flink" %% "flink-scala" % "1.2-SNAPSHOT"
> exclude("com.google.code.findbugs", "jsr305"),
>   "org.apache.flink" %% "flink-connector-kafka-0.8" % "1.2-SNAPSHOT"
> exclude("com.google.code.findbugs", "jsr305"),
>   "org.apache.flink" %% "flink-streaming-scala" % "1.2-SNAPSHOT"
> exclude("com.google.code.findbugs", "jsr305"),
>   "org.apache.flink" %% "flink-clients" % "1.2-SNAPSHOT"
> exclude("com.google.code.findbugs", "jsr305"),
>   "org.apache.httpcomponents" % "httpasyncclient" % "4.1.2"
> )
> {code}
> I use `sbt assembly` to get a fat jar.
> If I run the command
> {code}
>  java -cp flink-explore-assembly-1.0.jar Main
> {code}
> I get the result 
> {code}
> instance = 
> org.apache.http.impl.nio.conn.ManagedNHttpClientConnectionFactory@4909b8da
> log4j:WARN No appenders could be found for logger 
> (org.apache.flink.api.scala.ClosureCleaner$).
> log4j:WARN Please initialize the log4j system properly.
> log4j:WARN See http://logging.apache.org/log4j/1.2/faq.html#noconfig for more 
> info.
> Connected to JobManager at Actor[akka://flink/user/jobmanager_1#-1177584915]
> 09/07/2016 12:05:26   Job execution switched to status RUNNING.
> 09/07/2016 12:05:26   Source: Collection Source(1/1) switched to SCHEDULED
> 09/07/2016 12:05:26   Source: Collection Source(1/1) switched to DEPLOYING
> ...
> 09/07/2016 12:05:26   Map -> Sink: Unnamed(20/24) switched to RUNNING
> 09/07/2016 12:05:26   Map -> Sink: Unnamed(19/24) switched to RUNNING
> 15> 30
> 20> 184
> ...
> 19> 182
> 1> 194
> 8> 160
> 09/07/2016 12:05:26   Source: Collection Source(1/1) switched to FINISHED
> ...
> 09/07/2016 12:05:26   Map -> Sink: Unnamed(1/24) switched to FINISHED
> 09/07/2016 12:05:26   Job execution switched to status FINISHED.
> {code}
> Nothing special.
> But if I run the jar by
> {code}
> ./bin/flink run shop-monitor-flink-assembly-1.0.jar
> {code}
> I will get an error
> {code}
> $ ./bin/flink run flink-explore-assembly-1.0.jar
> Cluster configuration: Standalone cluster with JobManager at /127.0.0.1:6123
> Using address 127.0.0.1:6123 to connect to JobManager.
> JobManager web interface address http://127.0.0.1:8081
> Starting execution of program
> 
>  The program finished with the following exception:
> java.lang.NoSuchFieldError: INSTANCE
>   at 
> org.apache.http.impl.nio.codecs.DefaultHttpRequestWriterFactory.(DefaultHttpRequestWriterFactory.java:53)
>   at 
> org.apache.http.impl.nio.codecs.DefaultHttpRequestWriterFactory.(DefaultHttpRequestWriterFactory.java:57)
>   at 
> org.apache.http.impl.nio.codecs.DefaultHttpRequestWriterFactory.(DefaultHttpRequestWriterFactory.java:47)
>   at 
> org.apache.http.impl.nio.conn.ManagedNHttpClientConnectionFactory.(ManagedNHttpClientConnectionFactory.java:75)
>   at 
> org.apache.http.impl.nio.conn.ManagedNHttpClientConnectionFactory.(ManagedNHttpClientConnectionFactory.java:83)
>   at 
> org.apache.http.impl.nio.conn.ManagedNHttpClientConnectionFactory.(ManagedNHttpClientConnectionFactory.java:64)
>   at Main$.main(Main.scala:9)
>   at Main.main(Main.scala)
>   at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
>   at 
> sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
>   at 
> sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
>   at 

[jira] [Created] (FLINK-4587) Yet another java.lang.NoSuchFieldError: INSTANCE

[Renkai Ge (JIRA)]

Renkai Ge created FLINK-4587:


 Summary: Yet another java.lang.NoSuchFieldError: INSTANCE
 Key: FLINK-4587
 URL: https://issues.apache.org/jira/browse/FLINK-4587
 Project: Flink
  Issue Type: Bug
  Components: Local Runtime
Affects Versions: 1.2.0
 Environment: Latest SNAPSHOT
Reporter: Renkai Ge


For some reason I need to use org.apache.httpcomponents:httpasyncclient:4.1.2 
in flink.
The source file is:
{code}
import org.apache.flink.streaming.api.scala._
import org.apache.http.impl.nio.conn.ManagedNHttpClientConnectionFactory

/**
  * Created by renkai on 16/9/7.
  */
object Main {
  def main(args: Array[String]): Unit = {
val instance = ManagedNHttpClientConnectionFactory.INSTANCE
println("instance = " + instance)

val env = StreamExecutionEnvironment.getExecutionEnvironment
val stream = env.fromCollection(1 to 100)
val result = stream.map { x =>
  x * 2
}
result.print()
env.execute("xixi")
  }
}

{code}

and 
{code}
name := "flink-explore"

version := "1.0"

scalaVersion := "2.11.8"

crossPaths := false

libraryDependencies ++= Seq(
  "org.apache.flink" %% "flink-scala" % "1.2-SNAPSHOT"
exclude("com.google.code.findbugs", "jsr305"),
  "org.apache.flink" %% "flink-connector-kafka-0.8" % "1.2-SNAPSHOT"
exclude("com.google.code.findbugs", "jsr305"),
  "org.apache.flink" %% "flink-streaming-scala" % "1.2-SNAPSHOT"
exclude("com.google.code.findbugs", "jsr305"),
  "org.apache.flink" %% "flink-clients" % "1.2-SNAPSHOT"
exclude("com.google.code.findbugs", "jsr305"),
  "org.apache.httpcomponents" % "httpasyncclient" % "4.1.2"
)
{code}
I use `sbt assembly` to get a fat jar.

If I run the command
{code}
 java -cp flink-explore-assembly-1.0.jar Main
{code}
I get the result 

{code}
instance = 
org.apache.http.impl.nio.conn.ManagedNHttpClientConnectionFactory@4909b8da
log4j:WARN No appenders could be found for logger 
(org.apache.flink.api.scala.ClosureCleaner$).
log4j:WARN Please initialize the log4j system properly.
log4j:WARN See http://logging.apache.org/log4j/1.2/faq.html#noconfig for more 
info.
Connected to JobManager at Actor[akka://flink/user/jobmanager_1#-1177584915]
09/07/2016 12:05:26 Job execution switched to status RUNNING.
09/07/2016 12:05:26 Source: Collection Source(1/1) switched to SCHEDULED
09/07/2016 12:05:26 Source: Collection Source(1/1) switched to DEPLOYING
...
09/07/2016 12:05:26 Map -> Sink: Unnamed(20/24) switched to RUNNING
09/07/2016 12:05:26 Map -> Sink: Unnamed(19/24) switched to RUNNING
15> 30
20> 184
...
19> 182
1> 194
8> 160
09/07/2016 12:05:26 Source: Collection Source(1/1) switched to FINISHED
...
09/07/2016 12:05:26 Map -> Sink: Unnamed(1/24) switched to FINISHED
09/07/2016 12:05:26 Job execution switched to status FINISHED.
{code}

Nothing special.

But if I run the jar by
{code}
./bin/flink run shop-monitor-flink-assembly-1.0.jar
{code}

I will get an error

{code}
$ ./bin/flink run flink-explore-assembly-1.0.jar
Cluster configuration: Standalone cluster with JobManager at /127.0.0.1:6123
Using address 127.0.0.1:6123 to connect to JobManager.
JobManager web interface address http://127.0.0.1:8081
Starting execution of program


 The program finished with the following exception:

java.lang.NoSuchFieldError: INSTANCE
at 
org.apache.http.impl.nio.codecs.DefaultHttpRequestWriterFactory.(DefaultHttpRequestWriterFactory.java:53)
at 
org.apache.http.impl.nio.codecs.DefaultHttpRequestWriterFactory.(DefaultHttpRequestWriterFactory.java:57)
at 
org.apache.http.impl.nio.codecs.DefaultHttpRequestWriterFactory.(DefaultHttpRequestWriterFactory.java:47)
at 
org.apache.http.impl.nio.conn.ManagedNHttpClientConnectionFactory.(ManagedNHttpClientConnectionFactory.java:75)
at 
org.apache.http.impl.nio.conn.ManagedNHttpClientConnectionFactory.(ManagedNHttpClientConnectionFactory.java:83)
at 
org.apache.http.impl.nio.conn.ManagedNHttpClientConnectionFactory.(ManagedNHttpClientConnectionFactory.java:64)
at Main$.main(Main.scala:9)
at Main.main(Main.scala)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at 
sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at 
sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:497)
at 
org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:509)
at 
org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:403)
at 
org.apache.flink.client.program.ClusterClient.run(ClusterClient.java:322)
at 
org.apache.flink.client.CliFrontend.executeProgram(CliFrontend.java:774)
at 

[jira] [Commented] (FLINK-4535) ResourceManager registration with TaskExecutor

[ASF GitHub Bot (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLINK-4535?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15469471#comment-15469471
 ] 

ASF GitHub Bot commented on FLINK-4535:
---

Github user beyond1920 commented on a diff in the pull request:

https://github.com/apache/flink/pull/2451#discussion_r77757943
  
--- Diff: 
flink-runtime/src/main/java/org/apache/flink/runtime/rpc/resourcemanager/ResourceManager.java
 ---
@@ -138,22 +150,46 @@ public SlotAssignment requestSlot(SlotRequest 
slotRequest) {
 
 
/**
-*
-* @param resourceManagerLeaderId  The fencing token for the 
ResourceManager leader 
+* Register a taskExecutor at the resource manager
+* @param resourceManagerLeaderId  The fencing token for the 
ResourceManager leader
 * @param taskExecutorAddress  The address of the TaskExecutor that 
registers
 * @param resourceID   The resource ID of the TaskExecutor 
that registers
 *
 * @return The response by the ResourceManager.
 */
@RpcMethod
-   public org.apache.flink.runtime.rpc.registration.RegistrationResponse 
registerTaskExecutor(
-   UUID resourceManagerLeaderId,
-   String taskExecutorAddress,
-   ResourceID resourceID) {
+   public Future registerTaskExecutor(
+   final UUID resourceManagerLeaderId,
+   final String taskExecutorAddress,
+   final ResourceID resourceID) {
+
+   if(!leaderSessionID.equals(resourceManagerLeaderId)) {
+   log.warn("Discard registration from TaskExecutor {} at 
({}) because the expected leader session ID {} did not equal the received 
leader session ID  {}",
+   resourceID, taskExecutorAddress, 
leaderSessionID, resourceManagerLeaderId);
+   return Futures.failed(new 
UnmatchedLeaderSessionIDException(leaderSessionID, resourceManagerLeaderId));
--- End diff --

Do you prefer to send a Decline message under the condition?


> ResourceManager registration with TaskExecutor
> --
>
> Key: FLINK-4535
> URL: https://issues.apache.org/jira/browse/FLINK-4535
> Project: Flink
>  Issue Type: Sub-task
>  Components: Cluster Management
>Reporter: zhangjing
>Assignee: zhangjing
>
> When TaskExecutor register at ResourceManager, it takes the following 3 input 
> parameters:
> 1. resourceManagerLeaderId:  the fencing token for the ResourceManager leader 
> which is kept by taskExecutor who send the registration
> 2.  taskExecutorAddress: the address of taskExecutor
> 3. resourceID: The resource ID of the TaskExecutor that registers
> ResourceManager need to process the registration event based on the following 
> steps:
> 1. Check whether input resourceManagerLeaderId is as same as the current 
> leadershipSessionId of resourceManager. If not, it means that maybe two or 
> more resourceManager exists at the same time, and current resourceManager is 
> not the proper rm. so it  rejects or ignores the registration.
> 2. Check whether exists a valid taskExecutor at the giving address by 
> connecting to the address. Reject the registration from invalid address.
> 3. Check whether it is a duplicate registration by input resourceId, reject 
> the registration
> 4. Keep resourceID and taskExecutorGateway mapping relationships, And 
> optionally keep resourceID and container mapping relationships in yarn mode.
> 5. Create the connection between resourceManager and taskExecutor, and ensure 
> its healthy based on heartbeat rpc calls between rm and tm ?
> 6. Send registration successful ack to the taskExecutor.
> Discussion:
> Maybe we need import errorCode or several registration decline subclass to 
> distinguish the different causes of decline registration. 



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[GitHub] flink pull request #2451: [FLINK-4535] [cluster management] resourceManager ...

[beyond1920]

Github user beyond1920 commented on a diff in the pull request:

https://github.com/apache/flink/pull/2451#discussion_r77757943
  
--- Diff: 
flink-runtime/src/main/java/org/apache/flink/runtime/rpc/resourcemanager/ResourceManager.java
 ---
@@ -138,22 +150,46 @@ public SlotAssignment requestSlot(SlotRequest 
slotRequest) {
 
 
/**
-*
-* @param resourceManagerLeaderId  The fencing token for the 
ResourceManager leader 
+* Register a taskExecutor at the resource manager
+* @param resourceManagerLeaderId  The fencing token for the 
ResourceManager leader
 * @param taskExecutorAddress  The address of the TaskExecutor that 
registers
 * @param resourceID   The resource ID of the TaskExecutor 
that registers
 *
 * @return The response by the ResourceManager.
 */
@RpcMethod
-   public org.apache.flink.runtime.rpc.registration.RegistrationResponse 
registerTaskExecutor(
-   UUID resourceManagerLeaderId,
-   String taskExecutorAddress,
-   ResourceID resourceID) {
+   public Future registerTaskExecutor(
+   final UUID resourceManagerLeaderId,
+   final String taskExecutorAddress,
+   final ResourceID resourceID) {
+
+   if(!leaderSessionID.equals(resourceManagerLeaderId)) {
+   log.warn("Discard registration from TaskExecutor {} at 
({}) because the expected leader session ID {} did not equal the received 
leader session ID  {}",
+   resourceID, taskExecutorAddress, 
leaderSessionID, resourceManagerLeaderId);
+   return Futures.failed(new 
UnmatchedLeaderSessionIDException(leaderSessionID, resourceManagerLeaderId));
--- End diff --

Do you prefer to send a Decline message under the condition?


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[jira] [Commented] (FLINK-4565) Support for SQL IN operator

[Jark Wu (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLINK-4565?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15469402#comment-15469402
 ] 

Jark Wu commented on FLINK-4565:


I think what is confusing [~chobeat] is what's "IN" 's counterpart in DataSet 
transformation, right ?  Maybe we can use broadcast variables [1] to support 
this feature. In addition, [FLINK-4541] may be could implemented together. 

[1] 
https://ci.apache.org/projects/flink/flink-docs-master/dev/batch/index.html#broadcast-variables


> Support for SQL IN operator
> ---
>
> Key: FLINK-4565
> URL: https://issues.apache.org/jira/browse/FLINK-4565
> Project: Flink
>  Issue Type: Improvement
>  Components: Table API & SQL
>Reporter: Timo Walther
>
> It seems that Flink SQL supports the uncorrelated sub-query IN operator. But 
> it should also be available in the Table API and tested.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Updated] (FLINK-4547) when call connect method in AkkaRpcService using same address and same rpc gateway class, the returned gateways are equal with respect to equals and hashCode

[zhangjing (JIRA)]

 [ 
https://issues.apache.org/jira/browse/FLINK-4547?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

zhangjing updated FLINK-4547:
-
Summary: when call connect method in AkkaRpcService using same address and 
same rpc gateway class, the returned gateways are equal with respect to equals 
and hashCode  (was: Return same object when call connect method in 
AkkaRpcService using same address and same rpc gateway class)

> when call connect method in AkkaRpcService using same address and same rpc 
> gateway class, the returned gateways are equal with respect to equals and 
> hashCode
> -
>
> Key: FLINK-4547
> URL: https://issues.apache.org/jira/browse/FLINK-4547
> Project: Flink
>  Issue Type: Sub-task
>  Components: Distributed Coordination
>Reporter: zhangjing
>Assignee: zhangjing
>
> Now every time call connect method in AkkaRpcService class using same address 
> and same rpc gateway class, the return gateway object is totally different 
> with each other which equals and hashcode are not same. 
> Maybe it’s reasonable to have the same result (equals return true, and 
> hashcode is same) when using the same address and same Gateway class.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (FLINK-4547) Return same object when call connect method in AkkaRpcService using same address and same rpc gateway class

[ASF GitHub Bot (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLINK-4547?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15469345#comment-15469345
 ] 

ASF GitHub Bot commented on FLINK-4547:
---

Github user beyond1920 commented on a diff in the pull request:

https://github.com/apache/flink/pull/2455#discussion_r77754084
  
--- Diff: 
flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/AkkaInvocationHandler.java
 ---
@@ -189,7 +191,49 @@ public void stop() {
rpcEndpoint.tell(Processing.STOP, ActorRef.noSender());
}
 
-   // 

+   @Override
+   public boolean equals(Object o) {
+   if (this == o) {
+   return true;
+   }
+
+   if (o == null) {
+   return false;
+   }
+
+   if(Proxy.isProxyClass(o.getClass())) {
+   return o.equals(this);
+   }
--- End diff --

@tillrohrmann ,  It does not means that AkkaInvocationHandler be a proxy 
class. In fact it means if input parameter class is a proxy class, then return 
o.equals(this) result. Here is my reason.  when call connect method in 
AkkaRpcService, the returned gateway which is wrapped in Future is in fact a 
Proxy. When I call equals method to compare two gateway, the equals method of 
AkkaInvocationHandler will be called. But the input parameter is still another 
Gateway  which class is Proxy class instead of AkkaInvocationHandler. 


> Return same object when call connect method in AkkaRpcService using same 
> address and same rpc gateway class
> ---
>
> Key: FLINK-4547
> URL: https://issues.apache.org/jira/browse/FLINK-4547
> Project: Flink
>  Issue Type: Sub-task
>  Components: Distributed Coordination
>Reporter: zhangjing
>Assignee: zhangjing
>
> Now every time call connect method in AkkaRpcService class using same address 
> and same rpc gateway class, the return gateway object is totally different 
> with each other which equals and hashcode are not same. 
> Maybe it’s reasonable to have the same result (equals return true, and 
> hashcode is same) when using the same address and same Gateway class.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[GitHub] flink pull request #2455: [FLINK-4547] [cluster management] when call connec...

[beyond1920]

Github user beyond1920 commented on a diff in the pull request:

https://github.com/apache/flink/pull/2455#discussion_r77754084
  
--- Diff: 
flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/AkkaInvocationHandler.java
 ---
@@ -189,7 +191,49 @@ public void stop() {
rpcEndpoint.tell(Processing.STOP, ActorRef.noSender());
}
 
-   // 

+   @Override
+   public boolean equals(Object o) {
+   if (this == o) {
+   return true;
+   }
+
+   if (o == null) {
+   return false;
+   }
+
+   if(Proxy.isProxyClass(o.getClass())) {
+   return o.equals(this);
+   }
--- End diff --

@tillrohrmann ,  It does not means that AkkaInvocationHandler be a proxy 
class. In fact it means if input parameter class is a proxy class, then return 
o.equals(this) result. Here is my reason.  when call connect method in 
AkkaRpcService, the returned gateway which is wrapped in Future is in fact a 
Proxy. When I call equals method to compare two gateway, the equals method of 
AkkaInvocationHandler will be called. But the input parameter is still another 
Gateway  which class is Proxy class instead of AkkaInvocationHandler. 


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[GitHub] flink pull request #2455: [FLINK-4547] [cluster management] Return same obje...

[beyond1920]

Github user beyond1920 commented on a diff in the pull request:

https://github.com/apache/flink/pull/2455#discussion_r77753174
  
--- Diff: 
flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/AkkaInvocationHandler.java
 ---
@@ -189,7 +191,49 @@ public void stop() {
rpcEndpoint.tell(Processing.STOP, ActorRef.noSender());
}
 
-   // 

+   @Override
+   public boolean equals(Object o) {
+   if (this == o) {
+   return true;
+   }
+
+   if (o == null) {
+   return false;
+   }
+
+   if(Proxy.isProxyClass(o.getClass())) {
+   return o.equals(this);
+   }
--- End diff --

@StephanEwen , as till said, the subject of this pr is misleading. I means 
When call connect method in AkkaRpcService using same address and same rpc 
gateway class, the returned gateways are equals instead of return same gateway. 
I changed the subject.


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[jira] [Commented] (FLINK-4547) Return same object when call connect method in AkkaRpcService using same address and same rpc gateway class

[ASF GitHub Bot (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLINK-4547?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15469325#comment-15469325
 ] 

ASF GitHub Bot commented on FLINK-4547:
---

Github user beyond1920 commented on a diff in the pull request:

https://github.com/apache/flink/pull/2455#discussion_r77753174
  
--- Diff: 
flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/AkkaInvocationHandler.java
 ---
@@ -189,7 +191,49 @@ public void stop() {
rpcEndpoint.tell(Processing.STOP, ActorRef.noSender());
}
 
-   // 

+   @Override
+   public boolean equals(Object o) {
+   if (this == o) {
+   return true;
+   }
+
+   if (o == null) {
+   return false;
+   }
+
+   if(Proxy.isProxyClass(o.getClass())) {
+   return o.equals(this);
+   }
--- End diff --

@StephanEwen , as till said, the subject of this pr is misleading. I means 
When call connect method in AkkaRpcService using same address and same rpc 
gateway class, the returned gateways are equals instead of return same gateway. 
I changed the subject.


> Return same object when call connect method in AkkaRpcService using same 
> address and same rpc gateway class
> ---
>
> Key: FLINK-4547
> URL: https://issues.apache.org/jira/browse/FLINK-4547
> Project: Flink
>  Issue Type: Sub-task
>  Components: Distributed Coordination
>Reporter: zhangjing
>Assignee: zhangjing
>
> Now every time call connect method in AkkaRpcService class using same address 
> and same rpc gateway class, the return gateway object is totally different 
> with each other which equals and hashcode are not same. 
> Maybe it’s reasonable to have the same result (equals return true, and 
> hashcode is same) when using the same address and same Gateway class.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[GitHub] flink issue #2452: [Flink-4450] add a new module "flink-apache-storm" to sup...

[liuyuzhong]

Github user liuyuzhong commented on the issue:

https://github.com/apache/flink/pull/2452
  
@mxm 
Yes, all of this code are copy from old code adapted to Strom 1.0.0. So we 
close this pull request and wait for decision?  

Actually, I was submit a pull request for old code adpate to storm version 
to 1.0.0.  
https://github.com/apache/flink/pull/2439



---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[jira] [Commented] (FLINK-4586) NumberSequenceIterator and Accumulator threading issue

[Johannes (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLINK-4586?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15468729#comment-15468729
 ] 

Johannes commented on FLINK-4586:
-

The testcase also fails, when just using a plain collection and a rebalance to 
create some form of a parallel collection

{code}
fromCollection(1 to 100).rebalance() 
{code}

So it seems to be not specific to the NumberSequenceIterator.

So either the initialization of the accumulator is wrong in the sample code, or 
there is a deeper issue.

> NumberSequenceIterator and Accumulator threading issue
> --
>
> Key: FLINK-4586
> URL: https://issues.apache.org/jira/browse/FLINK-4586
> Project: Flink
>  Issue Type: Bug
>  Components: DataSet API
>Affects Versions: 1.1.2
>Reporter: Johannes
>Priority: Minor
> Attachments: FLINK4586Test.scala
>
>
> There is a strange problem when using the NumberSequenceIterator in 
> combination with an AverageAccumulator.
> It seems like the individual accumulators are reinitialized and overwrite 
> parts of intermediate solutions.
> The following scala snippit exemplifies the problem.
> Instead of printing the correct average, the result should be {{50.5}} but is 
> something completely different, like {{8.08}}, dependent on the number of 
> cores used.
> If the parallelism is set to {{1}} the result is correct, which indicates a 
> likely threading problem. 
> The problem occurs using the java and scala API.
> {code}
> env
>   .fromParallelCollection(new NumberSequenceIterator(1, 100))
>   .map(new RichMapFunction[Long, Long] {
>   var a : AverageAccumulator = _
>   override def map(value: Long): Long = {
> a.add(value)
> value
>   }
>   override def open(parameters: Configuration): Unit = {
> a = new AverageAccumulator
> getRuntimeContext.addAccumulator("test", a)
>   }
>   })
>   .reduce((a, b) => a + b)
>   .print()
> val lastJobExecutionResult: JobExecutionResult = env.getLastJobExecutionResult
> println(lastJobExecutionResult.getAccumulatorResult("test"))
> {code}



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Updated] (FLINK-4586) NumberSequenceIterator and Accumulator threading issue

[Johannes (JIRA)]

 [ 
https://issues.apache.org/jira/browse/FLINK-4586?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Johannes updated FLINK-4586:

Description: 
There is a strange problem when using the NumberSequenceIterator in combination 
with an AverageAccumulator.

It seems like the individual accumulators are reinitialized and overwrite parts 
of intermediate solutions.

The following scala snippit exemplifies the problem.

Instead of printing the correct average, the result should be {{50.5}} but is 
something completely different, like {{8.08}}, dependent on the number of cores 
used.

If the parallelism is set to {{1}} the result is correct, which indicates a 
likely threading problem. 

The problem occurs using the java and scala API.

{code}
env
  .fromParallelCollection(new NumberSequenceIterator(1, 100))
  .map(new RichMapFunction[Long, Long] {
var a : AverageAccumulator = _

override def map(value: Long): Long = {
  a.add(value)
  value
}

override def open(parameters: Configuration): Unit = {
  a = new AverageAccumulator
  getRuntimeContext.addAccumulator("test", a)
}
  })
  .reduce((a, b) => a + b)
  .print()


val lastJobExecutionResult: JobExecutionResult = env.getLastJobExecutionResult

println(lastJobExecutionResult.getAccumulatorResult("test"))
{code}

  was:
There is a strange problem when using the NumberSequenceIterator in combination 
with an AverageAccumulator.

It seems like the individual accumulators are reinitialized and overwrite parts 
of intermediate solution.

The following scala snippit exemplifies the problem.

Instead of printing the correct average, the result should be {{50.5}} but is 
something completely different, like {{8.08}}, dependent on the number of cores 
used.

If the parallelism is set to {{1}} the result is correct, which indicates a 
likely threading problem. 

The problem occurs using the java and scala API.

{code}
env
  .fromParallelCollection(new NumberSequenceIterator(1, 100))
  .map(new RichMapFunction[Long, Long] {
var a : AverageAccumulator = _

override def map(value: Long): Long = {
  a.add(value)
  value
}

override def open(parameters: Configuration): Unit = {
  a = new AverageAccumulator
  getRuntimeContext.addAccumulator("test", a)
}
  })
  .reduce((a, b) => a + b)
  .print()


val lastJobExecutionResult: JobExecutionResult = env.getLastJobExecutionResult

println(lastJobExecutionResult.getAccumulatorResult("test"))
{code}


> NumberSequenceIterator and Accumulator threading issue
> --
>
> Key: FLINK-4586
> URL: https://issues.apache.org/jira/browse/FLINK-4586
> Project: Flink
>  Issue Type: Bug
>  Components: DataSet API
>Affects Versions: 1.1.2
>Reporter: Johannes
>Priority: Minor
> Attachments: FLINK4586Test.scala
>
>
> There is a strange problem when using the NumberSequenceIterator in 
> combination with an AverageAccumulator.
> It seems like the individual accumulators are reinitialized and overwrite 
> parts of intermediate solutions.
> The following scala snippit exemplifies the problem.
> Instead of printing the correct average, the result should be {{50.5}} but is 
> something completely different, like {{8.08}}, dependent on the number of 
> cores used.
> If the parallelism is set to {{1}} the result is correct, which indicates a 
> likely threading problem. 
> The problem occurs using the java and scala API.
> {code}
> env
>   .fromParallelCollection(new NumberSequenceIterator(1, 100))
>   .map(new RichMapFunction[Long, Long] {
>   var a : AverageAccumulator = _
>   override def map(value: Long): Long = {
> a.add(value)
> value
>   }
>   override def open(parameters: Configuration): Unit = {
> a = new AverageAccumulator
> getRuntimeContext.addAccumulator("test", a)
>   }
>   })
>   .reduce((a, b) => a + b)
>   .print()
> val lastJobExecutionResult: JobExecutionResult = env.getLastJobExecutionResult
> println(lastJobExecutionResult.getAccumulatorResult("test"))
> {code}



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (FLINK-4410) Split checkpoint times into synchronous and asynchronous part

[Ivan Mushketyk (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLINK-4410?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15468721#comment-15468721
 ] 

Ivan Mushketyk commented on FLINK-4410:
---

Hi [~uce].

Just to make sure that I understand correctly what do you mean by synchronous 
and asynchronous parts. Do I understand correctly that they are:
* synchronous - time span between checkpoint is initiated and the moment when 
TriggerCheckpoint messages are sent
* asynchronous - time between all TriggerCheckpoint messages are sent and all 
replies are received



> Split checkpoint times into synchronous and asynchronous part
> -
>
> Key: FLINK-4410
> URL: https://issues.apache.org/jira/browse/FLINK-4410
> Project: Flink
>  Issue Type: Improvement
>  Components: Webfrontend
>Reporter: Ufuk Celebi
>Priority: Minor
>
> Checkpoint statistics contain the duration of a checkpoint. We should split 
> this time into the synchronous and asynchronous part. This will give more 
> insight into the inner workings of the checkpointing mechanism and help users 
> better understand what's going on.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Updated] (FLINK-4586) NumberSequenceIterator and Accumulator threading issue

[Johannes (JIRA)]

 [ 
https://issues.apache.org/jira/browse/FLINK-4586?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Johannes updated FLINK-4586:

Description: 
There is a strange problem when using the NumberSequenceIterator in combination 
with an AverageAccumulator.

It seems like the individual accumulators are reinitialized and overwrite parts 
of intermediate solution.

The following scala snippit exemplifies the problem.

Instead of printing the correct average, the result should be {{50.5}} but is 
something completely different, like {{8.08}}, dependent on the number of cores 
used.

If the parallelism is set to {{1}} the result is correct, which indicates a 
likely threading problem. 

The problem occurs using the java and scala API.

{code}
env
  .fromParallelCollection(new NumberSequenceIterator(1, 100))
  .map(new RichMapFunction[Long, Long] {
var a : AverageAccumulator = _

override def map(value: Long): Long = {
  a.add(value)
  value
}

override def open(parameters: Configuration): Unit = {
  a = new AverageAccumulator
  getRuntimeContext.addAccumulator("test", a)
}
  })
  .reduce((a, b) => a + b)
  .print()


val lastJobExecutionResult: JobExecutionResult = env.getLastJobExecutionResult

println(lastJobExecutionResult.getAccumulatorResult("test"))
{code}

  was:
There is a strange problem when using the NumberSequenceIterator in combination 
with an AverageAccumulator.

It seems like the individual accumulators are reinitialized and overwrite parts 
of intermediate solution.

The following scala snippit exemplifies the problem.
Instead of printing the correct average, the result should be {{50.5}} but is 
something completely different, like {{8.08}}, dependent on the number of cores 
used.
If the parallelism is set to {{1}} the result is correct, which seems like 
there is a problem with threading. The problem occurs using the java and scala 
API.

{code}
env
  .fromParallelCollection(new NumberSequenceIterator(1, 100))
  .map(new RichMapFunction[Long, Long] {
var a : AverageAccumulator = _

override def map(value: Long): Long = {
  a.add(value)
  value
}

override def open(parameters: Configuration): Unit = {
  a = new AverageAccumulator
  getRuntimeContext.addAccumulator("test", a)
}
  })
  .reduce((a, b) => a + b)
  .print()


val lastJobExecutionResult: JobExecutionResult = env.getLastJobExecutionResult

println(lastJobExecutionResult.getAccumulatorResult("test"))
{code}


> NumberSequenceIterator and Accumulator threading issue
> --
>
> Key: FLINK-4586
> URL: https://issues.apache.org/jira/browse/FLINK-4586
> Project: Flink
>  Issue Type: Bug
>  Components: DataSet API
>Affects Versions: 1.1.2
>Reporter: Johannes
>Priority: Minor
> Attachments: FLINK4586Test.scala
>
>
> There is a strange problem when using the NumberSequenceIterator in 
> combination with an AverageAccumulator.
> It seems like the individual accumulators are reinitialized and overwrite 
> parts of intermediate solution.
> The following scala snippit exemplifies the problem.
> Instead of printing the correct average, the result should be {{50.5}} but is 
> something completely different, like {{8.08}}, dependent on the number of 
> cores used.
> If the parallelism is set to {{1}} the result is correct, which indicates a 
> likely threading problem. 
> The problem occurs using the java and scala API.
> {code}
> env
>   .fromParallelCollection(new NumberSequenceIterator(1, 100))
>   .map(new RichMapFunction[Long, Long] {
>   var a : AverageAccumulator = _
>   override def map(value: Long): Long = {
> a.add(value)
> value
>   }
>   override def open(parameters: Configuration): Unit = {
> a = new AverageAccumulator
> getRuntimeContext.addAccumulator("test", a)
>   }
>   })
>   .reduce((a, b) => a + b)
>   .print()
> val lastJobExecutionResult: JobExecutionResult = env.getLastJobExecutionResult
> println(lastJobExecutionResult.getAccumulatorResult("test"))
> {code}



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Updated] (FLINK-4586) NumberSequenceIterator and Accumulator threading issue

[Johannes (JIRA)]

 [ 
https://issues.apache.org/jira/browse/FLINK-4586?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Johannes updated FLINK-4586:

Attachment: FLINK4586Test.scala

Complete Scala Testcase

> NumberSequenceIterator and Accumulator threading issue
> --
>
> Key: FLINK-4586
> URL: https://issues.apache.org/jira/browse/FLINK-4586
> Project: Flink
>  Issue Type: Bug
>  Components: DataSet API
>Affects Versions: 1.1.2
>Reporter: Johannes
>Priority: Minor
> Attachments: FLINK4586Test.scala
>
>
> There is a strange problem when using the NumberSequenceIterator in 
> combination with an AverageAccumulator.
> It seems like the individual accumulators are reinitialized and overwrite 
> parts of intermediate solution.
> The following scala snippit exemplifies the problem.
> Instead of printing the correct average, the result should be {{50.5}} but is 
> something completely different, like {{8.08}}, dependent on the number of 
> cores used.
> If the parallelism is set to {{1}} the result is correct, which seems like 
> there is a problem with threading. The problem occurs using the java and 
> scala API.
> {code}
> env
>   .fromParallelCollection(new NumberSequenceIterator(1, 100))
>   .map(new RichMapFunction[Long, Long] {
>   var a : AverageAccumulator = _
>   override def map(value: Long): Long = {
> a.add(value)
> value
>   }
>   override def open(parameters: Configuration): Unit = {
> a = new AverageAccumulator
> getRuntimeContext.addAccumulator("test", a)
>   }
>   })
>   .reduce((a, b) => a + b)
>   .print()
> val lastJobExecutionResult: JobExecutionResult = env.getLastJobExecutionResult
> println(lastJobExecutionResult.getAccumulatorResult("test"))
> {code}



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Created] (FLINK-4586) NumberSequenceIterator and Accumulator threading issue

[Johannes (JIRA)]

Johannes created FLINK-4586:
---

 Summary: NumberSequenceIterator and Accumulator threading issue
 Key: FLINK-4586
 URL: https://issues.apache.org/jira/browse/FLINK-4586
 Project: Flink
  Issue Type: Bug
  Components: DataSet API
Affects Versions: 1.1.2
Reporter: Johannes
Priority: Minor


There is a strange problem when using the NumberSequenceIterator in combination 
with an AverageAccumulator.

It seems like the individual accumulators are reinitialized and overwrite parts 
of intermediate solution.

The following scala snippit exemplifies the problem.
Instead of printing the correct average, the result should be {{50.5}} but is 
something completely different, like {{8.08}}, dependent on the number of cores 
used.
If the parallelism is set to {{1}} the result is correct, which seems like 
there is a problem with threading. The problem occurs using the java and scala 
API.

{code}
env
  .fromParallelCollection(new NumberSequenceIterator(1, 100))
  .map(new RichMapFunction[Long, Long] {
var a : AverageAccumulator = _

override def map(value: Long): Long = {
  a.add(value)
  value
}

override def open(parameters: Configuration): Unit = {
  a = new AverageAccumulator
  getRuntimeContext.addAccumulator("test", a)
}
  })
  .reduce((a, b) => a + b)
  .print()


val lastJobExecutionResult: JobExecutionResult = env.getLastJobExecutionResult

println(lastJobExecutionResult.getAccumulatorResult("test"))
{code}



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (FLINK-3930) Implement Service-Level Authorization

[ASF GitHub Bot (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLINK-3930?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15468585#comment-15468585
 ] 

ASF GitHub Bot commented on FLINK-3930:
---

Github user vijikarthi commented on the issue:

https://github.com/apache/flink/pull/2425
  
>
How is the secret transferred to the TaskManagers on YARN?

Cookie is transferred to TM container through container environment 
variable and further gets populated to in-memory Flink configuration instance. 
The secure cookie is vulnerable (so as the Keytab file) to the users who has 
access to the container local resource storage area and that's the limitation 
we may have to deal with.

>Is using the JobManagerMessages.getRequestBlobManagerSecureCookie() 
message always secure?

I believe it is safe since Akka endpoints are secured using the shared 
token (cookie) and for someone to access the cookie using the 
"JobManagerMessages.getRequestBlobManagerSecureCookie()", they should have been 
authenticated first.

>
Maybe it also makes sense to start adding a page into the internals 
documentation section, explaining how the secure cookie is implemented.

I am planning to add a separate page to internals documentation explaining 
how the shared secret stuff is implemented for various deployment options - 
Standalone, Yarn, Mesos (tbd)


> Implement Service-Level Authorization
> -
>
> Key: FLINK-3930
> URL: https://issues.apache.org/jira/browse/FLINK-3930
> Project: Flink
>  Issue Type: New Feature
>  Components: Security
>Reporter: Eron Wright 
>Assignee: Vijay Srinivasaraghavan
>  Labels: security
>   Original Estimate: 672h
>  Remaining Estimate: 672h
>
> _This issue is part of a series of improvements detailed in the [Secure Data 
> Access|https://docs.google.com/document/d/1-GQB6uVOyoaXGwtqwqLV8BHDxWiMO2WnVzBoJ8oPaAs/edit?usp=sharing]
>  design doc._
> Service-level authorization is the initial authorization mechanism to ensure 
> clients (or servers) connecting to the Flink cluster are authorized to do so. 
>   The purpose is to prevent a cluster from being used by an unauthorized 
> user, whether to execute jobs, disrupt cluster functionality, or gain access 
> to secrets stored within the cluster.
> Implement service-level authorization as described in the design doc.
> - Introduce a shared secret cookie
> - Enable Akka security cookie
> - Implement data transfer authentication
> - Secure the web dashboard



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[GitHub] flink issue #2425: FLINK-3930 Added shared secret based authorization for Fl...

[vijikarthi]

Github user vijikarthi commented on the issue:

https://github.com/apache/flink/pull/2425
  
>
How is the secret transferred to the TaskManagers on YARN?

Cookie is transferred to TM container through container environment 
variable and further gets populated to in-memory Flink configuration instance. 
The secure cookie is vulnerable (so as the Keytab file) to the users who has 
access to the container local resource storage area and that's the limitation 
we may have to deal with.

>Is using the JobManagerMessages.getRequestBlobManagerSecureCookie() 
message always secure?

I believe it is safe since Akka endpoints are secured using the shared 
token (cookie) and for someone to access the cookie using the 
"JobManagerMessages.getRequestBlobManagerSecureCookie()", they should have been 
authenticated first.

>
Maybe it also makes sense to start adding a page into the internals 
documentation section, explaining how the secure cookie is implemented.

I am planning to add a separate page to internals documentation explaining 
how the shared secret stuff is implemented for various deployment options - 
Standalone, Yarn, Mesos (tbd)


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[jira] [Commented] (FLINK-3709) [streaming] Graph event rates over time

[Ivan Mushketyk (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLINK-3709?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15468557#comment-15468557
 ] 

Ivan Mushketyk commented on FLINK-3709:
---

[~rmetzger] I could work on it, but I wonder where would be a good idea to 
store rate history?

> [streaming] Graph event rates over time
> ---
>
> Key: FLINK-3709
> URL: https://issues.apache.org/jira/browse/FLINK-3709
> Project: Flink
>  Issue Type: Improvement
>  Components: Webfrontend
>Reporter: Nick Dimiduk
>
> The streaming server job page displays bytes and records sent and received, 
> which answers the question "is data moving?" The next obvious question is "is 
> data moving over time?" That could be answered by a chart displaying 
> bytes/events rates. This would be a great chart to add to this display.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (FLINK-3427) Add watermark monitoring to JobManager web frontend

[Ivan Mushketyk (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLINK-3427?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15468538#comment-15468538
 ] 

Ivan Mushketyk commented on FLINK-3427:
---

[~rmetzger], I not half-way through the task yet, so it's okay if you take the 
task. I'll work on something different.
Just out of curiosity, how do you plan to implement it? 

> Add watermark monitoring to JobManager web frontend
> ---
>
> Key: FLINK-3427
> URL: https://issues.apache.org/jira/browse/FLINK-3427
> Project: Flink
>  Issue Type: Improvement
>  Components: Streaming, Webfrontend
>Reporter: Robert Metzger
>Assignee: Ivan Mushketyk
>
> Currently, its quite hard to figure out issues with the watermarks.
> I think we can improve the situation by reporting the following information 
> through the metrics system:
> - Report the current low watermark for each operator (this way, you can see 
> if one operator is preventing the watermarks to rise)
> - Report the number of events arrived after the low watermark (users can see 
> how accurate the watermarks are)



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (FLINK-3427) Add watermark monitoring to JobManager web frontend

[Robert Metzger (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLINK-3427?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15468531#comment-15468531
 ] 

Robert Metzger commented on FLINK-3427:
---

[~ivan.mushketyk] did you already start working on this issue? I'm asking 
because I have somebody who wanted to implement this soon, and we have some 
concrete plans how to do it.
If you're already half-way through with the implementation, you can get the 
task and finish it. However, if you did not start yet, I'm wondering whether it 
would be okay for you if I can take the task?

I'm sorry that I didn't see your comment three days ago earlier.

> Add watermark monitoring to JobManager web frontend
> ---
>
> Key: FLINK-3427
> URL: https://issues.apache.org/jira/browse/FLINK-3427
> Project: Flink
>  Issue Type: Improvement
>  Components: Streaming, Webfrontend
>Reporter: Robert Metzger
>Assignee: Ivan Mushketyk
>
> Currently, its quite hard to figure out issues with the watermarks.
> I think we can improve the situation by reporting the following information 
> through the metrics system:
> - Report the current low watermark for each operator (this way, you can see 
> if one operator is preventing the watermarks to rise)
> - Report the number of events arrived after the low watermark (users can see 
> how accurate the watermarks are)



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (FLINK-3427) Add watermark monitoring to JobManager web frontend

[Robert Metzger (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLINK-3427?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15468520#comment-15468520
 ] 

Robert Metzger commented on FLINK-3427:
---

Afaik there is already a gauge for getting the low watermark.

> Add watermark monitoring to JobManager web frontend
> ---
>
> Key: FLINK-3427
> URL: https://issues.apache.org/jira/browse/FLINK-3427
> Project: Flink
>  Issue Type: Improvement
>  Components: Streaming, Webfrontend
>Reporter: Robert Metzger
>Assignee: Ivan Mushketyk
>
> Currently, its quite hard to figure out issues with the watermarks.
> I think we can improve the situation by reporting the following information 
> through the metrics system:
> - Report the current low watermark for each operator (this way, you can see 
> if one operator is preventing the watermarks to rise)
> - Report the number of events arrived after the low watermark (users can see 
> how accurate the watermarks are)



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Updated] (FLINK-4585) Fix broken links in index.md

[Alexander Pivovarov (JIRA)]

 [ 
https://issues.apache.org/jira/browse/FLINK-4585?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Alexander Pivovarov updated FLINK-4585:
---
Description: 
The following links are broken

DataSet API
link: 
http://ci.apache.org/projects/flink/flink-docs-master/apis/batch/index.html
correct link: 
https://ci.apache.org/projects/flink/flink-docs-master/dev/batch/index.html

Table API
link: 
http://ci.apache.org/projects/flink/flink-docs-master/apis/batch/libs/table.html
correct link: 
https://ci.apache.org/projects/flink/flink-docs-master/dev/table_api.html

Gelly
link: 
http://ci.apache.org/projects/flink/flink-docs-master/apis/batch/libs/gelly.html
correct link: 
https://ci.apache.org/projects/flink/flink-docs-master/dev/libs/gelly/index.html

The following links show "Page 'X' Has Moved to" for 1-2 sec  and then redirect 
to another page

DataStream API
link: 
http://ci.apache.org/projects/flink/flink-docs-master/apis/streaming/index.html
redirects-to: 
https://ci.apache.org/projects/flink/flink-docs-master/dev/datastream_api.html

programming guide
link: 
http://ci.apache.org/projects/flink/flink-docs-master/apis/programming_guide.html
redirects-to DataSet API: 
https://ci.apache.org/projects/flink/flink-docs-master/dev/batch/index.html
probably it should be "Basic API Concepts" 
https://ci.apache.org/projects/flink/flink-docs-master/dev/api_concepts.html
or Quick Start - 
https://ci.apache.org/projects/flink/flink-docs-master/quickstart/setup_quickstart.html

CEP
link: 
http://ci.apache.org/projects/flink/flink-docs-master/apis/streaming/libs/cep.html
redirects-to: 
https://ci.apache.org/projects/flink/flink-docs-master/dev/libs/cep.html

ML
link: 
http://ci.apache.org/projects/flink/flink-docs-master/apis/batch/libs/ml/index.html
redirects-to: 
https://ci.apache.org/projects/flink/flink-docs-master/dev/libs/ml/index.html

  was:
The following links are broken

DataSet API
link: 
http://ci.apache.org/projects/flink/flink-docs-master/apis/batch/index.html
correct link: 
https://ci.apache.org/projects/flink/flink-docs-master/dev/batch/index.html

Table API
link: 
http://ci.apache.org/projects/flink/flink-docs-master/apis/batch/libs/table.html
correct link: 
https://ci.apache.org/projects/flink/flink-docs-master/dev/table_api.html

Gelly
link: 
http://ci.apache.org/projects/flink/flink-docs-master/apis/batch/libs/gelly.html
correct link: 
https://ci.apache.org/projects/flink/flink-docs-master/dev/libs/gelly/index.html

The following links show "Page 'X' Has Moved to" for 1-2 sec  and then redirect 
to another page

DataStream API
link: 
http://ci.apache.org/projects/flink/flink-docs-master/apis/streaming/index.html
redirects-to: 
https://ci.apache.org/projects/flink/flink-docs-master/dev/datastream_api.html

programming guide
link: 
http://ci.apache.org/projects/flink/flink-docs-master/apis/programming_guide.html
redirects-to DataStream API: 
https://ci.apache.org/projects/flink/flink-docs-master/dev/batch/index.html
probably it should be "Basic API Concepts" 
https://ci.apache.org/projects/flink/flink-docs-master/dev/api_concepts.html
or Quick Start - 
https://ci.apache.org/projects/flink/flink-docs-master/quickstart/setup_quickstart.html

CEP
link: 
http://ci.apache.org/projects/flink/flink-docs-master/apis/streaming/libs/cep.html
redirects-to: 
https://ci.apache.org/projects/flink/flink-docs-master/dev/libs/cep.html

ML
link: 
http://ci.apache.org/projects/flink/flink-docs-master/apis/batch/libs/ml/index.html
redirects-to: 
https://ci.apache.org/projects/flink/flink-docs-master/dev/libs/ml/index.html


> Fix broken links in index.md
> 
>
> Key: FLINK-4585
> URL: https://issues.apache.org/jira/browse/FLINK-4585
> Project: Flink
>  Issue Type: Bug
>  Components: Project Website
>Reporter: Alexander Pivovarov
>Priority: Minor
>
> The following links are broken
> DataSet API
> link: 
> http://ci.apache.org/projects/flink/flink-docs-master/apis/batch/index.html
> correct link: 
> https://ci.apache.org/projects/flink/flink-docs-master/dev/batch/index.html
> Table API
> link: 
> http://ci.apache.org/projects/flink/flink-docs-master/apis/batch/libs/table.html
> correct link: 
> https://ci.apache.org/projects/flink/flink-docs-master/dev/table_api.html
> Gelly
> link: 
> http://ci.apache.org/projects/flink/flink-docs-master/apis/batch/libs/gelly.html
> correct link: 
> https://ci.apache.org/projects/flink/flink-docs-master/dev/libs/gelly/index.html
> The following links show "Page 'X' Has Moved to" for 1-2 sec  and then 
> redirect to another page
> DataStream API
> link: 
> http://ci.apache.org/projects/flink/flink-docs-master/apis/streaming/index.html
> redirects-to: 
> https://ci.apache.org/projects/flink/flink-docs-master/dev/datastream_api.html
> programming guide
> link: 
> 

[jira] [Commented] (FLINK-2254) Add Bipartite Graph Support for Gelly

[Ivan Mushketyk (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLINK-2254?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15468506#comment-15468506
 ] 

Ivan Mushketyk commented on FLINK-2254:
---

Hi Greg. Thank you for your comment.
Could you please elaborate why there is a need for a BipartiateEdge class? Will 
vertices DataSets have different id types?
I would like to read [~vkalavri] feedback, but I think there are pros and cons 
of both solutions. Common hierarchy can help to reuse some code, but with 
separate classes we can have cleaner interfaces.

I we will go with different classes, we can add methods:
topProjection() and bottomProjection() that will convert bipartite graph into a 
one-mode Graph to bridge the gap between two types. 

> Add Bipartite Graph Support for Gelly
> -
>
> Key: FLINK-2254
> URL: https://issues.apache.org/jira/browse/FLINK-2254
> Project: Flink
>  Issue Type: New Feature
>  Components: Gelly
>Affects Versions: 0.10.0
>Reporter: Andra Lungu
>Assignee: Ivan Mushketyk
>  Labels: requires-design-doc
>
> A bipartite graph is a graph for which the set of vertices can be divided 
> into two disjoint sets such that each edge having a source vertex in the 
> first set, will have a target vertex in the second set. We would like to 
> support efficient operations for this type of graphs along with a set of 
> metrics(http://jponnela.com/web_documents/twomode.pdf). 



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (FLINK-3042) Define a way to let types create their own TypeInformation

[ASF GitHub Bot (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLINK-3042?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15468507#comment-15468507
 ] 

ASF GitHub Bot commented on FLINK-3042:
---

Github user greghogan commented on a diff in the pull request:

https://github.com/apache/flink/pull/2337#discussion_r77714875
  
--- Diff: 
flink-core/src/main/java/org/apache/flink/api/common/typeinfo/TypeInformation.java
 ---
@@ -122,14 +123,25 @@
public abstract Class getTypeClass();
 
/**
-* Returns the generic parameters of this type.
+* Optional method for giving Flink's type extraction system 
information about the mapping
+* of a generic type parameter to the type information of a subtype. 
This information is necessary
+* in cases where type information should be deduced from an input type.
 *
-* @return The list of generic parameters. This list can be empty.
+* For instance, a method for a {@link Tuple2} would look like this:
+* 
+* Map m = new HashMap();
+* m.put("T0", this.getTypeAt(0));
+* m.put("T1", this.getTypeAt(1));
+* return m;
+* 
+*
+* @return map of inferred subtypes; it must not contain all generic 
parameters as key;
--- End diff --

What is meant by "it must not contain all generic parameters as key"?


> Define a way to let types create their own TypeInformation
> --
>
> Key: FLINK-3042
> URL: https://issues.apache.org/jira/browse/FLINK-3042
> Project: Flink
>  Issue Type: Improvement
>  Components: Core
>Affects Versions: 0.10.0
>Reporter: Stephan Ewen
>Assignee: Timo Walther
> Fix For: 1.0.0
>
>
> Currently, introducing new Types that should have specific TypeInformation 
> requires
>   - Either integration with the TypeExtractor
>   - Or manually constructing the TypeInformation (potentially at every place) 
> and using type hints everywhere.
> I propose to add a way to allow classes to create their own TypeInformation 
> (like a static method "createTypeInfo()").
> To support generic nested types (like Optional / Either), the type extractor 
> would provide a Map of what generic variables map to what types (deduced from 
> the input). The class can use that to create the correct nested 
> TypeInformation (possibly by calling the TypeExtractor again, passing the Map 
> of generic bindings).



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[GitHub] flink pull request #2337: [FLINK-3042] [FLINK-3060] [types] Define a way to ...

[greghogan]

Github user greghogan commented on a diff in the pull request:

https://github.com/apache/flink/pull/2337#discussion_r77714875
  
--- Diff: 
flink-core/src/main/java/org/apache/flink/api/common/typeinfo/TypeInformation.java
 ---
@@ -122,14 +123,25 @@
public abstract Class getTypeClass();
 
/**
-* Returns the generic parameters of this type.
+* Optional method for giving Flink's type extraction system 
information about the mapping
+* of a generic type parameter to the type information of a subtype. 
This information is necessary
+* in cases where type information should be deduced from an input type.
 *
-* @return The list of generic parameters. This list can be empty.
+* For instance, a method for a {@link Tuple2} would look like this:
+* 
+* Map m = new HashMap();
+* m.put("T0", this.getTypeAt(0));
+* m.put("T1", this.getTypeAt(1));
+* return m;
+* 
+*
+* @return map of inferred subtypes; it must not contain all generic 
parameters as key;
--- End diff --

What is meant by "it must not contain all generic parameters as key"?


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[jira] [Commented] (FLINK-3930) Implement Service-Level Authorization

[ASF GitHub Bot (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLINK-3930?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15468482#comment-15468482
 ] 

ASF GitHub Bot commented on FLINK-3930:
---

Github user rmetzger commented on the issue:

https://github.com/apache/flink/pull/2425
  
Thank you for addressing my comments.

I've looked into the design document [1] again and some details are missing 
there. In particular its not clearly specified how and where the shared secrets 
are created and transferred in the cluster.
How is the secret transferred to the TaskManagers on YARN?
Is the using the  `JobManagerMessages.getRequestBlobManagerSecureCookie()` 
message always secure?
I think its important for security-related changes to first have a clear, 
documented security concept, and then an implementation that matches the 
documentation.
Maybe it also makes sense to start adding a page into the internals 
documentation section, explaining how the secure cookie is implemented.

@StephanEwen can you maybe add your thoughts here?

[1] 
https://docs.google.com/document/d/1-GQB6uVOyoaXGwtqwqLV8BHDxWiMO2WnVzBoJ8oPaAs/edit


> Implement Service-Level Authorization
> -
>
> Key: FLINK-3930
> URL: https://issues.apache.org/jira/browse/FLINK-3930
> Project: Flink
>  Issue Type: New Feature
>  Components: Security
>Reporter: Eron Wright 
>Assignee: Vijay Srinivasaraghavan
>  Labels: security
>   Original Estimate: 672h
>  Remaining Estimate: 672h
>
> _This issue is part of a series of improvements detailed in the [Secure Data 
> Access|https://docs.google.com/document/d/1-GQB6uVOyoaXGwtqwqLV8BHDxWiMO2WnVzBoJ8oPaAs/edit?usp=sharing]
>  design doc._
> Service-level authorization is the initial authorization mechanism to ensure 
> clients (or servers) connecting to the Flink cluster are authorized to do so. 
>   The purpose is to prevent a cluster from being used by an unauthorized 
> user, whether to execute jobs, disrupt cluster functionality, or gain access 
> to secrets stored within the cluster.
> Implement service-level authorization as described in the design doc.
> - Introduce a shared secret cookie
> - Enable Akka security cookie
> - Implement data transfer authentication
> - Secure the web dashboard



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[GitHub] flink issue #2425: FLINK-3930 Added shared secret based authorization for Fl...

[rmetzger]

Github user rmetzger commented on the issue:

https://github.com/apache/flink/pull/2425
  
Thank you for addressing my comments.

I've looked into the design document [1] again and some details are missing 
there. In particular its not clearly specified how and where the shared secrets 
are created and transferred in the cluster.
How is the secret transferred to the TaskManagers on YARN?
Is the using the  `JobManagerMessages.getRequestBlobManagerSecureCookie()` 
message always secure?
I think its important for security-related changes to first have a clear, 
documented security concept, and then an implementation that matches the 
documentation.
Maybe it also makes sense to start adding a page into the internals 
documentation section, explaining how the secure cookie is implemented.

@StephanEwen can you maybe add your thoughts here?

[1] 
https://docs.google.com/document/d/1-GQB6uVOyoaXGwtqwqLV8BHDxWiMO2WnVzBoJ8oPaAs/edit


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[jira] [Updated] (FLINK-4585) Fix broken links in index.md

[Alexander Pivovarov (JIRA)]

 [ 
https://issues.apache.org/jira/browse/FLINK-4585?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Alexander Pivovarov updated FLINK-4585:
---
Summary: Fix broken links in index.md  (was: Fix broken links on 
flink.apache.org)

> Fix broken links in index.md
> 
>
> Key: FLINK-4585
> URL: https://issues.apache.org/jira/browse/FLINK-4585
> Project: Flink
>  Issue Type: Bug
>  Components: Project Website
>Reporter: Alexander Pivovarov
>Priority: Minor
>
> The following links are broken
> DataSet API
> link: 
> http://ci.apache.org/projects/flink/flink-docs-master/apis/batch/index.html
> correct link: 
> https://ci.apache.org/projects/flink/flink-docs-master/dev/batch/index.html
> Table API
> link: 
> http://ci.apache.org/projects/flink/flink-docs-master/apis/batch/libs/table.html
> correct link: 
> https://ci.apache.org/projects/flink/flink-docs-master/dev/table_api.html
> Gelly
> link: 
> http://ci.apache.org/projects/flink/flink-docs-master/apis/batch/libs/gelly.html
> correct link: 
> https://ci.apache.org/projects/flink/flink-docs-master/dev/libs/gelly/index.html
> The following links show "Page 'X' Has Moved to" for 1-2 sec  and then 
> redirect to another page
> DataStream API
> link: 
> http://ci.apache.org/projects/flink/flink-docs-master/apis/streaming/index.html
> redirects-to: 
> https://ci.apache.org/projects/flink/flink-docs-master/dev/datastream_api.html
> programming guide
> link: 
> http://ci.apache.org/projects/flink/flink-docs-master/apis/programming_guide.html
> redirects-to DataStream API: 
> https://ci.apache.org/projects/flink/flink-docs-master/dev/batch/index.html
> probably it should be "Basic API Concepts" 
> https://ci.apache.org/projects/flink/flink-docs-master/dev/api_concepts.html
> or Quick Start - 
> https://ci.apache.org/projects/flink/flink-docs-master/quickstart/setup_quickstart.html
> CEP
> link: 
> http://ci.apache.org/projects/flink/flink-docs-master/apis/streaming/libs/cep.html
> redirects-to: 
> https://ci.apache.org/projects/flink/flink-docs-master/dev/libs/cep.html
> ML
> link: 
> http://ci.apache.org/projects/flink/flink-docs-master/apis/batch/libs/ml/index.html
> redirects-to: 
> https://ci.apache.org/projects/flink/flink-docs-master/dev/libs/ml/index.html



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Updated] (FLINK-4585) Fix broken links on flink.apache.org

[Alexander Pivovarov (JIRA)]

 [ 
https://issues.apache.org/jira/browse/FLINK-4585?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Alexander Pivovarov updated FLINK-4585:
---
Description: 
The following links are broken

DataSet API
link: 
http://ci.apache.org/projects/flink/flink-docs-master/apis/batch/index.html
correct link: 
https://ci.apache.org/projects/flink/flink-docs-master/dev/batch/index.html

Table API
link: 
http://ci.apache.org/projects/flink/flink-docs-master/apis/batch/libs/table.html
correct link: 
https://ci.apache.org/projects/flink/flink-docs-master/dev/table_api.html

Gelly
link: 
http://ci.apache.org/projects/flink/flink-docs-master/apis/batch/libs/gelly.html
correct link: 
https://ci.apache.org/projects/flink/flink-docs-master/dev/libs/gelly/index.html

The following links show "Page 'X' Has Moved to" for 1-2 sec  and then redirect 
to another page

DataStream API
link: 
http://ci.apache.org/projects/flink/flink-docs-master/apis/streaming/index.html
redirects-to: 
https://ci.apache.org/projects/flink/flink-docs-master/dev/datastream_api.html

programming guide
link: 
http://ci.apache.org/projects/flink/flink-docs-master/apis/programming_guide.html
redirects-to DataStream API: 
https://ci.apache.org/projects/flink/flink-docs-master/dev/batch/index.html
probably it should be "Basic API Concepts" 
https://ci.apache.org/projects/flink/flink-docs-master/dev/api_concepts.html
or Quick Start - 
https://ci.apache.org/projects/flink/flink-docs-master/quickstart/setup_quickstart.html

CEP
link: 
http://ci.apache.org/projects/flink/flink-docs-master/apis/streaming/libs/cep.html
redirects-to: 
https://ci.apache.org/projects/flink/flink-docs-master/dev/libs/cep.html

ML
link: 
http://ci.apache.org/projects/flink/flink-docs-master/apis/batch/libs/ml/index.html
redirects-to: 
https://ci.apache.org/projects/flink/flink-docs-master/dev/libs/ml/index.html

  was:
The following links are broken

DataSet API
link: 
http://ci.apache.org/projects/flink/flink-docs-master/apis/batch/index.html
correct link: 
https://ci.apache.org/projects/flink/flink-docs-master/dev/batch/index.html

Table API
link: 
http://ci.apache.org/projects/flink/flink-docs-master/apis/batch/libs/table.html
correct link: 
https://ci.apache.org/projects/flink/flink-docs-master/dev/table_api.html


The following links show "Page 'X' Has Moved to" for 1-2 sec  and then redirect 
to another page

DataStream API
link: 
http://ci.apache.org/projects/flink/flink-docs-master/apis/streaming/index.html
redirects-to: 
https://ci.apache.org/projects/flink/flink-docs-master/dev/datastream_api.html

programming guide
link: 
http://ci.apache.org/projects/flink/flink-docs-master/apis/programming_guide.html
redirects-to DataStream API: 
https://ci.apache.org/projects/flink/flink-docs-master/dev/batch/index.html
probably it should be "Basic API Concepts" 
https://ci.apache.org/projects/flink/flink-docs-master/dev/api_concepts.html
or Quick Start - 
https://ci.apache.org/projects/flink/flink-docs-master/quickstart/setup_quickstart.html


> Fix broken links on flink.apache.org
> 
>
> Key: FLINK-4585
> URL: https://issues.apache.org/jira/browse/FLINK-4585
> Project: Flink
>  Issue Type: Bug
>  Components: Project Website
>Reporter: Alexander Pivovarov
>Priority: Minor
>
> The following links are broken
> DataSet API
> link: 
> http://ci.apache.org/projects/flink/flink-docs-master/apis/batch/index.html
> correct link: 
> https://ci.apache.org/projects/flink/flink-docs-master/dev/batch/index.html
> Table API
> link: 
> http://ci.apache.org/projects/flink/flink-docs-master/apis/batch/libs/table.html
> correct link: 
> https://ci.apache.org/projects/flink/flink-docs-master/dev/table_api.html
> Gelly
> link: 
> http://ci.apache.org/projects/flink/flink-docs-master/apis/batch/libs/gelly.html
> correct link: 
> https://ci.apache.org/projects/flink/flink-docs-master/dev/libs/gelly/index.html
> The following links show "Page 'X' Has Moved to" for 1-2 sec  and then 
> redirect to another page
> DataStream API
> link: 
> http://ci.apache.org/projects/flink/flink-docs-master/apis/streaming/index.html
> redirects-to: 
> https://ci.apache.org/projects/flink/flink-docs-master/dev/datastream_api.html
> programming guide
> link: 
> http://ci.apache.org/projects/flink/flink-docs-master/apis/programming_guide.html
> redirects-to DataStream API: 
> https://ci.apache.org/projects/flink/flink-docs-master/dev/batch/index.html
> probably it should be "Basic API Concepts" 
> https://ci.apache.org/projects/flink/flink-docs-master/dev/api_concepts.html
> or Quick Start - 
> https://ci.apache.org/projects/flink/flink-docs-master/quickstart/setup_quickstart.html
> CEP
> link: 
> http://ci.apache.org/projects/flink/flink-docs-master/apis/streaming/libs/cep.html
> redirects-to: 
> 

[jira] [Commented] (FLINK-4585) Fix broken links on flink.apache.org

[ASF GitHub Bot (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLINK-4585?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15468416#comment-15468416
 ] 

ASF GitHub Bot commented on FLINK-4585:
---

Github user apivovarov commented on the issue:

https://github.com/apache/flink/pull/2471
  
https://issues.apache.org/jira/browse/FLINK-4585
I'll submit a PR to flink-web


> Fix broken links on flink.apache.org
> 
>
> Key: FLINK-4585
> URL: https://issues.apache.org/jira/browse/FLINK-4585
> Project: Flink
>  Issue Type: Bug
>  Components: Project Website
>Reporter: Alexander Pivovarov
>Priority: Minor
>
> The following links are broken
> DataSet API
> link: 
> http://ci.apache.org/projects/flink/flink-docs-master/apis/batch/index.html
> correct link: 
> https://ci.apache.org/projects/flink/flink-docs-master/dev/batch/index.html
> Table API
> link: 
> http://ci.apache.org/projects/flink/flink-docs-master/apis/batch/libs/table.html
> correct link: 
> https://ci.apache.org/projects/flink/flink-docs-master/dev/table_api.html
> The following links show "Page 'X' Has Moved to" for 1-2 sec  and then 
> redirect to another page
> DataStream API
> link: 
> http://ci.apache.org/projects/flink/flink-docs-master/apis/streaming/index.html
> redirects-to: 
> https://ci.apache.org/projects/flink/flink-docs-master/dev/datastream_api.html
> programming guide
> link: 
> http://ci.apache.org/projects/flink/flink-docs-master/apis/programming_guide.html
> redirects-to DataStream API: 
> https://ci.apache.org/projects/flink/flink-docs-master/dev/batch/index.html
> probably it should be "Basic API Concepts" 
> https://ci.apache.org/projects/flink/flink-docs-master/dev/api_concepts.html
> or Quick Start - 
> https://ci.apache.org/projects/flink/flink-docs-master/quickstart/setup_quickstart.html



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[GitHub] flink pull request #2471: Broken links on website

[apivovarov]

Github user apivovarov closed the pull request at:

https://github.com/apache/flink/pull/2471


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[GitHub] flink issue #2471: Broken links on website

[apivovarov]

Github user apivovarov commented on the issue:

https://github.com/apache/flink/pull/2471
  
https://issues.apache.org/jira/browse/FLINK-4585
I'll submit a PR to flink-web


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[jira] [Updated] (FLINK-4585) Fix broken links on flink.apache.org

[Alexander Pivovarov (JIRA)]

 [ 
https://issues.apache.org/jira/browse/FLINK-4585?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Alexander Pivovarov updated FLINK-4585:
---
Description: 
The following links are broken

DataSet API
link: 
http://ci.apache.org/projects/flink/flink-docs-master/apis/batch/index.html
correct link: 
https://ci.apache.org/projects/flink/flink-docs-master/dev/batch/index.html

Table API
link: 
http://ci.apache.org/projects/flink/flink-docs-master/apis/batch/libs/table.html
correct link: 
https://ci.apache.org/projects/flink/flink-docs-master/dev/table_api.html


The following links show "Page 'X' Has Moved to" for 1-2 sec  and then redirect 
to another page

DataStream API
link: 
http://ci.apache.org/projects/flink/flink-docs-master/apis/streaming/index.html
redirects-to: 
https://ci.apache.org/projects/flink/flink-docs-master/dev/datastream_api.html

programming guide
link: 
http://ci.apache.org/projects/flink/flink-docs-master/apis/programming_guide.html
redirects-to DataStream API: 
https://ci.apache.org/projects/flink/flink-docs-master/dev/batch/index.html
probably it should be "Basic API Concepts" 
https://ci.apache.org/projects/flink/flink-docs-master/dev/api_concepts.html
or Quick Start - 
https://ci.apache.org/projects/flink/flink-docs-master/quickstart/setup_quickstart.html

  was:
The following links are broken

DataSet API
link: 
http://ci.apache.org/projects/flink/flink-docs-master/apis/batch/index.html
correct link: 
https://ci.apache.org/projects/flink/flink-docs-master/dev/batch/index.html

Table API
link: 
http://ci.apache.org/projects/flink/flink-docs-master/apis/batch/libs/table.html
correct link: 
https://ci.apache.org/projects/flink/flink-docs-master/dev/table_api.html


The following links show "Page 'X' Has Moved to" for 1-2 sec  and then redirect 
to correct page

DataStream API
link: 
http://ci.apache.org/projects/flink/flink-docs-master/apis/streaming/index.html
redirects-to: 
https://ci.apache.org/projects/flink/flink-docs-master/dev/datastream_api.html

programming guide
link: 
http://ci.apache.org/projects/flink/flink-docs-master/apis/programming_guide.html
redirects-to DataStream API: 
https://ci.apache.org/projects/flink/flink-docs-master/dev/batch/index.html
probably it should be "Basic API Concepts" 
https://ci.apache.org/projects/flink/flink-docs-master/dev/api_concepts.html
or Quick Start - 
https://ci.apache.org/projects/flink/flink-docs-master/quickstart/setup_quickstart.html


> Fix broken links on flink.apache.org
> 
>
> Key: FLINK-4585
> URL: https://issues.apache.org/jira/browse/FLINK-4585
> Project: Flink
>  Issue Type: Bug
>  Components: Project Website
>Reporter: Alexander Pivovarov
>Priority: Minor
>
> The following links are broken
> DataSet API
> link: 
> http://ci.apache.org/projects/flink/flink-docs-master/apis/batch/index.html
> correct link: 
> https://ci.apache.org/projects/flink/flink-docs-master/dev/batch/index.html
> Table API
> link: 
> http://ci.apache.org/projects/flink/flink-docs-master/apis/batch/libs/table.html
> correct link: 
> https://ci.apache.org/projects/flink/flink-docs-master/dev/table_api.html
> The following links show "Page 'X' Has Moved to" for 1-2 sec  and then 
> redirect to another page
> DataStream API
> link: 
> http://ci.apache.org/projects/flink/flink-docs-master/apis/streaming/index.html
> redirects-to: 
> https://ci.apache.org/projects/flink/flink-docs-master/dev/datastream_api.html
> programming guide
> link: 
> http://ci.apache.org/projects/flink/flink-docs-master/apis/programming_guide.html
> redirects-to DataStream API: 
> https://ci.apache.org/projects/flink/flink-docs-master/dev/batch/index.html
> probably it should be "Basic API Concepts" 
> https://ci.apache.org/projects/flink/flink-docs-master/dev/api_concepts.html
> or Quick Start - 
> https://ci.apache.org/projects/flink/flink-docs-master/quickstart/setup_quickstart.html



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Updated] (FLINK-4585) Fix broken links on flink.apache.org

[Alexander Pivovarov (JIRA)]

 [ 
https://issues.apache.org/jira/browse/FLINK-4585?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Alexander Pivovarov updated FLINK-4585:
---
Description: 
The following links are broken

DataSet API
link: 
http://ci.apache.org/projects/flink/flink-docs-master/apis/batch/index.html
correct link: 
https://ci.apache.org/projects/flink/flink-docs-master/dev/batch/index.html

Table API
link: 
http://ci.apache.org/projects/flink/flink-docs-master/apis/batch/libs/table.html
correct link: 
https://ci.apache.org/projects/flink/flink-docs-master/dev/table_api.html


The following links show "Page 'X' Has Moved to" for 1-2 sec  and then redirect 
to correct page

DataStream API
link: 
http://ci.apache.org/projects/flink/flink-docs-master/apis/streaming/index.html
redirects-to: 
https://ci.apache.org/projects/flink/flink-docs-master/dev/datastream_api.html

programming guide
link: 
http://ci.apache.org/projects/flink/flink-docs-master/apis/programming_guide.html
redirects-to DataStream API: 
https://ci.apache.org/projects/flink/flink-docs-master/dev/batch/index.html
probably it should be "Basic API Concepts" 
https://ci.apache.org/projects/flink/flink-docs-master/dev/api_concepts.html
or Quick Start - 
https://ci.apache.org/projects/flink/flink-docs-master/quickstart/setup_quickstart.html

  was:
The following links are broken
link: 
http://ci.apache.org/projects/flink/flink-docs-master/apis/batch/index.html
correct link: 
https://ci.apache.org/projects/flink/flink-docs-master/dev/batch/index.html

link: 
http://ci.apache.org/projects/flink/flink-docs-master/apis/batch/libs/table.html
correct link: 
https://ci.apache.org/projects/flink/flink-docs-master/dev/table_api.html


The following links show "Page 'X' Has Moved to" for 1-2 sec  and then redirect 
to correct page
link: 
http://ci.apache.org/projects/flink/flink-docs-master/apis/streaming/index.html
redirect-to: 
https://ci.apache.org/projects/flink/flink-docs-master/dev/datastream_api.html

link: 
http://ci.apache.org/projects/flink/flink-docs-master/apis/programming_guide.html
redirect-to: 
https://ci.apache.org/projects/flink/flink-docs-master/dev/batch/index.html



> Fix broken links on flink.apache.org
> 
>
> Key: FLINK-4585
> URL: https://issues.apache.org/jira/browse/FLINK-4585
> Project: Flink
>  Issue Type: Bug
>  Components: Project Website
>Reporter: Alexander Pivovarov
>Priority: Minor
>
> The following links are broken
> DataSet API
> link: 
> http://ci.apache.org/projects/flink/flink-docs-master/apis/batch/index.html
> correct link: 
> https://ci.apache.org/projects/flink/flink-docs-master/dev/batch/index.html
> Table API
> link: 
> http://ci.apache.org/projects/flink/flink-docs-master/apis/batch/libs/table.html
> correct link: 
> https://ci.apache.org/projects/flink/flink-docs-master/dev/table_api.html
> The following links show "Page 'X' Has Moved to" for 1-2 sec  and then 
> redirect to correct page
> DataStream API
> link: 
> http://ci.apache.org/projects/flink/flink-docs-master/apis/streaming/index.html
> redirects-to: 
> https://ci.apache.org/projects/flink/flink-docs-master/dev/datastream_api.html
> programming guide
> link: 
> http://ci.apache.org/projects/flink/flink-docs-master/apis/programming_guide.html
> redirects-to DataStream API: 
> https://ci.apache.org/projects/flink/flink-docs-master/dev/batch/index.html
> probably it should be "Basic API Concepts" 
> https://ci.apache.org/projects/flink/flink-docs-master/dev/api_concepts.html
> or Quick Start - 
> https://ci.apache.org/projects/flink/flink-docs-master/quickstart/setup_quickstart.html



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (FLINK-4571) Configurable little parallelism in Gelly drivers

[ASF GitHub Bot (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLINK-4571?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15468406#comment-15468406
 ] 

ASF GitHub Bot commented on FLINK-4571:
---

GitHub user greghogan opened a pull request:

https://github.com/apache/flink/pull/2475

[FLINK-4571] [gelly] Configurable little parallelism in Gelly drivers




You can merge this pull request into a Git repository by running:

$ git pull https://github.com/greghogan/flink 
4571_configurable_little_parallelism_in_gelly_drivers

Alternatively you can review and apply these changes as the patch at:

https://github.com/apache/flink/pull/2475.patch

To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:

This closes #2475


commit 4e42a617d79e0400fc55dba0d46dc6c9012b0c59
Author: Greg Hogan 
Date:   2016-09-02T15:53:08Z

[FLINK-4571] [gelly] Configurable little parallelism in Gelly drivers




> Configurable little parallelism in Gelly drivers
> 
>
> Key: FLINK-4571
> URL: https://issues.apache.org/jira/browse/FLINK-4571
> Project: Flink
>  Issue Type: Improvement
>  Components: Gelly
>Affects Versions: 1.2.0
>Reporter: Greg Hogan
>Assignee: Greg Hogan
>Priority: Minor
>
> Several Gelly library implementations support a configurable "little 
> parallelism" which is important when scaling to large data sets. These 
> algorithms include operators at the beginning and end which process data on 
> the order of the original DataSet, as well as middle operators that exchange 
> 100s or 1000s more data. The "little parallelism" should be configurable in 
> the appropriate Gelly drivers in the flink-gelly-examples module.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[GitHub] flink pull request #2475: [FLINK-4571] [gelly] Configurable little paralleli...

[greghogan]

GitHub user greghogan opened a pull request:

https://github.com/apache/flink/pull/2475

[FLINK-4571] [gelly] Configurable little parallelism in Gelly drivers




You can merge this pull request into a Git repository by running:

$ git pull https://github.com/greghogan/flink 
4571_configurable_little_parallelism_in_gelly_drivers

Alternatively you can review and apply these changes as the patch at:

https://github.com/apache/flink/pull/2475.patch

To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:

This closes #2475


commit 4e42a617d79e0400fc55dba0d46dc6c9012b0c59
Author: Greg Hogan 
Date:   2016-09-02T15:53:08Z

[FLINK-4571] [gelly] Configurable little parallelism in Gelly drivers




---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[GitHub] flink pull request #2474: [FLINK-4257] [gelly] Handle delegating algorithm c...

[greghogan]

GitHub user greghogan opened a pull request:

https://github.com/apache/flink/pull/2474

[FLINK-4257] [gelly] Handle delegating algorithm change of class

Replaces Delegate with NoOpOperator.

You can merge this pull request into a Git repository by running:

$ git pull https://github.com/greghogan/flink 
4257_handle_delegating_algorithm_change_of_class

Alternatively you can review and apply these changes as the patch at:

https://github.com/apache/flink/pull/2474.patch

To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:

This closes #2474


commit f8d1aa966ba907b2e3a306f5fe498ed8ac886656
Author: Greg Hogan 
Date:   2016-07-25T13:09:27Z

[FLINK-4257 to be squashed] [gelly] Handle delegating algorithm change of 
class

A class created by ProxyFactory can intercept and reinterpret method
calls using its MethodHandler, but is restricted in that

* the type of the proxy class cannot be changed
* method return types must be honored

We have algorithms such as VertexDegree and TriangleListing that change
return type depending on configuration, even between single and dual
input functions. This can be problematic, e.g. in OperatorTranslation
where we test dataSet instanceof SingleInputOperator or dataSet
instanceof TwoInputOperator.

Even simply changing operator can be problematic, e.g.
MapOperator.translateToDataFlow returns MapOperatorBase whereas
ReduceOperator.translateToDataFlow returns SingleInputOperator.

Making changes only within Gelly we can append a "no-op" pass-through
MapFunction to any algorithm output which is not a SingleInputOperator.
And Delegate can also walk the superclass hierarchy such we are always
proxying SingleInputOperator.

There is one additional issue. When we call DataSet.output the
delegate's MethodHandler must reinterpret this call to add itself to the
list of sinks.

commit 3209b05ab5575aba08f9cd6284b720ef9dfc5a02
Author: Greg Hogan 
Date:   2016-07-25T13:41:31Z

[FLINK-4257] [gelly] Handle delegating algorithm change of class

Replaces Delegate with NoOpOperator.




---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[jira] [Commented] (FLINK-4257) Handle delegating algorithm change of class

[ASF GitHub Bot (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLINK-4257?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15468397#comment-15468397
 ] 

ASF GitHub Bot commented on FLINK-4257:
---

GitHub user greghogan opened a pull request:

https://github.com/apache/flink/pull/2474

[FLINK-4257] [gelly] Handle delegating algorithm change of class

Replaces Delegate with NoOpOperator.

You can merge this pull request into a Git repository by running:

$ git pull https://github.com/greghogan/flink 
4257_handle_delegating_algorithm_change_of_class

Alternatively you can review and apply these changes as the patch at:

https://github.com/apache/flink/pull/2474.patch

To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:

This closes #2474


commit f8d1aa966ba907b2e3a306f5fe498ed8ac886656
Author: Greg Hogan 
Date:   2016-07-25T13:09:27Z

[FLINK-4257 to be squashed] [gelly] Handle delegating algorithm change of 
class

A class created by ProxyFactory can intercept and reinterpret method
calls using its MethodHandler, but is restricted in that

* the type of the proxy class cannot be changed
* method return types must be honored

We have algorithms such as VertexDegree and TriangleListing that change
return type depending on configuration, even between single and dual
input functions. This can be problematic, e.g. in OperatorTranslation
where we test dataSet instanceof SingleInputOperator or dataSet
instanceof TwoInputOperator.

Even simply changing operator can be problematic, e.g.
MapOperator.translateToDataFlow returns MapOperatorBase whereas
ReduceOperator.translateToDataFlow returns SingleInputOperator.

Making changes only within Gelly we can append a "no-op" pass-through
MapFunction to any algorithm output which is not a SingleInputOperator.
And Delegate can also walk the superclass hierarchy such we are always
proxying SingleInputOperator.

There is one additional issue. When we call DataSet.output the
delegate's MethodHandler must reinterpret this call to add itself to the
list of sinks.

commit 3209b05ab5575aba08f9cd6284b720ef9dfc5a02
Author: Greg Hogan 
Date:   2016-07-25T13:41:31Z

[FLINK-4257] [gelly] Handle delegating algorithm change of class

Replaces Delegate with NoOpOperator.




> Handle delegating algorithm change of class
> ---
>
> Key: FLINK-4257
> URL: https://issues.apache.org/jira/browse/FLINK-4257
> Project: Flink
>  Issue Type: Bug
>  Components: Gelly
>Affects Versions: 1.1.0
>Reporter: Greg Hogan
>Assignee: Greg Hogan
>
> A class created by {{ProxyFactory}} can intercept and reinterpret method 
> calls using its {{MethodHandler}}, but is restricted in that
> * the type of the proxy class cannot be changed
> * method return types must be honored
> We have algorithms such as {{VertexDegree}} and {{TriangleListing}} that 
> change return type depending on configuration, even between single and dual 
> input functions. This can be problematic, e.g. in {{OperatorTranslation}} 
> where we test {{dataSet instanceof SingleInputOperator}} or {{dataSet 
> instanceof TwoInputOperator}}.
> Even simply changing operator can be problematic, e.g. 
> {{MapOperator.translateToDataFlow}} returns {{MapOperatorBase}} whereas 
> {{ReduceOperator.translateToDataFlow}} returns {{SingleInputOperator}}.
> I see two ways to solve these issues. By adding a simple {{NoOpOperator}} 
> that is skipped over during {{OperatorTranslation}} we could wrap all 
> algorithm output and always be proxying the same class.
> Alternatively, making changes only within Gelly we can append a "no-op" 
> pass-through {{MapFunction}} to any algorithm output which is not a 
> {{SingleInputOperator}}. And {{Delegate}} can also walk the superclass 
> hierarchy such we are always proxying {{SingleInputOperator}}.
> There is one additional issue. When we call {{DataSet.output}} the delegate's 
> {{MethodHandler}} must reinterpret this call to add itself to the list of 
> sinks.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Created] (FLINK-4585) Fix broken links on flink.apache.org

[Alexander Pivovarov (JIRA)]

Alexander Pivovarov created FLINK-4585:
--

 Summary: Fix broken links on flink.apache.org
 Key: FLINK-4585
 URL: https://issues.apache.org/jira/browse/FLINK-4585
 Project: Flink
  Issue Type: Bug
  Components: Project Website
Reporter: Alexander Pivovarov
Priority: Minor


The following links are broken
link: 
http://ci.apache.org/projects/flink/flink-docs-master/apis/batch/index.html
correct link: 
https://ci.apache.org/projects/flink/flink-docs-master/dev/batch/index.html

link: 
http://ci.apache.org/projects/flink/flink-docs-master/apis/batch/libs/table.html
correct link: 
https://ci.apache.org/projects/flink/flink-docs-master/dev/table_api.html


The following links show "Page 'X' Has Moved to" for 1-2 sec  and then redirect 
to correct page
link: 
http://ci.apache.org/projects/flink/flink-docs-master/apis/streaming/index.html
redirect-to: 
https://ci.apache.org/projects/flink/flink-docs-master/dev/datastream_api.html

link: 
http://ci.apache.org/projects/flink/flink-docs-master/apis/programming_guide.html
redirect-to: 
https://ci.apache.org/projects/flink/flink-docs-master/dev/batch/index.html




--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Created] (FLINK-4584) Fix broken links on flink.apache.org

[Alexander Pivovarov (JIRA)]

Alexander Pivovarov created FLINK-4584:
--

 Summary: Fix broken links on flink.apache.org
 Key: FLINK-4584
 URL: https://issues.apache.org/jira/browse/FLINK-4584
 Project: Flink
  Issue Type: Bug
  Components: Project Website
Reporter: Alexander Pivovarov
Priority: Minor


The following links are broken
link: 
http://ci.apache.org/projects/flink/flink-docs-master/apis/batch/index.html
correct link: 
https://ci.apache.org/projects/flink/flink-docs-master/dev/batch/index.html

link: 
http://ci.apache.org/projects/flink/flink-docs-master/apis/batch/libs/table.html
correct link: 
https://ci.apache.org/projects/flink/flink-docs-master/dev/table_api.html


The following links show "Page 'X' Has Moved to" for 1-2 sec  and then redirect 
to correct page
link: 
http://ci.apache.org/projects/flink/flink-docs-master/apis/streaming/index.html
redirect-to: 
https://ci.apache.org/projects/flink/flink-docs-master/dev/datastream_api.html

link: 
http://ci.apache.org/projects/flink/flink-docs-master/apis/programming_guide.html
redirect-to: 
https://ci.apache.org/projects/flink/flink-docs-master/dev/batch/index.html




--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (FLINK-2254) Add Bipartite Graph Support for Gelly

[Greg Hogan (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLINK-2254?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15468368#comment-15468368
 ] 

Greg Hogan commented on FLINK-2254:
---

Hi Ivan, I haven't read the document, and still look for [~vkalavri] to chime 
in, but from reading the earlier comments each of the two vertex DataSets will 
have a different label type. This necessitates a new BipartiteEdge class. I 
think separate Graph and BipartiteGraph classes will work better than a common 
hierarchy.

> Add Bipartite Graph Support for Gelly
> -
>
> Key: FLINK-2254
> URL: https://issues.apache.org/jira/browse/FLINK-2254
> Project: Flink
>  Issue Type: New Feature
>  Components: Gelly
>Affects Versions: 0.10.0
>Reporter: Andra Lungu
>Assignee: Ivan Mushketyk
>  Labels: requires-design-doc
>
> A bipartite graph is a graph for which the set of vertices can be divided 
> into two disjoint sets such that each edge having a source vertex in the 
> first set, will have a target vertex in the second set. We would like to 
> support efficient operations for this type of graphs along with a set of 
> metrics(http://jponnela.com/web_documents/twomode.pdf). 



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (FLINK-4554) Add support for array types

[Timo Walther (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLINK-4554?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15468154#comment-15468154
 ] 

Timo Walther commented on FLINK-4554:
-

Apache Calcite does the parsing for us and they strictly follow the SQL 
standard. I don't know if they support this syntax.

> Add support for array types
> ---
>
> Key: FLINK-4554
> URL: https://issues.apache.org/jira/browse/FLINK-4554
> Project: Flink
>  Issue Type: New Feature
>  Components: Table API & SQL
>Reporter: Timo Walther
>
> Support creating arrays:
> {code}ARRAY[1, 2, 3]{code}
> Access array values:
> {code}myArray[3]{code}
> And operations like:
> {{UNNEST, UNNEST WITH ORDINALITY, CARDINALITY}}



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (FLINK-4565) Support for SQL IN operator

[Simone Robutti (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLINK-4565?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15467945#comment-15467945
 ] 

Simone Robutti commented on FLINK-4565:
---

I haven't assigned the issue to myself because I'm still studying to clear the 
remaining grey areas. Tomorrow I will, if no one else did. 

> Support for SQL IN operator
> ---
>
> Key: FLINK-4565
> URL: https://issues.apache.org/jira/browse/FLINK-4565
> Project: Flink
>  Issue Type: Improvement
>  Components: Table API & SQL
>Reporter: Timo Walther
>
> It seems that Flink SQL supports the uncorrelated sub-query IN operator. But 
> it should also be available in the Table API and tested.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[GitHub] flink pull request #2332: [FLINK-2055] Implement Streaming HBaseSink

[delding]

Github user delding commented on a diff in the pull request:

https://github.com/apache/flink/pull/2332#discussion_r77676668
  
--- Diff: 
flink-streaming-connectors/flink-connector-hbase/src/main/java/org/apache/flink/streaming/connectors/hbase/HBaseMapper.java
 ---
@@ -0,0 +1,46 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.connectors.hbase;
+
+import org.apache.flink.api.common.functions.Function;
+
+import java.io.Serializable;
+
+/**
+ * Maps a input value to a row in HBase table.
+ *
+ * @param  input type
+ */
+public interface HBaseMapper extends Function, Serializable {
+
+   /**
+* Given an input value return the HBase row key. Row key cannot be 
null.
+*
+* @param value
+* @return row key
+*/
+   byte[] rowKey(IN value);
--- End diff --

I did forget to add a connect method. Will add one shortly.


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[jira] [Commented] (FLINK-2055) Implement Streaming HBaseSink

[ASF GitHub Bot (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLINK-2055?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15467934#comment-15467934
 ] 

ASF GitHub Bot commented on FLINK-2055:
---

Github user delding commented on a diff in the pull request:

https://github.com/apache/flink/pull/2332#discussion_r77676668
  
--- Diff: 
flink-streaming-connectors/flink-connector-hbase/src/main/java/org/apache/flink/streaming/connectors/hbase/HBaseMapper.java
 ---
@@ -0,0 +1,46 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.connectors.hbase;
+
+import org.apache.flink.api.common.functions.Function;
+
+import java.io.Serializable;
+
+/**
+ * Maps a input value to a row in HBase table.
+ *
+ * @param  input type
+ */
+public interface HBaseMapper extends Function, Serializable {
+
+   /**
+* Given an input value return the HBase row key. Row key cannot be 
null.
+*
+* @param value
+* @return row key
+*/
+   byte[] rowKey(IN value);
--- End diff --

I did forget to add a connect method. Will add one shortly.


> Implement Streaming HBaseSink
> -
>
> Key: FLINK-2055
> URL: https://issues.apache.org/jira/browse/FLINK-2055
> Project: Flink
>  Issue Type: New Feature
>  Components: Streaming, Streaming Connectors
>Affects Versions: 0.9
>Reporter: Robert Metzger
>Assignee: Erli Ding
>
> As per : 
> http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/Write-Stream-to-HBase-td1300.html



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (FLINK-2055) Implement Streaming HBaseSink

[ASF GitHub Bot (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLINK-2055?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15467928#comment-15467928
 ] 

ASF GitHub Bot commented on FLINK-2055:
---

Github user delding commented on a diff in the pull request:

https://github.com/apache/flink/pull/2332#discussion_r77676350
  
--- Diff: 
flink-streaming-connectors/flink-connector-hbase/src/main/java/org/apache/flink/streaming/connectors/hbase/HBaseMapper.java
 ---
@@ -0,0 +1,46 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.connectors.hbase;
+
+import org.apache.flink.api.common.functions.Function;
+
+import java.io.Serializable;
+
+/**
+ * Maps a input value to a row in HBase table.
+ *
+ * @param  input type
+ */
+public interface HBaseMapper extends Function, Serializable {
+
+   /**
+* Given an input value return the HBase row key. Row key cannot be 
null.
+*
+* @param value
+* @return row key
+*/
+   byte[] rowKey(IN value);
--- End diff --

The assumption is that for each input value there will be only one 
corresponding row to be modified. Users define how to extract a row key and 
actions to be performed on this row key from a single input value. In other 
words, the row key returned from rowKey(IN value) will be passed to 
createMutations(byte[] rowKey, boolean writeToWAL) method called on the 
instance returned from actions(IN value).


> Implement Streaming HBaseSink
> -
>
> Key: FLINK-2055
> URL: https://issues.apache.org/jira/browse/FLINK-2055
> Project: Flink
>  Issue Type: New Feature
>  Components: Streaming, Streaming Connectors
>Affects Versions: 0.9
>Reporter: Robert Metzger
>Assignee: Erli Ding
>
> As per : 
> http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/Write-Stream-to-HBase-td1300.html



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[GitHub] flink pull request #2332: [FLINK-2055] Implement Streaming HBaseSink

[delding]

Github user delding commented on a diff in the pull request:

https://github.com/apache/flink/pull/2332#discussion_r77676350
  
--- Diff: 
flink-streaming-connectors/flink-connector-hbase/src/main/java/org/apache/flink/streaming/connectors/hbase/HBaseMapper.java
 ---
@@ -0,0 +1,46 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.connectors.hbase;
+
+import org.apache.flink.api.common.functions.Function;
+
+import java.io.Serializable;
+
+/**
+ * Maps a input value to a row in HBase table.
+ *
+ * @param  input type
+ */
+public interface HBaseMapper extends Function, Serializable {
+
+   /**
+* Given an input value return the HBase row key. Row key cannot be 
null.
+*
+* @param value
+* @return row key
+*/
+   byte[] rowKey(IN value);
--- End diff --

The assumption is that for each input value there will be only one 
corresponding row to be modified. Users define how to extract a row key and 
actions to be performed on this row key from a single input value. In other 
words, the row key returned from rowKey(IN value) will be passed to 
createMutations(byte[] rowKey, boolean writeToWAL) method called on the 
instance returned from actions(IN value).


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[jira] [Commented] (FLINK-4315) Remove Hadoop Dependency from flink-java

[Konstantin Boudnik (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLINK-4315?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15467921#comment-15467921
 ] 

Konstantin Boudnik commented on FLINK-4315:
---

I have a hunch, once the dependency is severed, it will need to be reflected in 
the integration with Apache Bigtop to avoid needless ties.

> Remove Hadoop Dependency from flink-java
> 
>
> Key: FLINK-4315
> URL: https://issues.apache.org/jira/browse/FLINK-4315
> Project: Flink
>  Issue Type: Sub-task
>  Components: Java API
>Reporter: Stephan Ewen
> Fix For: 2.0.0
>
>
> The API projects should be independent of Hadoop, because Hadoop is not an 
> integral part of the Flink stack, and we should have the option to offer 
> Flink without Hadoop dependencies.
> The current batch APIs have a hard dependency on Hadoop, mainly because the 
> API has utility methods like `readHadoopFile(...)`.
> I suggest to remove those methods and instead add helpers in the 
> `flink-hadoop-compatibility` project.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (FLINK-4554) Add support for array types

[Konstantin Boudnik (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLINK-4554?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15467918#comment-15467918
 ] 

Konstantin Boudnik commented on FLINK-4554:
---

Do you think adding a support for ranges a-la Groovy style ie {{ARRAY[1..3]}} 
should also be supported?

> Add support for array types
> ---
>
> Key: FLINK-4554
> URL: https://issues.apache.org/jira/browse/FLINK-4554
> Project: Flink
>  Issue Type: New Feature
>  Components: Table API & SQL
>Reporter: Timo Walther
>
> Support creating arrays:
> {code}ARRAY[1, 2, 3]{code}
> Access array values:
> {code}myArray[3]{code}
> And operations like:
> {{UNNEST, UNNEST WITH ORDINALITY, CARDINALITY}}



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (FLINK-4565) Support for SQL IN operator

[Konstantin Boudnik (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLINK-4565?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15467913#comment-15467913
 ] 

Konstantin Boudnik commented on FLINK-4565:
---

If someone starts working on an issue, shouldn't it be assigned to that person, 
so no two people work on the same task simultaneously? What is the accepted 
drill here, guys? Appreciate the info!

> Support for SQL IN operator
> ---
>
> Key: FLINK-4565
> URL: https://issues.apache.org/jira/browse/FLINK-4565
> Project: Flink
>  Issue Type: Improvement
>  Components: Table API & SQL
>Reporter: Timo Walther
>
> It seems that Flink SQL supports the uncorrelated sub-query IN operator. But 
> it should also be available in the Table API and tested.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Updated] (FLINK-4257) Handle delegating algorithm change of class

[Greg Hogan (JIRA)]

 [ 
https://issues.apache.org/jira/browse/FLINK-4257?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Greg Hogan updated FLINK-4257:
--
Description: 
A class created by {{ProxyFactory}} can intercept and reinterpret method calls 
using its {{MethodHandler}}, but is restricted in that
* the type of the proxy class cannot be changed
* method return types must be honored

We have algorithms such as {{VertexDegree}} and {{TriangleListing}} that change 
return type depending on configuration, even between single and dual input 
functions. This can be problematic, e.g. in {{OperatorTranslation}} where we 
test {{dataSet instanceof SingleInputOperator}} or {{dataSet instanceof 
TwoInputOperator}}.

Even simply changing operator can be problematic, e.g. 
{{MapOperator.translateToDataFlow}} returns {{MapOperatorBase}} whereas 
{{ReduceOperator.translateToDataFlow}} returns {{SingleInputOperator}}.

I see two ways to solve these issues. By adding a simple {{NoOpOperator}} that 
is skipped over during {{OperatorTranslation}} we could wrap all algorithm 
output and always be proxying the same class.

Alternatively, making changes only within Gelly we can append a "no-op" 
pass-through {{MapFunction}} to any algorithm output which is not a 
{{SingleInputOperator}}. And {{Delegate}} can also walk the superclass 
hierarchy such we are always proxying {{SingleInputOperator}}.

There is one additional issue. When we call {{DataSet.output}} the delegate's 
{{MethodHandler}} must reinterpret this call to add itself to the list of sinks.

  was:
A class created by {{ProxyFactory}} can intercept and reinterpret method calls 
using its {{MethodHandler}}, but is restricted in that
* the type of the proxy class cannot be changed
* method return types must be honored

We have algorithms such as {{VertexDegree}} and {{TriangleListing}} that change 
return type depending on configuration, even between single and dual input 
functions. This can be problematic, e.g. in {{OperatorTranslation}} where we 
test {{dataSet instanceof SingleInputOperator}} or {{dataSet instanceof 
TwoInputOperator}}.

Even simply changing operator can be problematic, e.g. 
{{MapOperator.translateToDataFlow}} returns {{MapOperatorBase}} whereas 
{{ReduceOperator.translateToDataFlow}} returns {{SingleInputOperator}}.

I see two ways to solve these issues. By adding a simple {{NoOpOperator}} that 
is skipped over during {{OperatorTranslation}} we could wrap all algorithm 
output and always be proxying the same class.

Alternatively, making changes only within Gelly we can append a "no-op" 
pass-through {{MapFunction}} to any algorithm output which is not a 
{{SingleInputOperator}}. And {{Delegate}} can also walk the superclass 
hierarchy such we are always proxying {{SingleInputOperator}}.

There is one additional issue. When we call {{DataSet.output}} the delegate's 
{{MethodHandler}} must reinterpret this call to add itself to the list of sinks.

As part of this issue I will also add manual tests to Gelly for the library 
algorithms which do not have integration tests.


> Handle delegating algorithm change of class
> ---
>
> Key: FLINK-4257
> URL: https://issues.apache.org/jira/browse/FLINK-4257
> Project: Flink
>  Issue Type: Bug
>  Components: Gelly
>Affects Versions: 1.1.0
>Reporter: Greg Hogan
>Assignee: Greg Hogan
>
> A class created by {{ProxyFactory}} can intercept and reinterpret method 
> calls using its {{MethodHandler}}, but is restricted in that
> * the type of the proxy class cannot be changed
> * method return types must be honored
> We have algorithms such as {{VertexDegree}} and {{TriangleListing}} that 
> change return type depending on configuration, even between single and dual 
> input functions. This can be problematic, e.g. in {{OperatorTranslation}} 
> where we test {{dataSet instanceof SingleInputOperator}} or {{dataSet 
> instanceof TwoInputOperator}}.
> Even simply changing operator can be problematic, e.g. 
> {{MapOperator.translateToDataFlow}} returns {{MapOperatorBase}} whereas 
> {{ReduceOperator.translateToDataFlow}} returns {{SingleInputOperator}}.
> I see two ways to solve these issues. By adding a simple {{NoOpOperator}} 
> that is skipped over during {{OperatorTranslation}} we could wrap all 
> algorithm output and always be proxying the same class.
> Alternatively, making changes only within Gelly we can append a "no-op" 
> pass-through {{MapFunction}} to any algorithm output which is not a 
> {{SingleInputOperator}}. And {{Delegate}} can also walk the superclass 
> hierarchy such we are always proxying {{SingleInputOperator}}.
> There is one additional issue. When we call {{DataSet.output}} the delegate's 
> {{MethodHandler}} must reinterpret this call to add itself to the list of 
> sinks.



--
This message was sent by Atlassian JIRA

[jira] [Commented] (FLINK-3427) Add watermark monitoring to JobManager web frontend

[Ivan Mushketyk (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLINK-3427?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15467767#comment-15467767
 ] 

Ivan Mushketyk commented on FLINK-3427:
---

[~rmetzger] If this will help in the discussing was planning to add two new 
metric gauges for last watermal and number of events.

> Add watermark monitoring to JobManager web frontend
> ---
>
> Key: FLINK-3427
> URL: https://issues.apache.org/jira/browse/FLINK-3427
> Project: Flink
>  Issue Type: Improvement
>  Components: Streaming, Webfrontend
>Reporter: Robert Metzger
>Assignee: Ivan Mushketyk
>
> Currently, its quite hard to figure out issues with the watermarks.
> I think we can improve the situation by reporting the following information 
> through the metrics system:
> - Report the current low watermark for each operator (this way, you can see 
> if one operator is preventing the watermarks to rise)
> - Report the number of events arrived after the low watermark (users can see 
> how accurate the watermarks are)



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (FLINK-3427) Add watermark monitoring to JobManager web frontend

[Ivan Mushketyk (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLINK-3427?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15467660#comment-15467660
 ] 

Ivan Mushketyk commented on FLINK-3427:
---

Hey [~rmetzger], my idea was to record a timestamp of the latest watermark on 
every vertex and display timestamps in the "Subtask" panel with other metrics 
(here is a panel I am talking about: 
https://cloud.githubusercontent.com/assets/592286/18147690/c287947e-6fcd-11e6-8968-ff47283635b7.png).

On the other hand, we could enhance the graph representation and display 
watermarks and number of events for every vertex there. Maybe something similar 
to Google Dataflow UI: https://youtu.be/TnLiEWglqHk?t=22m4s

> Add watermark monitoring to JobManager web frontend
> ---
>
> Key: FLINK-3427
> URL: https://issues.apache.org/jira/browse/FLINK-3427
> Project: Flink
>  Issue Type: Improvement
>  Components: Streaming, Webfrontend
>Reporter: Robert Metzger
>Assignee: Ivan Mushketyk
>
> Currently, its quite hard to figure out issues with the watermarks.
> I think we can improve the situation by reporting the following information 
> through the metrics system:
> - Report the current low watermark for each operator (this way, you can see 
> if one operator is preventing the watermarks to rise)
> - Report the number of events arrived after the low watermark (users can see 
> how accurate the watermarks are)



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (FLINK-4538) Implement slot allocation protocol with JobMaster

[ASF GitHub Bot (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLINK-4538?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15467602#comment-15467602
 ] 

ASF GitHub Bot commented on FLINK-4538:
---

Github user tillrohrmann commented on a diff in the pull request:

https://github.com/apache/flink/pull/2463#discussion_r77650617
  
--- Diff: 
flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/slotmanager/SlotProtocolTest.java
 ---
@@ -0,0 +1,228 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.runtime.resourcemanager.slotmanager;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.runtime.clusterframework.types.AllocationID;
+import org.apache.flink.runtime.clusterframework.types.ResourceID;
+import org.apache.flink.runtime.clusterframework.types.ResourceProfile;
+import org.apache.flink.runtime.clusterframework.types.SlotID;
+import org.apache.flink.runtime.highavailability.LeaderIdRegistry;
+import org.apache.flink.runtime.highavailability.NonHaServices;
+import org.apache.flink.runtime.jobmaster.JobMasterGateway;
+import org.apache.flink.runtime.resourcemanager.JobMasterRegistration;
+import org.apache.flink.runtime.resourcemanager.RegistrationResponse;
+import org.apache.flink.runtime.resourcemanager.ResourceManager;
+import org.apache.flink.runtime.resourcemanager.SlotRequest;
+import org.apache.flink.runtime.resourcemanager.SlotRequestRegistered;
+import org.apache.flink.runtime.rpc.TestingRpcService;
+import org.apache.flink.runtime.taskexecutor.SlotReport;
+import org.apache.flink.runtime.taskexecutor.SlotStatus;
+import org.apache.flink.runtime.taskexecutor.TaskExecutor;
+import org.apache.flink.runtime.taskexecutor.TaskExecutorGateway;
+import org.junit.AfterClass;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.mockito.Mockito;
+import scala.concurrent.Await;
+import scala.concurrent.Future;
+import scala.concurrent.duration.Duration;
+
+import java.util.Collections;
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
+
+import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.anyLong;
+import static org.mockito.Matchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.timeout;
+import static org.mockito.Mockito.verify;
+
+public class SlotProtocolTest {
+
+   private static TestingRpcService testRpcService;
+
+   @BeforeClass
+   public static void beforeClass() {
+   testRpcService = new TestingRpcService();
+
+   }
+
+   @AfterClass
+   public static void afterClass() {
+   testRpcService.stopService();
+   testRpcService = null;
+   }
+
+   @Before
+   public void beforeTest(){
+   testRpcService.clearGateways();
+   }
+
+   /**
+* Tests whether
+* 1) SlotRequest is routed to the SlotManager
+* 2) SlotRequest leads to a container allocation
+* 3) SlotRequest is confirmed
+* 4) Slot becomes available and TaskExecutor gets a SlotRequest
+*/
+   @Test
+   public void testSlotsUnavailableRequest() throws Exception {
+   final String rmAddress = "/rm1";
+   final String jmAddress = "/jm1";
+   final JobID jobID = new JobID();
+
+   testRpcService.registerGateway(jmAddress, 
mock(JobMasterGateway.class));
+
+
+   TestingSlotManager slotManager = Mockito.spy(new 
TestingSlotManager());
+   ResourceManager resourceManager =
+   new ResourceManager(testRpcService, new 
NonHaServices(rmAddress), slotManager);
+   resourceManager.start();
+
+   Future registrationFuture =
+   resourceManager.registerJobMaster(new 
JobMasterRegistration(jmAddress, 

[jira] [Created] (FLINK-4583) NullPointerException in CliFrontend

[Greg Hogan (JIRA)]

Greg Hogan created FLINK-4583:
-

 Summary: NullPointerException in CliFrontend
 Key: FLINK-4583
 URL: https://issues.apache.org/jira/browse/FLINK-4583
 Project: Flink
  Issue Type: Bug
  Components: Client
Affects Versions: 1.2.0
Reporter: Greg Hogan
Assignee: Greg Hogan
Priority: Minor


If no Flink program is executed the following exception message is printed. 
This can happen when a driver prints usage due to insufficient or improper 
configuration.

{noformat}
 The program finished with the following exception:

java.lang.NullPointerException
at 
org.apache.flink.client.CliFrontend.executeProgram(CliFrontend.java:781)
at org.apache.flink.client.CliFrontend.run(CliFrontend.java:250)
at 
org.apache.flink.client.CliFrontend.parseParameters(CliFrontend.java:1002)
at org.apache.flink.client.CliFrontend.main(CliFrontend.java:1045)
{noformat}



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (FLINK-4538) Implement slot allocation protocol with JobMaster

[ASF GitHub Bot (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLINK-4538?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15467612#comment-15467612
 ] 

ASF GitHub Bot commented on FLINK-4538:
---

Github user tillrohrmann commented on a diff in the pull request:

https://github.com/apache/flink/pull/2463#discussion_r77651399
  
--- Diff: 
flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManager.java
 ---
@@ -52,15 +58,28 @@
  * 
  */
 public class ResourceManager extends RpcEndpoint {
-   private final Map jobMasterGateways;
+
+   private final Logger LOG = LoggerFactory.getLogger(getClass());
--- End diff --

But then this field should probably marked as `protected` instead of 
`private`.


> Implement slot allocation protocol with JobMaster
> -
>
> Key: FLINK-4538
> URL: https://issues.apache.org/jira/browse/FLINK-4538
> Project: Flink
>  Issue Type: Sub-task
>  Components: Cluster Management
>Reporter: Maximilian Michels
>Assignee: Maximilian Michels
>




--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[GitHub] flink pull request #2463: [FLINK-4538][FLINK-4348] ResourceManager slot allo...

[tillrohrmann]

Github user tillrohrmann commented on a diff in the pull request:

https://github.com/apache/flink/pull/2463#discussion_r77651399
  
--- Diff: 
flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManager.java
 ---
@@ -52,15 +58,28 @@
  * 
  */
 public class ResourceManager extends RpcEndpoint {
-   private final Map jobMasterGateways;
+
+   private final Logger LOG = LoggerFactory.getLogger(getClass());
--- End diff --

But then this field should probably marked as `protected` instead of 
`private`.


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[GitHub] flink pull request #2463: [FLINK-4538][FLINK-4348] ResourceManager slot allo...

[tillrohrmann]

Github user tillrohrmann commented on a diff in the pull request:

https://github.com/apache/flink/pull/2463#discussion_r77651201
  
--- Diff: 
flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/slotmanager/SlotProtocolTest.java
 ---
@@ -0,0 +1,228 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.runtime.resourcemanager.slotmanager;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.runtime.clusterframework.types.AllocationID;
+import org.apache.flink.runtime.clusterframework.types.ResourceID;
+import org.apache.flink.runtime.clusterframework.types.ResourceProfile;
+import org.apache.flink.runtime.clusterframework.types.SlotID;
+import org.apache.flink.runtime.highavailability.LeaderIdRegistry;
+import org.apache.flink.runtime.highavailability.NonHaServices;
+import org.apache.flink.runtime.jobmaster.JobMasterGateway;
+import org.apache.flink.runtime.resourcemanager.JobMasterRegistration;
+import org.apache.flink.runtime.resourcemanager.RegistrationResponse;
+import org.apache.flink.runtime.resourcemanager.ResourceManager;
+import org.apache.flink.runtime.resourcemanager.SlotRequest;
+import org.apache.flink.runtime.resourcemanager.SlotRequestRegistered;
+import org.apache.flink.runtime.rpc.TestingRpcService;
+import org.apache.flink.runtime.taskexecutor.SlotReport;
+import org.apache.flink.runtime.taskexecutor.SlotStatus;
+import org.apache.flink.runtime.taskexecutor.TaskExecutor;
+import org.apache.flink.runtime.taskexecutor.TaskExecutorGateway;
+import org.junit.AfterClass;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.mockito.Mockito;
+import scala.concurrent.Await;
+import scala.concurrent.Future;
+import scala.concurrent.duration.Duration;
+
+import java.util.Collections;
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
+
+import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.anyLong;
+import static org.mockito.Matchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.timeout;
+import static org.mockito.Mockito.verify;
+
+public class SlotProtocolTest {
+
+   private static TestingRpcService testRpcService;
+
+   @BeforeClass
+   public static void beforeClass() {
+   testRpcService = new TestingRpcService();
+
+   }
+
+   @AfterClass
+   public static void afterClass() {
+   testRpcService.stopService();
+   testRpcService = null;
+   }
+
+   @Before
+   public void beforeTest(){
+   testRpcService.clearGateways();
+   }
+
+   /**
+* Tests whether
+* 1) SlotRequest is routed to the SlotManager
+* 2) SlotRequest leads to a container allocation
+* 3) SlotRequest is confirmed
+* 4) Slot becomes available and TaskExecutor gets a SlotRequest
+*/
+   @Test
+   public void testSlotsUnavailableRequest() throws Exception {
+   final String rmAddress = "/rm1";
+   final String jmAddress = "/jm1";
+   final JobID jobID = new JobID();
+
+   testRpcService.registerGateway(jmAddress, 
mock(JobMasterGateway.class));
+
+
+   TestingSlotManager slotManager = Mockito.spy(new 
TestingSlotManager());
+   ResourceManager resourceManager =
+   new ResourceManager(testRpcService, new 
NonHaServices(rmAddress), slotManager);
+   resourceManager.start();
+
+   Future registrationFuture =
+   resourceManager.registerJobMaster(new 
JobMasterRegistration(jmAddress, jobID));
+   try {
+   Await.ready(registrationFuture, Duration.create(5, 
TimeUnit.SECONDS));
+   } catch (Exception e) {
+   Assert.fail("JobManager registration Future didn't 

[jira] [Commented] (FLINK-4538) Implement slot allocation protocol with JobMaster

[ASF GitHub Bot (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLINK-4538?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15467609#comment-15467609
 ] 

ASF GitHub Bot commented on FLINK-4538:
---

Github user tillrohrmann commented on a diff in the pull request:

https://github.com/apache/flink/pull/2463#discussion_r77651201
  
--- Diff: 
flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/slotmanager/SlotProtocolTest.java
 ---
@@ -0,0 +1,228 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.runtime.resourcemanager.slotmanager;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.runtime.clusterframework.types.AllocationID;
+import org.apache.flink.runtime.clusterframework.types.ResourceID;
+import org.apache.flink.runtime.clusterframework.types.ResourceProfile;
+import org.apache.flink.runtime.clusterframework.types.SlotID;
+import org.apache.flink.runtime.highavailability.LeaderIdRegistry;
+import org.apache.flink.runtime.highavailability.NonHaServices;
+import org.apache.flink.runtime.jobmaster.JobMasterGateway;
+import org.apache.flink.runtime.resourcemanager.JobMasterRegistration;
+import org.apache.flink.runtime.resourcemanager.RegistrationResponse;
+import org.apache.flink.runtime.resourcemanager.ResourceManager;
+import org.apache.flink.runtime.resourcemanager.SlotRequest;
+import org.apache.flink.runtime.resourcemanager.SlotRequestRegistered;
+import org.apache.flink.runtime.rpc.TestingRpcService;
+import org.apache.flink.runtime.taskexecutor.SlotReport;
+import org.apache.flink.runtime.taskexecutor.SlotStatus;
+import org.apache.flink.runtime.taskexecutor.TaskExecutor;
+import org.apache.flink.runtime.taskexecutor.TaskExecutorGateway;
+import org.junit.AfterClass;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.mockito.Mockito;
+import scala.concurrent.Await;
+import scala.concurrent.Future;
+import scala.concurrent.duration.Duration;
+
+import java.util.Collections;
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
+
+import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.anyLong;
+import static org.mockito.Matchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.timeout;
+import static org.mockito.Mockito.verify;
+
+public class SlotProtocolTest {
+
+   private static TestingRpcService testRpcService;
+
+   @BeforeClass
+   public static void beforeClass() {
+   testRpcService = new TestingRpcService();
+
+   }
+
+   @AfterClass
+   public static void afterClass() {
+   testRpcService.stopService();
+   testRpcService = null;
+   }
+
+   @Before
+   public void beforeTest(){
+   testRpcService.clearGateways();
+   }
+
+   /**
+* Tests whether
+* 1) SlotRequest is routed to the SlotManager
+* 2) SlotRequest leads to a container allocation
+* 3) SlotRequest is confirmed
+* 4) Slot becomes available and TaskExecutor gets a SlotRequest
+*/
+   @Test
+   public void testSlotsUnavailableRequest() throws Exception {
+   final String rmAddress = "/rm1";
+   final String jmAddress = "/jm1";
+   final JobID jobID = new JobID();
+
+   testRpcService.registerGateway(jmAddress, 
mock(JobMasterGateway.class));
+
+
+   TestingSlotManager slotManager = Mockito.spy(new 
TestingSlotManager());
+   ResourceManager resourceManager =
+   new ResourceManager(testRpcService, new 
NonHaServices(rmAddress), slotManager);
+   resourceManager.start();
+
+   Future registrationFuture =
+   resourceManager.registerJobMaster(new 
JobMasterRegistration(jmAddress, 

[GitHub] flink pull request #2463: [FLINK-4538][FLINK-4348] ResourceManager slot allo...

[tillrohrmann]

Github user tillrohrmann commented on a diff in the pull request:

https://github.com/apache/flink/pull/2463#discussion_r77650487
  
--- Diff: 
flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/slotmanager/SlotProtocolTest.java
 ---
@@ -0,0 +1,228 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.runtime.resourcemanager.slotmanager;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.runtime.clusterframework.types.AllocationID;
+import org.apache.flink.runtime.clusterframework.types.ResourceID;
+import org.apache.flink.runtime.clusterframework.types.ResourceProfile;
+import org.apache.flink.runtime.clusterframework.types.SlotID;
+import org.apache.flink.runtime.highavailability.LeaderIdRegistry;
+import org.apache.flink.runtime.highavailability.NonHaServices;
+import org.apache.flink.runtime.jobmaster.JobMasterGateway;
+import org.apache.flink.runtime.resourcemanager.JobMasterRegistration;
+import org.apache.flink.runtime.resourcemanager.RegistrationResponse;
+import org.apache.flink.runtime.resourcemanager.ResourceManager;
+import org.apache.flink.runtime.resourcemanager.SlotRequest;
+import org.apache.flink.runtime.resourcemanager.SlotRequestRegistered;
+import org.apache.flink.runtime.rpc.TestingRpcService;
+import org.apache.flink.runtime.taskexecutor.SlotReport;
+import org.apache.flink.runtime.taskexecutor.SlotStatus;
+import org.apache.flink.runtime.taskexecutor.TaskExecutor;
+import org.apache.flink.runtime.taskexecutor.TaskExecutorGateway;
+import org.junit.AfterClass;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.mockito.Mockito;
+import scala.concurrent.Await;
+import scala.concurrent.Future;
+import scala.concurrent.duration.Duration;
+
+import java.util.Collections;
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
+
+import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.anyLong;
+import static org.mockito.Matchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.timeout;
+import static org.mockito.Mockito.verify;
+
+public class SlotProtocolTest {
--- End diff --

Should extend `TestLogger`


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[GitHub] flink pull request #2463: [FLINK-4538][FLINK-4348] ResourceManager slot allo...

[tillrohrmann]

Github user tillrohrmann commented on a diff in the pull request:

https://github.com/apache/flink/pull/2463#discussion_r77650617
  
--- Diff: 
flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/slotmanager/SlotProtocolTest.java
 ---
@@ -0,0 +1,228 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.runtime.resourcemanager.slotmanager;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.runtime.clusterframework.types.AllocationID;
+import org.apache.flink.runtime.clusterframework.types.ResourceID;
+import org.apache.flink.runtime.clusterframework.types.ResourceProfile;
+import org.apache.flink.runtime.clusterframework.types.SlotID;
+import org.apache.flink.runtime.highavailability.LeaderIdRegistry;
+import org.apache.flink.runtime.highavailability.NonHaServices;
+import org.apache.flink.runtime.jobmaster.JobMasterGateway;
+import org.apache.flink.runtime.resourcemanager.JobMasterRegistration;
+import org.apache.flink.runtime.resourcemanager.RegistrationResponse;
+import org.apache.flink.runtime.resourcemanager.ResourceManager;
+import org.apache.flink.runtime.resourcemanager.SlotRequest;
+import org.apache.flink.runtime.resourcemanager.SlotRequestRegistered;
+import org.apache.flink.runtime.rpc.TestingRpcService;
+import org.apache.flink.runtime.taskexecutor.SlotReport;
+import org.apache.flink.runtime.taskexecutor.SlotStatus;
+import org.apache.flink.runtime.taskexecutor.TaskExecutor;
+import org.apache.flink.runtime.taskexecutor.TaskExecutorGateway;
+import org.junit.AfterClass;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.mockito.Mockito;
+import scala.concurrent.Await;
+import scala.concurrent.Future;
+import scala.concurrent.duration.Duration;
+
+import java.util.Collections;
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
+
+import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.anyLong;
+import static org.mockito.Matchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.timeout;
+import static org.mockito.Mockito.verify;
+
+public class SlotProtocolTest {
+
+   private static TestingRpcService testRpcService;
+
+   @BeforeClass
+   public static void beforeClass() {
+   testRpcService = new TestingRpcService();
+
+   }
+
+   @AfterClass
+   public static void afterClass() {
+   testRpcService.stopService();
+   testRpcService = null;
+   }
+
+   @Before
+   public void beforeTest(){
+   testRpcService.clearGateways();
+   }
+
+   /**
+* Tests whether
+* 1) SlotRequest is routed to the SlotManager
+* 2) SlotRequest leads to a container allocation
+* 3) SlotRequest is confirmed
+* 4) Slot becomes available and TaskExecutor gets a SlotRequest
+*/
+   @Test
+   public void testSlotsUnavailableRequest() throws Exception {
+   final String rmAddress = "/rm1";
+   final String jmAddress = "/jm1";
+   final JobID jobID = new JobID();
+
+   testRpcService.registerGateway(jmAddress, 
mock(JobMasterGateway.class));
+
+
+   TestingSlotManager slotManager = Mockito.spy(new 
TestingSlotManager());
+   ResourceManager resourceManager =
+   new ResourceManager(testRpcService, new 
NonHaServices(rmAddress), slotManager);
+   resourceManager.start();
+
+   Future registrationFuture =
+   resourceManager.registerJobMaster(new 
JobMasterRegistration(jmAddress, jobID));
+   try {
+   Await.ready(registrationFuture, Duration.create(5, 
TimeUnit.SECONDS));
+   } catch (Exception e) {
+   Assert.fail("JobManager registration Future didn't 

[jira] [Commented] (FLINK-4538) Implement slot allocation protocol with JobMaster

[ASF GitHub Bot (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLINK-4538?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15467599#comment-15467599
 ] 

ASF GitHub Bot commented on FLINK-4538:
---

Github user tillrohrmann commented on a diff in the pull request:

https://github.com/apache/flink/pull/2463#discussion_r77650487
  
--- Diff: 
flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/slotmanager/SlotProtocolTest.java
 ---
@@ -0,0 +1,228 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.runtime.resourcemanager.slotmanager;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.runtime.clusterframework.types.AllocationID;
+import org.apache.flink.runtime.clusterframework.types.ResourceID;
+import org.apache.flink.runtime.clusterframework.types.ResourceProfile;
+import org.apache.flink.runtime.clusterframework.types.SlotID;
+import org.apache.flink.runtime.highavailability.LeaderIdRegistry;
+import org.apache.flink.runtime.highavailability.NonHaServices;
+import org.apache.flink.runtime.jobmaster.JobMasterGateway;
+import org.apache.flink.runtime.resourcemanager.JobMasterRegistration;
+import org.apache.flink.runtime.resourcemanager.RegistrationResponse;
+import org.apache.flink.runtime.resourcemanager.ResourceManager;
+import org.apache.flink.runtime.resourcemanager.SlotRequest;
+import org.apache.flink.runtime.resourcemanager.SlotRequestRegistered;
+import org.apache.flink.runtime.rpc.TestingRpcService;
+import org.apache.flink.runtime.taskexecutor.SlotReport;
+import org.apache.flink.runtime.taskexecutor.SlotStatus;
+import org.apache.flink.runtime.taskexecutor.TaskExecutor;
+import org.apache.flink.runtime.taskexecutor.TaskExecutorGateway;
+import org.junit.AfterClass;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.mockito.Mockito;
+import scala.concurrent.Await;
+import scala.concurrent.Future;
+import scala.concurrent.duration.Duration;
+
+import java.util.Collections;
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
+
+import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.anyLong;
+import static org.mockito.Matchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.timeout;
+import static org.mockito.Mockito.verify;
+
+public class SlotProtocolTest {
--- End diff --

Should extend `TestLogger`


> Implement slot allocation protocol with JobMaster
> -
>
> Key: FLINK-4538
> URL: https://issues.apache.org/jira/browse/FLINK-4538
> Project: Flink
>  Issue Type: Sub-task
>  Components: Cluster Management
>Reporter: Maximilian Michels
>Assignee: Maximilian Michels
>




--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (FLINK-4538) Implement slot allocation protocol with JobMaster

[ASF GitHub Bot (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLINK-4538?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15467596#comment-15467596
 ] 

ASF GitHub Bot commented on FLINK-4538:
---

Github user tillrohrmann commented on a diff in the pull request:

https://github.com/apache/flink/pull/2463#discussion_r77650313
  
--- Diff: 
flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/LeaderIdRegistry.java
 ---
@@ -15,11 +15,27 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
+package org.apache.flink.runtime.highavailability;
 
-package org.apache.flink.runtime.resourcemanager;
+import java.util.UUID;
 
-import java.io.Serializable;
+/**
+ * Registry class to keep track of the current leader ID.
+ */
+public class LeaderIdRegistry {
--- End diff --

Alright, but then this class should be made thread safe and the docs should 
state the purpose.


> Implement slot allocation protocol with JobMaster
> -
>
> Key: FLINK-4538
> URL: https://issues.apache.org/jira/browse/FLINK-4538
> Project: Flink
>  Issue Type: Sub-task
>  Components: Cluster Management
>Reporter: Maximilian Michels
>Assignee: Maximilian Michels
>




--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[GitHub] flink pull request #2463: [FLINK-4538][FLINK-4348] ResourceManager slot allo...

[tillrohrmann]

Github user tillrohrmann commented on a diff in the pull request:

https://github.com/apache/flink/pull/2463#discussion_r77650313
  
--- Diff: 
flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/LeaderIdRegistry.java
 ---
@@ -15,11 +15,27 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
+package org.apache.flink.runtime.highavailability;
 
-package org.apache.flink.runtime.resourcemanager;
+import java.util.UUID;
 
-import java.io.Serializable;
+/**
+ * Registry class to keep track of the current leader ID.
+ */
+public class LeaderIdRegistry {
--- End diff --

Alright, but then this class should be made thread safe and the docs should 
state the purpose.


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[jira] [Closed] (FLINK-4265) Add a NoOpOperator

[Greg Hogan (JIRA)]

 [ 
https://issues.apache.org/jira/browse/FLINK-4265?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Greg Hogan closed FLINK-4265.
-
Resolution: Implemented

Implemented in 66d4b8724b9e9b09225d2bbd3132dc2efdcf843a

> Add a NoOpOperator
> --
>
> Key: FLINK-4265
> URL: https://issues.apache.org/jira/browse/FLINK-4265
> Project: Flink
>  Issue Type: New Feature
>  Components: DataSet API
>Affects Versions: 1.1.0
>Reporter: Greg Hogan
>Assignee: Greg Hogan
>Priority: Minor
>
> One recent feature of Gelly is algorithms which detect duplicated or similar 
> computation which can be shared. My initial implementation could only reuse a 
> {{DataSet}} result. Before committing to Flink this was updated to use a 
> javassist {{ProxyFactory}} allowing configuration to be merged and results to 
> be replaced. There were some issues, as identified in FLINK-4257. With a 
> {{NoOpOperator}} we can remove the use of {{ProxyFactory}} and resolve the 
> identified issues.
> This ticket adds a {{NoOpOperator}} which is unwound in 
> {{OperatorTranslation.translate}}. The {{NoOpOperator}} contains a 
> {{DataSet}} which is accessed by a getter and setter.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (FLINK-4265) Add a NoOpOperator

[ASF GitHub Bot (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLINK-4265?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15467510#comment-15467510
 ] 

ASF GitHub Bot commented on FLINK-4265:
---

Github user asfgit closed the pull request at:

https://github.com/apache/flink/pull/2294


> Add a NoOpOperator
> --
>
> Key: FLINK-4265
> URL: https://issues.apache.org/jira/browse/FLINK-4265
> Project: Flink
>  Issue Type: New Feature
>  Components: DataSet API
>Affects Versions: 1.1.0
>Reporter: Greg Hogan
>Assignee: Greg Hogan
>Priority: Minor
>
> One recent feature of Gelly is algorithms which detect duplicated or similar 
> computation which can be shared. My initial implementation could only reuse a 
> {{DataSet}} result. Before committing to Flink this was updated to use a 
> javassist {{ProxyFactory}} allowing configuration to be merged and results to 
> be replaced. There were some issues, as identified in FLINK-4257. With a 
> {{NoOpOperator}} we can remove the use of {{ProxyFactory}} and resolve the 
> identified issues.
> This ticket adds a {{NoOpOperator}} which is unwound in 
> {{OperatorTranslation.translate}}. The {{NoOpOperator}} contains a 
> {{DataSet}} which is accessed by a getter and setter.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[GitHub] flink pull request #2294: [FLINK-4265] [dataset api] Add a NoOpOperator

[asfgit]

Github user asfgit closed the pull request at:

https://github.com/apache/flink/pull/2294


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[jira] [Updated] (FLINK-4582) Allow FlinkKinesisConsumer to adapt for AWS DynamoDB Streams

[Tzu-Li (Gordon) Tai (JIRA)]

 [ 
https://issues.apache.org/jira/browse/FLINK-4582?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Tzu-Li (Gordon) Tai updated FLINK-4582:
---
Description: 
AWS DynamoDB is a NoSQL database service that has a CDC-like (change data 
capture) feature called DynamoDB Streams 
(http://docs.aws.amazon.com/amazondynamodb/latest/developerguide/Streams.html), 
which is a stream feed of item-level table activities.

The DynamoDB Streams shard abstraction follows that of Kinesis Streams with 
only a slight difference in resharding behaviours, so it is possible to build 
on the internals of our Flink Kinesis Consumer for an exactly-once DynamoDB 
Streams source.

I propose an API something like this:
{code}
DataStream dynamoItemsCdc = 
  FlinkKinesisConsumer.asDynamoDBStream(tableNames, schema, config)
{code}

The feature adds more connectivity to popular AWS services for Flink, and 
combining what Flink has for exactly-once semantics, out-of-core state 
backends, and queryable state with CDC can have very strong use cases. For this 
feature there should only be an extra dependency to the AWS Java SDK for 
DynamoDB, which has Apache License 2.0.

  was:
AWS DynamoDB is a NoSQL database service that has a CDC-like (change data 
capture) feature called DynamoDB Streams 
(http://docs.aws.amazon.com/amazondynamodb/latest/developerguide/Streams.html), 
which is a stream feed of item-level table activities.

The DynamoDB Streams shard abstraction follows that of Kinesis Streams with 
only a slight difference in resharding behaviours, so it is possible to build 
on the internals of our Flink Kinesis Consumer for an exactly-once DynamoDB 
Streams source.

I propose an API something like this:
{code}
DataStream dynamoItemsCdc = 
  FlinkKinesisConsumer.asDynamoDBStream(tableNames, schema, config)
{code}

The feature adds more connectivity to popular AWS services for Flink, and 
combining what Flink has for exactly-once semantics and out-of-core state 
backends with CDC can have very strong use cases. For this feature there should 
only be an extra dependency to the AWS Java SDK for DynamoDB, which has Apache 
License 2.0.


> Allow FlinkKinesisConsumer to adapt for AWS DynamoDB Streams
> 
>
> Key: FLINK-4582
> URL: https://issues.apache.org/jira/browse/FLINK-4582
> Project: Flink
>  Issue Type: New Feature
>  Components: Kinesis Connector, Streaming Connectors
>Reporter: Tzu-Li (Gordon) Tai
> Fix For: 1.2.0
>
>
> AWS DynamoDB is a NoSQL database service that has a CDC-like (change data 
> capture) feature called DynamoDB Streams 
> (http://docs.aws.amazon.com/amazondynamodb/latest/developerguide/Streams.html),
>  which is a stream feed of item-level table activities.
> The DynamoDB Streams shard abstraction follows that of Kinesis Streams with 
> only a slight difference in resharding behaviours, so it is possible to build 
> on the internals of our Flink Kinesis Consumer for an exactly-once DynamoDB 
> Streams source.
> I propose an API something like this:
> {code}
> DataStream dynamoItemsCdc = 
>   FlinkKinesisConsumer.asDynamoDBStream(tableNames, schema, config)
> {code}
> The feature adds more connectivity to popular AWS services for Flink, and 
> combining what Flink has for exactly-once semantics, out-of-core state 
> backends, and queryable state with CDC can have very strong use cases. For 
> this feature there should only be an extra dependency to the AWS Java SDK for 
> DynamoDB, which has Apache License 2.0.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (FLINK-4538) Implement slot allocation protocol with JobMaster

[ASF GitHub Bot (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLINK-4538?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15467390#comment-15467390
 ] 

ASF GitHub Bot commented on FLINK-4538:
---

Github user mxm commented on a diff in the pull request:

https://github.com/apache/flink/pull/2463#discussion_r77630918
  
--- Diff: 
flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/slotmanager/SlotProtocolTest.java
 ---
@@ -0,0 +1,228 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.runtime.resourcemanager.slotmanager;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.runtime.clusterframework.types.AllocationID;
+import org.apache.flink.runtime.clusterframework.types.ResourceID;
+import org.apache.flink.runtime.clusterframework.types.ResourceProfile;
+import org.apache.flink.runtime.clusterframework.types.SlotID;
+import org.apache.flink.runtime.highavailability.LeaderIdRegistry;
+import org.apache.flink.runtime.highavailability.NonHaServices;
+import org.apache.flink.runtime.jobmaster.JobMasterGateway;
+import org.apache.flink.runtime.resourcemanager.JobMasterRegistration;
+import org.apache.flink.runtime.resourcemanager.RegistrationResponse;
+import org.apache.flink.runtime.resourcemanager.ResourceManager;
+import org.apache.flink.runtime.resourcemanager.SlotRequest;
+import org.apache.flink.runtime.resourcemanager.SlotRequestRegistered;
+import org.apache.flink.runtime.rpc.TestingRpcService;
+import org.apache.flink.runtime.taskexecutor.SlotReport;
+import org.apache.flink.runtime.taskexecutor.SlotStatus;
+import org.apache.flink.runtime.taskexecutor.TaskExecutor;
+import org.apache.flink.runtime.taskexecutor.TaskExecutorGateway;
+import org.junit.AfterClass;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.mockito.Mockito;
+import scala.concurrent.Await;
+import scala.concurrent.Future;
+import scala.concurrent.duration.Duration;
+
+import java.util.Collections;
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
+
+import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.anyLong;
+import static org.mockito.Matchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.timeout;
+import static org.mockito.Mockito.verify;
+
+public class SlotProtocolTest {
+
+   private static TestingRpcService testRpcService;
+
+   @BeforeClass
+   public static void beforeClass() {
+   testRpcService = new TestingRpcService();
+
+   }
+
+   @AfterClass
+   public static void afterClass() {
+   testRpcService.stopService();
+   testRpcService = null;
+   }
+
+   @Before
+   public void beforeTest(){
+   testRpcService.clearGateways();
+   }
+
+   /**
+* Tests whether
+* 1) SlotRequest is routed to the SlotManager
+* 2) SlotRequest leads to a container allocation
+* 3) SlotRequest is confirmed
+* 4) Slot becomes available and TaskExecutor gets a SlotRequest
+*/
+   @Test
+   public void testSlotsUnavailableRequest() throws Exception {
+   final String rmAddress = "/rm1";
+   final String jmAddress = "/jm1";
+   final JobID jobID = new JobID();
+
+   testRpcService.registerGateway(jmAddress, 
mock(JobMasterGateway.class));
+
+
+   TestingSlotManager slotManager = Mockito.spy(new 
TestingSlotManager());
+   ResourceManager resourceManager =
+   new ResourceManager(testRpcService, new 
NonHaServices(rmAddress), slotManager);
+   resourceManager.start();
+
+   Future registrationFuture =
+   resourceManager.registerJobMaster(new 
JobMasterRegistration(jmAddress, jobID));

[GitHub] flink pull request #2463: [FLINK-4538][FLINK-4348] ResourceManager slot allo...

[mxm]

Github user mxm commented on a diff in the pull request:

https://github.com/apache/flink/pull/2463#discussion_r77630918
  
--- Diff: 
flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/slotmanager/SlotProtocolTest.java
 ---
@@ -0,0 +1,228 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.runtime.resourcemanager.slotmanager;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.runtime.clusterframework.types.AllocationID;
+import org.apache.flink.runtime.clusterframework.types.ResourceID;
+import org.apache.flink.runtime.clusterframework.types.ResourceProfile;
+import org.apache.flink.runtime.clusterframework.types.SlotID;
+import org.apache.flink.runtime.highavailability.LeaderIdRegistry;
+import org.apache.flink.runtime.highavailability.NonHaServices;
+import org.apache.flink.runtime.jobmaster.JobMasterGateway;
+import org.apache.flink.runtime.resourcemanager.JobMasterRegistration;
+import org.apache.flink.runtime.resourcemanager.RegistrationResponse;
+import org.apache.flink.runtime.resourcemanager.ResourceManager;
+import org.apache.flink.runtime.resourcemanager.SlotRequest;
+import org.apache.flink.runtime.resourcemanager.SlotRequestRegistered;
+import org.apache.flink.runtime.rpc.TestingRpcService;
+import org.apache.flink.runtime.taskexecutor.SlotReport;
+import org.apache.flink.runtime.taskexecutor.SlotStatus;
+import org.apache.flink.runtime.taskexecutor.TaskExecutor;
+import org.apache.flink.runtime.taskexecutor.TaskExecutorGateway;
+import org.junit.AfterClass;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.mockito.Mockito;
+import scala.concurrent.Await;
+import scala.concurrent.Future;
+import scala.concurrent.duration.Duration;
+
+import java.util.Collections;
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
+
+import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.anyLong;
+import static org.mockito.Matchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.timeout;
+import static org.mockito.Mockito.verify;
+
+public class SlotProtocolTest {
+
+   private static TestingRpcService testRpcService;
+
+   @BeforeClass
+   public static void beforeClass() {
+   testRpcService = new TestingRpcService();
+
+   }
+
+   @AfterClass
+   public static void afterClass() {
+   testRpcService.stopService();
+   testRpcService = null;
+   }
+
+   @Before
+   public void beforeTest(){
+   testRpcService.clearGateways();
+   }
+
+   /**
+* Tests whether
+* 1) SlotRequest is routed to the SlotManager
+* 2) SlotRequest leads to a container allocation
+* 3) SlotRequest is confirmed
+* 4) Slot becomes available and TaskExecutor gets a SlotRequest
+*/
+   @Test
+   public void testSlotsUnavailableRequest() throws Exception {
+   final String rmAddress = "/rm1";
+   final String jmAddress = "/jm1";
+   final JobID jobID = new JobID();
+
+   testRpcService.registerGateway(jmAddress, 
mock(JobMasterGateway.class));
+
+
+   TestingSlotManager slotManager = Mockito.spy(new 
TestingSlotManager());
+   ResourceManager resourceManager =
+   new ResourceManager(testRpcService, new 
NonHaServices(rmAddress), slotManager);
+   resourceManager.start();
+
+   Future registrationFuture =
+   resourceManager.registerJobMaster(new 
JobMasterRegistration(jmAddress, jobID));
+   try {
+   Await.ready(registrationFuture, Duration.create(5, 
TimeUnit.SECONDS));
+   } catch (Exception e) {
+   Assert.fail("JobManager registration Future didn't 
become 

[jira] [Commented] (FLINK-4565) Support for SQL IN operator

[Timo Walther (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLINK-4565?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15467383#comment-15467383
 ] 

Timo Walther commented on FLINK-4565:
-

tableEnv.sql(...) returns a Table containing a LogicalRelNode (which is a 
LogicalNode and just wraps RelNodes; Calcite already did the validation). The 
Table API returns Tables containing specific LogicalNodes depending on the 
operation (Table API does the validation). All LogicalNodes have a "construct" 
method translating the Flink logical operators to Calcites RelNode 
representation for optimization. The magic happens in 
Batch/StreamTableEnvironment.translate() where the RelNodes are optimized using 
a specific set of rules and converted to "DataSet/DataStreamRel"s also by using 
specific rules.

> Support for SQL IN operator
> ---
>
> Key: FLINK-4565
> URL: https://issues.apache.org/jira/browse/FLINK-4565
> Project: Flink
>  Issue Type: Improvement
>  Components: Table API & SQL
>Reporter: Timo Walther
>
> It seems that Flink SQL supports the uncorrelated sub-query IN operator. But 
> it should also be available in the Table API and tested.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (FLINK-4538) Implement slot allocation protocol with JobMaster

[ASF GitHub Bot (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLINK-4538?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15467382#comment-15467382
 ] 

ASF GitHub Bot commented on FLINK-4538:
---

Github user mxm commented on a diff in the pull request:

https://github.com/apache/flink/pull/2463#discussion_r77630351
  
--- Diff: 
flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/slotmanager/SlotProtocolTest.java
 ---
@@ -0,0 +1,228 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.runtime.resourcemanager.slotmanager;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.runtime.clusterframework.types.AllocationID;
+import org.apache.flink.runtime.clusterframework.types.ResourceID;
+import org.apache.flink.runtime.clusterframework.types.ResourceProfile;
+import org.apache.flink.runtime.clusterframework.types.SlotID;
+import org.apache.flink.runtime.highavailability.LeaderIdRegistry;
+import org.apache.flink.runtime.highavailability.NonHaServices;
+import org.apache.flink.runtime.jobmaster.JobMasterGateway;
+import org.apache.flink.runtime.resourcemanager.JobMasterRegistration;
+import org.apache.flink.runtime.resourcemanager.RegistrationResponse;
+import org.apache.flink.runtime.resourcemanager.ResourceManager;
+import org.apache.flink.runtime.resourcemanager.SlotRequest;
+import org.apache.flink.runtime.resourcemanager.SlotRequestRegistered;
+import org.apache.flink.runtime.rpc.TestingRpcService;
+import org.apache.flink.runtime.taskexecutor.SlotReport;
+import org.apache.flink.runtime.taskexecutor.SlotStatus;
+import org.apache.flink.runtime.taskexecutor.TaskExecutor;
+import org.apache.flink.runtime.taskexecutor.TaskExecutorGateway;
+import org.junit.AfterClass;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.mockito.Mockito;
+import scala.concurrent.Await;
+import scala.concurrent.Future;
+import scala.concurrent.duration.Duration;
+
+import java.util.Collections;
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
+
+import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.anyLong;
+import static org.mockito.Matchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.timeout;
+import static org.mockito.Mockito.verify;
+
+public class SlotProtocolTest {
+
+   private static TestingRpcService testRpcService;
+
+   @BeforeClass
+   public static void beforeClass() {
+   testRpcService = new TestingRpcService();
+
+   }
+
+   @AfterClass
+   public static void afterClass() {
+   testRpcService.stopService();
+   testRpcService = null;
+   }
+
+   @Before
+   public void beforeTest(){
+   testRpcService.clearGateways();
+   }
+
+   /**
+* Tests whether
+* 1) SlotRequest is routed to the SlotManager
+* 2) SlotRequest leads to a container allocation
+* 3) SlotRequest is confirmed
+* 4) Slot becomes available and TaskExecutor gets a SlotRequest
+*/
+   @Test
+   public void testSlotsUnavailableRequest() throws Exception {
+   final String rmAddress = "/rm1";
+   final String jmAddress = "/jm1";
+   final JobID jobID = new JobID();
+
+   testRpcService.registerGateway(jmAddress, 
mock(JobMasterGateway.class));
+
+
+   TestingSlotManager slotManager = Mockito.spy(new 
TestingSlotManager());
+   ResourceManager resourceManager =
+   new ResourceManager(testRpcService, new 
NonHaServices(rmAddress), slotManager);
+   resourceManager.start();
+
+   Future registrationFuture =
+   resourceManager.registerJobMaster(new 
JobMasterRegistration(jmAddress, jobID));

[GitHub] flink pull request #2463: [FLINK-4538][FLINK-4348] ResourceManager slot allo...

[mxm]

Github user mxm commented on a diff in the pull request:

https://github.com/apache/flink/pull/2463#discussion_r77630351
  
--- Diff: 
flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/slotmanager/SlotProtocolTest.java
 ---
@@ -0,0 +1,228 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.runtime.resourcemanager.slotmanager;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.runtime.clusterframework.types.AllocationID;
+import org.apache.flink.runtime.clusterframework.types.ResourceID;
+import org.apache.flink.runtime.clusterframework.types.ResourceProfile;
+import org.apache.flink.runtime.clusterframework.types.SlotID;
+import org.apache.flink.runtime.highavailability.LeaderIdRegistry;
+import org.apache.flink.runtime.highavailability.NonHaServices;
+import org.apache.flink.runtime.jobmaster.JobMasterGateway;
+import org.apache.flink.runtime.resourcemanager.JobMasterRegistration;
+import org.apache.flink.runtime.resourcemanager.RegistrationResponse;
+import org.apache.flink.runtime.resourcemanager.ResourceManager;
+import org.apache.flink.runtime.resourcemanager.SlotRequest;
+import org.apache.flink.runtime.resourcemanager.SlotRequestRegistered;
+import org.apache.flink.runtime.rpc.TestingRpcService;
+import org.apache.flink.runtime.taskexecutor.SlotReport;
+import org.apache.flink.runtime.taskexecutor.SlotStatus;
+import org.apache.flink.runtime.taskexecutor.TaskExecutor;
+import org.apache.flink.runtime.taskexecutor.TaskExecutorGateway;
+import org.junit.AfterClass;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.mockito.Mockito;
+import scala.concurrent.Await;
+import scala.concurrent.Future;
+import scala.concurrent.duration.Duration;
+
+import java.util.Collections;
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
+
+import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.anyLong;
+import static org.mockito.Matchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.timeout;
+import static org.mockito.Mockito.verify;
+
+public class SlotProtocolTest {
+
+   private static TestingRpcService testRpcService;
+
+   @BeforeClass
+   public static void beforeClass() {
+   testRpcService = new TestingRpcService();
+
+   }
+
+   @AfterClass
+   public static void afterClass() {
+   testRpcService.stopService();
+   testRpcService = null;
+   }
+
+   @Before
+   public void beforeTest(){
+   testRpcService.clearGateways();
+   }
+
+   /**
+* Tests whether
+* 1) SlotRequest is routed to the SlotManager
+* 2) SlotRequest leads to a container allocation
+* 3) SlotRequest is confirmed
+* 4) Slot becomes available and TaskExecutor gets a SlotRequest
+*/
+   @Test
+   public void testSlotsUnavailableRequest() throws Exception {
+   final String rmAddress = "/rm1";
+   final String jmAddress = "/jm1";
+   final JobID jobID = new JobID();
+
+   testRpcService.registerGateway(jmAddress, 
mock(JobMasterGateway.class));
+
+
+   TestingSlotManager slotManager = Mockito.spy(new 
TestingSlotManager());
+   ResourceManager resourceManager =
+   new ResourceManager(testRpcService, new 
NonHaServices(rmAddress), slotManager);
+   resourceManager.start();
+
+   Future registrationFuture =
+   resourceManager.registerJobMaster(new 
JobMasterRegistration(jmAddress, jobID));
+   try {
+   Await.ready(registrationFuture, Duration.create(5, 
TimeUnit.SECONDS));
+   } catch (Exception e) {
+   Assert.fail("JobManager registration Future didn't 
become 

[jira] [Commented] (FLINK-4538) Implement slot allocation protocol with JobMaster

[ASF GitHub Bot (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLINK-4538?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15467379#comment-15467379
 ] 

ASF GitHub Bot commented on FLINK-4538:
---

Github user mxm commented on a diff in the pull request:

https://github.com/apache/flink/pull/2463#discussion_r7763
  
--- Diff: 
flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutorGateway.java
 ---
@@ -32,4 +33,11 @@
// 

 
void notifyOfNewResourceManagerLeader(String address, UUID 
resourceManagerLeaderId);
+
+   /**
+* Send by the ResourceManager to the TaskExecutor
+* @param allocationID id for the request
+* @param resourceManagerLeaderID current leader id of the 
ResourceManager
+*/
+   void requestSlot(AllocationID allocationID, UUID 
resourceManagerLeaderID);
--- End diff --

As of now, this is just a stub but we will have to acknowledge the message. 
Will change the signature to make that clear.


> Implement slot allocation protocol with JobMaster
> -
>
> Key: FLINK-4538
> URL: https://issues.apache.org/jira/browse/FLINK-4538
> Project: Flink
>  Issue Type: Sub-task
>  Components: Cluster Management
>Reporter: Maximilian Michels
>Assignee: Maximilian Michels
>




--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[GitHub] flink pull request #2463: [FLINK-4538][FLINK-4348] ResourceManager slot allo...

[mxm]

Github user mxm commented on a diff in the pull request:

https://github.com/apache/flink/pull/2463#discussion_r7763
  
--- Diff: 
flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutorGateway.java
 ---
@@ -32,4 +33,11 @@
// 

 
void notifyOfNewResourceManagerLeader(String address, UUID 
resourceManagerLeaderId);
+
+   /**
+* Send by the ResourceManager to the TaskExecutor
+* @param allocationID id for the request
+* @param resourceManagerLeaderID current leader id of the 
ResourceManager
+*/
+   void requestSlot(AllocationID allocationID, UUID 
resourceManagerLeaderID);
--- End diff --

As of now, this is just a stub but we will have to acknowledge the message. 
Will change the signature to make that clear.


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[jira] [Commented] (FLINK-4538) Implement slot allocation protocol with JobMaster

[ASF GitHub Bot (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLINK-4538?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15467376#comment-15467376
 ] 

ASF GitHub Bot commented on FLINK-4538:
---

Github user mxm commented on a diff in the pull request:

https://github.com/apache/flink/pull/2463#discussion_r77629858
  
--- Diff: 
flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/SlotStatus.java
 ---
@@ -46,13 +47,21 @@
/** if the slot is allocated, jobId identify which job this slot is 
allocated to; else, jobId is null */
private final JobID jobID;
 
-   public SlotStatus(SlotID slotID, ResourceProfile profiler) {
-   this(slotID, profiler, null, null);
+   /** Gateway to the TaskManager which reported the SlotStatus */
+   private final TaskExecutorGateway taskExecutorGateway;
--- End diff --

It comes with the SlotReport from the TaskExecutor. Yes, it breaks 
Serializable. Will change the code to contain the String address instead.


> Implement slot allocation protocol with JobMaster
> -
>
> Key: FLINK-4538
> URL: https://issues.apache.org/jira/browse/FLINK-4538
> Project: Flink
>  Issue Type: Sub-task
>  Components: Cluster Management
>Reporter: Maximilian Michels
>Assignee: Maximilian Michels
>




--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Created] (FLINK-4582) Allow FlinkKinesisConsumer to adapt for AWS DynamoDB Streams

[Tzu-Li (Gordon) Tai (JIRA)]

Tzu-Li (Gordon) Tai created FLINK-4582:
--

 Summary: Allow FlinkKinesisConsumer to adapt for AWS DynamoDB 
Streams
 Key: FLINK-4582
 URL: https://issues.apache.org/jira/browse/FLINK-4582
 Project: Flink
  Issue Type: New Feature
  Components: Kinesis Connector, Streaming Connectors
Reporter: Tzu-Li (Gordon) Tai
 Fix For: 1.2.0


AWS DynamoDB is a NoSQL database service that has a CDC-like (change data 
capture) feature called DynamoDB Streams 
(http://docs.aws.amazon.com/amazondynamodb/latest/developerguide/Streams.html), 
which is a stream feed of item-level table activities.

The DynamoDB Streams shard abstraction follows that of Kinesis Streams with 
only a slight difference in resharding behaviours, so it is possible to build 
on the internals of our Flink Kinesis Consumer for an exactly-once DynamoDB 
Streams source.

I propose an API something like this:
{code}
DataStream dynamoItemsCdc = 
  FlinkKinesisConsumer.asDynamoDBStream(tableNames, schema, config)
{code}

The feature adds more connectivity to popular AWS services for Flink, and 
combining what Flink has for exactly-once semantics and out-of-core state 
backends with CDC can have very strong use cases. For this feature there should 
only be an extra dependency to the AWS Java SDK for DynamoDB, which has Apache 
License 2.0.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[GitHub] flink pull request #2463: [FLINK-4538][FLINK-4348] ResourceManager slot allo...

[mxm]

Github user mxm commented on a diff in the pull request:

https://github.com/apache/flink/pull/2463#discussion_r77629858
  
--- Diff: 
flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/SlotStatus.java
 ---
@@ -46,13 +47,21 @@
/** if the slot is allocated, jobId identify which job this slot is 
allocated to; else, jobId is null */
private final JobID jobID;
 
-   public SlotStatus(SlotID slotID, ResourceProfile profiler) {
-   this(slotID, profiler, null, null);
+   /** Gateway to the TaskManager which reported the SlotStatus */
+   private final TaskExecutorGateway taskExecutorGateway;
--- End diff --

It comes with the SlotReport from the TaskExecutor. Yes, it breaks 
Serializable. Will change the code to contain the String address instead.


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[GitHub] flink pull request #2463: [FLINK-4538][FLINK-4348] ResourceManager slot allo...

[mxm]

Github user mxm commented on a diff in the pull request:

https://github.com/apache/flink/pull/2463#discussion_r77628362
  
--- Diff: 
flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/LeaderIdRegistry.java
 ---
@@ -15,11 +15,27 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
+package org.apache.flink.runtime.highavailability;
 
-package org.apache.flink.runtime.resourcemanager;
+import java.util.UUID;
 
-import java.io.Serializable;
+/**
+ * Registry class to keep track of the current leader ID.
+ */
+public class LeaderIdRegistry {
--- End diff --

In order to pass it on to components who want to retrieve the current 
leader UUID. Passing on only a single reference wouldn't work.


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[jira] [Commented] (FLINK-4538) Implement slot allocation protocol with JobMaster

[ASF GitHub Bot (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLINK-4538?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15467361#comment-15467361
 ] 

ASF GitHub Bot commented on FLINK-4538:
---

Github user mxm commented on a diff in the pull request:

https://github.com/apache/flink/pull/2463#discussion_r77629249
  
--- Diff: 
flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManager.java
 ---
@@ -131,9 +149,16 @@ public RegistrationResponse apply(final 
JobMasterGateway jobMasterGateway) {
 * @return Slot assignment
 */
@RpcMethod
-   public SlotAssignment requestSlot(SlotRequest slotRequest) {
-   System.out.println("SlotRequest: " + slotRequest);
-   return new SlotAssignment();
+   public SlotRequestRegistered requestSlot(SlotRequest slotRequest) {
+   final JobID jobId = slotRequest.getJobId();
+   final JobMasterGateway jobMasterGateway = 
jobMasterGateways.get(jobId);
+
+   if (jobMasterGateway != null) {
+   return slotManager.requestSlot(slotRequest);
+   } else {
+   LOG.info("Ignoring slot request for unknown JobMaster 
with JobID {}", jobId);
+   return null;
--- End diff --

The rationale here was to simply ignore this request because the JobManager 
is not registered. You're right, probably better to reply with a meaningful 
answer. 


> Implement slot allocation protocol with JobMaster
> -
>
> Key: FLINK-4538
> URL: https://issues.apache.org/jira/browse/FLINK-4538
> Project: Flink
>  Issue Type: Sub-task
>  Components: Cluster Management
>Reporter: Maximilian Michels
>Assignee: Maximilian Michels
>




--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (FLINK-4538) Implement slot allocation protocol with JobMaster

[ASF GitHub Bot (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLINK-4538?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15467350#comment-15467350
 ] 

ASF GitHub Bot commented on FLINK-4538:
---

Github user mxm commented on a diff in the pull request:

https://github.com/apache/flink/pull/2463#discussion_r77628362
  
--- Diff: 
flink-runtime/src/main/java/org/apache/flink/runtime/highavailability/LeaderIdRegistry.java
 ---
@@ -15,11 +15,27 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
+package org.apache.flink.runtime.highavailability;
 
-package org.apache.flink.runtime.resourcemanager;
+import java.util.UUID;
 
-import java.io.Serializable;
+/**
+ * Registry class to keep track of the current leader ID.
+ */
+public class LeaderIdRegistry {
--- End diff --

In order to pass it on to components who want to retrieve the current 
leader UUID. Passing on only a single reference wouldn't work.


> Implement slot allocation protocol with JobMaster
> -
>
> Key: FLINK-4538
> URL: https://issues.apache.org/jira/browse/FLINK-4538
> Project: Flink
>  Issue Type: Sub-task
>  Components: Cluster Management
>Reporter: Maximilian Michels
>Assignee: Maximilian Michels
>




--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[GitHub] flink pull request #2463: [FLINK-4538][FLINK-4348] ResourceManager slot allo...

[mxm]

Github user mxm commented on a diff in the pull request:

https://github.com/apache/flink/pull/2463#discussion_r77629249
  
--- Diff: 
flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManager.java
 ---
@@ -131,9 +149,16 @@ public RegistrationResponse apply(final 
JobMasterGateway jobMasterGateway) {
 * @return Slot assignment
 */
@RpcMethod
-   public SlotAssignment requestSlot(SlotRequest slotRequest) {
-   System.out.println("SlotRequest: " + slotRequest);
-   return new SlotAssignment();
+   public SlotRequestRegistered requestSlot(SlotRequest slotRequest) {
+   final JobID jobId = slotRequest.getJobId();
+   final JobMasterGateway jobMasterGateway = 
jobMasterGateways.get(jobId);
+
+   if (jobMasterGateway != null) {
+   return slotManager.requestSlot(slotRequest);
+   } else {
+   LOG.info("Ignoring slot request for unknown JobMaster 
with JobID {}", jobId);
+   return null;
--- End diff --

The rationale here was to simply ignore this request because the JobManager 
is not registered. You're right, probably better to reply with a meaningful 
answer. 


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[jira] [Commented] (FLINK-4538) Implement slot allocation protocol with JobMaster

[ASF GitHub Bot (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLINK-4538?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15467353#comment-15467353
 ] 

ASF GitHub Bot commented on FLINK-4538:
---

Github user mxm commented on a diff in the pull request:

https://github.com/apache/flink/pull/2463#discussion_r77628600
  
--- Diff: 
flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManager.java
 ---
@@ -52,15 +58,28 @@
  * 
  */
 public class ResourceManager extends RpcEndpoint {
-   private final Map jobMasterGateways;
+
+   private final Logger LOG = LoggerFactory.getLogger(getClass());
--- End diff --

No particular reason other than I want to make sure future subclasses log 
with the correct class name.


> Implement slot allocation protocol with JobMaster
> -
>
> Key: FLINK-4538
> URL: https://issues.apache.org/jira/browse/FLINK-4538
> Project: Flink
>  Issue Type: Sub-task
>  Components: Cluster Management
>Reporter: Maximilian Michels
>Assignee: Maximilian Michels
>




--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[GitHub] flink pull request #2463: [FLINK-4538][FLINK-4348] ResourceManager slot allo...

[mxm]

Github user mxm commented on a diff in the pull request:

https://github.com/apache/flink/pull/2463#discussion_r77628600
  
--- Diff: 
flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManager.java
 ---
@@ -52,15 +58,28 @@
  * 
  */
 public class ResourceManager extends RpcEndpoint {
-   private final Map jobMasterGateways;
+
+   private final Logger LOG = LoggerFactory.getLogger(getClass());
--- End diff --

No particular reason other than I want to make sure future subclasses log 
with the correct class name.


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[GitHub] flink issue #2452: [Flink-4450] add a new module "flink-apache-storm" to sup...

[mxm]

Github user mxm commented on the issue:

https://github.com/apache/flink/pull/2452
  
It looks like this is basically a copy of the old code adapted to the Storm 
1.0.0 dependency. This would duplicate a lot of code. I've started a thread on 
the mailing list, whether we want to add a new module for Storm 1.0.0 or stop 
supporting old Storm versions.


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[GitHub] flink issue #2452: [Flink-4450] add a new module "flink-apache-storm" to sup...

[mxm]

Github user mxm commented on the issue:

https://github.com/apache/flink/pull/2452
  
I think you ran out of memory. Travis CI build passed, so the PR's testing 
state seems to be fine.


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[jira] [Created] (FLINK-4581) Table API throws "No suitable driver found for jdbc:calcite"

[Timo Walther (JIRA)]

Timo Walther created FLINK-4581:
---

 Summary: Table API throws "No suitable driver found for 
jdbc:calcite"
 Key: FLINK-4581
 URL: https://issues.apache.org/jira/browse/FLINK-4581
 Project: Flink
  Issue Type: Bug
  Components: Table API & SQL
Reporter: Timo Walther


It seems that in certain cases the internal Calcite JDBC driver cannot be 
found. We should either try to get rid of the entire JDBC invocation or fix 
this bug.

>From ML: 
>http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/Stream-sql-query-in-Flink-tp8928.html

{code}
org.apache.flink.client.program.ProgramInvocationException: The main method
caused an error.
at
org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:524)
at
org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:403)
at
org.apache.flink.client.program.ClusterClient.run(ClusterClient.java:331)
at 
org.apache.flink.client.CliFrontend.executeProgram(CliFrontend.java:777)
at org.apache.flink.client.CliFrontend.run(CliFrontend.java:253)
at
org.apache.flink.client.CliFrontend.parseParameters(CliFrontend.java:1005)
at org.apache.flink.client.CliFrontend.main(CliFrontend.java:1048)
Caused by: java.lang.RuntimeException: java.sql.SQLException: No suitable
driver found for jdbc:calcite:
at org.apache.calcite.tools.Frameworks.withPrepare(Frameworks.java:151)
at org.apache.calcite.tools.Frameworks.withPlanner(Frameworks.java:106)
at org.apache.calcite.tools.Frameworks.withPlanner(Frameworks.java:127)
at
org.apache.flink.api.table.FlinkRelBuilder$.create(FlinkRelBuilder.scala:56)
at
org.apache.flink.api.table.TableEnvironment.(TableEnvironment.scala:73)
at
org.apache.flink.api.table.StreamTableEnvironment.(StreamTableEnvironment.scala:58)
at
org.apache.flink.api.java.table.StreamTableEnvironment.(StreamTableEnvironment.scala:45)
at
org.apache.flink.api.table.TableEnvironment$.getTableEnvironment(TableEnvironment.scala:376)
at
org.apache.flink.api.table.TableEnvironment.getTableEnvironment(TableEnvironment.scala)
at 
org.myorg.quickstart.ReadingFromKafka2.main(ReadingFromKafka2.java:48)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at
sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at
sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at
org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:509)
... 6 more
Caused by: java.sql.SQLException: No suitable driver found for jdbc:calcite:
at java.sql.DriverManager.getConnection(DriverManager.java:689)
at java.sql.DriverManager.getConnection(DriverManager.java:208)
at org.apache.calcite.tools.Frameworks.withPrepare(Frameworks.java:144)
... 20 more
{code}



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Assigned] (FLINK-4576) Low Watermark Service in JobManager for Streaming Sources

[Tzu-Li (Gordon) Tai (JIRA)]

 [ 
https://issues.apache.org/jira/browse/FLINK-4576?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Tzu-Li (Gordon) Tai reassigned FLINK-4576:
--

Assignee: Tzu-Li (Gordon) Tai

> Low Watermark Service in JobManager for Streaming Sources
> -
>
> Key: FLINK-4576
> URL: https://issues.apache.org/jira/browse/FLINK-4576
> Project: Flink
>  Issue Type: New Feature
>  Components: JobManager, Streaming, TaskManager
>Reporter: Tzu-Li (Gordon) Tai
>Assignee: Tzu-Li (Gordon) Tai
>Priority: Blocker
> Fix For: 1.2.0
>
>
> As per discussion in FLINK-4341 by [~aljoscha] and [~StephanEwen], we need a 
> low watermark service in the JobManager to support transparent resharding / 
> partition discovery for our Kafka and Kinesis consumers (and any future 
> streaming connectors in general for which the external system may elastically 
> scale up and down independently of the parallelism of sources in Flink). The 
> main idea is to let source subtasks that don't emit their own watermarks 
> (because they currently don't have data partitions to consume) emit the low 
> watermark across all subtasks, instead of simply emitting a Long.MAX_VALUE 
> watermark and forbidding them to be assigned partitions in the future.
> The proposed implementation, from a high-level: a {{LowWatermarkCoordinator}} 
> will be added to execution graphs, periodically triggering only the source 
> vertices with a {{RetrieveLowWatermark}} message. The tasks reply to the 
> JobManager through the actor gateway (or a new interface after FLINK-4456 
> gets merged) with a {{ReplyLowWatermark}} message. When the coordinator 
> collects all low watermarks for a particular source vertex and determines the 
> aggregated low watermark for this round (accounting only values that are 
> larger than the aggregated low watermark of the last round), it sends a 
> {{NotifyNewLowWatermark}} message to the source vertex's tasks.
> The messages will only be relevant to tasks that implement an internal 
> {{LowWatermarkCooperatingTask}} interface. For now, only {{SourceStreamTask}} 
> should implement {{LowWatermarkCooperatingTask}}.
> Source functions should implement a public {{LowWatermarkListener}} interface 
> if they wish to get notified of the aggregated low watermarks across 
> subtasks. Connectors like the Kinesis consumer can choose to emit this 
> watermark if the subtask currently does not have any shards, so that 
> downstream operators may still properly advance time windows (implementation 
> for this is tracked as a separate issue).
> Overall, the service will include -
> New messages between JobManager <-> TaskManager:
> {{RetrieveLowWatermark(jobId, jobVertexId, taskId, timestamp)}}
> {{ReplyLowWatermark(jobId, jobVertexId, taskId, currentLowWatermark)}}
> {{NotifyNewLowWatermark(jobId, jobVertexId, taskId, newLowWatermark, 
> timestamp)}}
> New internal task interface {{LowWatermarkCooperatingTask}} in flink-runtime
> New public interface {{LowWatermarkListener}} in flink-streaming-java
> Might also need to extend {{SourceFunction.SourceContext}} to support 
> retrieving the current low watermark of sources.
> Any feedback for this is appreciated!



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Resolved] (FLINK-4559) Kinesis Producer not setting credentials provider properly when AWS_CREDENTIALS_PROVIDER is "AUTO"

[Tzu-Li (Gordon) Tai (JIRA)]

 [ 
https://issues.apache.org/jira/browse/FLINK-4559?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Tzu-Li (Gordon) Tai resolved FLINK-4559.

Resolution: Resolved

> Kinesis Producer not setting credentials provider properly when 
> AWS_CREDENTIALS_PROVIDER is "AUTO"
> --
>
> Key: FLINK-4559
> URL: https://issues.apache.org/jira/browse/FLINK-4559
> Project: Flink
>  Issue Type: Bug
>  Components: Kinesis Connector
>Reporter: Tzu-Li (Gordon) Tai
>Assignee: Tzu-Li (Gordon) Tai
>Priority: Minor
> Fix For: 1.2.0
>
>
> If the {{AWS_CREDENTIAL_PROVIDER}} option is set to {{AUTO}}, 
> {{AWSUtils.getCredentialsProvider}} will return {{null}}, so 
> {{setCredentialsProvider}} should not be explicitly called on the internally 
> built {{KinesisProducerConfiguration}}.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (FLINK-4559) Kinesis Producer not setting credentials provider properly when AWS_CREDENTIALS_PROVIDER is "AUTO"

[Tzu-Li (Gordon) Tai (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLINK-4559?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15467287#comment-15467287
 ] 

Tzu-Li (Gordon) Tai commented on FLINK-4559:


Resolved for master in 
http://git-wip-us.apache.org/repos/asf/flink/commit/cab76f6e

> Kinesis Producer not setting credentials provider properly when 
> AWS_CREDENTIALS_PROVIDER is "AUTO"
> --
>
> Key: FLINK-4559
> URL: https://issues.apache.org/jira/browse/FLINK-4559
> Project: Flink
>  Issue Type: Bug
>  Components: Kinesis Connector
>Reporter: Tzu-Li (Gordon) Tai
>Assignee: Tzu-Li (Gordon) Tai
>Priority: Minor
> Fix For: 1.2.0
>
>
> If the {{AWS_CREDENTIAL_PROVIDER}} option is set to {{AUTO}}, 
> {{AWSUtils.getCredentialsProvider}} will return {{null}}, so 
> {{setCredentialsProvider}} should not be explicitly called on the internally 
> built {{KinesisProducerConfiguration}}.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (FLINK-4559) Kinesis Producer not setting credentials provider properly when AWS_CREDENTIALS_PROVIDER is "AUTO"

[ASF GitHub Bot (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLINK-4559?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15467281#comment-15467281
 ] 

ASF GitHub Bot commented on FLINK-4559:
---

Github user asfgit closed the pull request at:

https://github.com/apache/flink/pull/2470


> Kinesis Producer not setting credentials provider properly when 
> AWS_CREDENTIALS_PROVIDER is "AUTO"
> --
>
> Key: FLINK-4559
> URL: https://issues.apache.org/jira/browse/FLINK-4559
> Project: Flink
>  Issue Type: Bug
>  Components: Kinesis Connector
>Reporter: Tzu-Li (Gordon) Tai
>Assignee: Tzu-Li (Gordon) Tai
>Priority: Minor
> Fix For: 1.2.0
>
>
> If the {{AWS_CREDENTIAL_PROVIDER}} option is set to {{AUTO}}, 
> {{AWSUtils.getCredentialsProvider}} will return {{null}}, so 
> {{setCredentialsProvider}} should not be explicitly called on the internally 
> built {{KinesisProducerConfiguration}}.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[GitHub] flink pull request #2470: [FLINK-4559][kinesis-connector] Fix AWSUtil.getCre...

[asfgit]

Github user asfgit closed the pull request at:

https://github.com/apache/flink/pull/2470


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[jira] [Updated] (FLINK-4559) Kinesis Producer not setting credentials provider properly when AWS_CREDENTIALS_PROVIDER is "AUTO"

[Tzu-Li (Gordon) Tai (JIRA)]

 [ 
https://issues.apache.org/jira/browse/FLINK-4559?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Tzu-Li (Gordon) Tai updated FLINK-4559:
---
Fix Version/s: (was: 1.1.3)
   1.2.0

> Kinesis Producer not setting credentials provider properly when 
> AWS_CREDENTIALS_PROVIDER is "AUTO"
> --
>
> Key: FLINK-4559
> URL: https://issues.apache.org/jira/browse/FLINK-4559
> Project: Flink
>  Issue Type: Bug
>  Components: Kinesis Connector
>Reporter: Tzu-Li (Gordon) Tai
>Assignee: Tzu-Li (Gordon) Tai
>Priority: Minor
> Fix For: 1.2.0
>
>
> If the {{AWS_CREDENTIAL_PROVIDER}} option is set to {{AUTO}}, 
> {{AWSUtils.getCredentialsProvider}} will return {{null}}, so 
> {{setCredentialsProvider}} should not be explicitly called on the internally 
> built {{KinesisProducerConfiguration}}.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Updated] (FLINK-4559) Kinesis Producer not setting credentials provider properly when AWS_CREDENTIALS_PROVIDER is "AUTO"

[Tzu-Li (Gordon) Tai (JIRA)]

 [ 
https://issues.apache.org/jira/browse/FLINK-4559?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Tzu-Li (Gordon) Tai updated FLINK-4559:
---
Affects Version/s: (was: 1.1.2)
   (was: 1.1.1)
   (was: 1.1.0)

> Kinesis Producer not setting credentials provider properly when 
> AWS_CREDENTIALS_PROVIDER is "AUTO"
> --
>
> Key: FLINK-4559
> URL: https://issues.apache.org/jira/browse/FLINK-4559
> Project: Flink
>  Issue Type: Bug
>  Components: Kinesis Connector
>Reporter: Tzu-Li (Gordon) Tai
>Assignee: Tzu-Li (Gordon) Tai
>Priority: Minor
> Fix For: 1.2.0
>
>
> If the {{AWS_CREDENTIAL_PROVIDER}} option is set to {{AUTO}}, 
> {{AWSUtils.getCredentialsProvider}} will return {{null}}, so 
> {{setCredentialsProvider}} should not be explicitly called on the internally 
> built {{KinesisProducerConfiguration}}.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (FLINK-4559) Kinesis Producer not setting credentials provider properly when AWS_CREDENTIALS_PROVIDER is "AUTO"

[ASF GitHub Bot (JIRA)]

[ 
https://issues.apache.org/jira/browse/FLINK-4559?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15467271#comment-15467271
 ] 

ASF GitHub Bot commented on FLINK-4559:
---

Github user tzulitai commented on the issue:

https://github.com/apache/flink/pull/2470
  
Merging ...


> Kinesis Producer not setting credentials provider properly when 
> AWS_CREDENTIALS_PROVIDER is "AUTO"
> --
>
> Key: FLINK-4559
> URL: https://issues.apache.org/jira/browse/FLINK-4559
> Project: Flink
>  Issue Type: Bug
>  Components: Kinesis Connector
>Reporter: Tzu-Li (Gordon) Tai
>Assignee: Tzu-Li (Gordon) Tai
>Priority: Minor
> Fix For: 1.2.0
>
>
> If the {{AWS_CREDENTIAL_PROVIDER}} option is set to {{AUTO}}, 
> {{AWSUtils.getCredentialsProvider}} will return {{null}}, so 
> {{setCredentialsProvider}} should not be explicitly called on the internally 
> built {{KinesisProducerConfiguration}}.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)