[GitHub] [flink] rmetzger commented on issue #9938: [FLINK-14175] [flink-connector-kinesis] Update KPL version to 0.13.1
rmetzger commented on issue #9938: [FLINK-14175] [flink-connector-kinesis] Update KPL version to 0.13.1 URL: https://github.com/apache/flink/pull/9938#issuecomment-544990038 Related: https://github.com/apache/flink/pull/9939 This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #9747: [FLINK-13985] Use unsafe memory for managed memory
flinkbot edited a comment on issue #9747: [FLINK-13985] Use unsafe memory for managed memory URL: https://github.com/apache/flink/pull/9747#issuecomment-534087338 ## CI report: * 1967191401dfac1a56eb3d82a0e97bd080e4c588 : SUCCESS [Build](https://travis-ci.com/flink-ci/flink/builds/128743926) * 79f81fd5897e2f42a4918178bda23cbcc9174514 : SUCCESS [Build](https://travis-ci.com/flink-ci/flink/builds/131974566) * 92e38d91c935017fab82f6a53a2b4bcbe528376f : SUCCESS [Build](https://travis-ci.com/flink-ci/flink/builds/132123636) * addcfd190caf98fc8bbd64cb65b0938b4069 : SUCCESS [Build](https://travis-ci.com/flink-ci/flink/builds/132989666) * 866835a644e4027c3c334561f1754c37fee0b12d : SUCCESS [Build](https://travis-ci.com/flink-ci/flink/builds/132998270) Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] TisonKun commented on a change in pull request #9832: [FLINK-11843] Bind lifespan of Dispatcher to leader session
TisonKun commented on a change in pull request #9832: [FLINK-11843] Bind
lifespan of Dispatcher to leader session
URL: https://github.com/apache/flink/pull/9832#discussion_r337549852
##
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java
##
@@ -869,78 +823,6 @@ private void jobMasterFailed(JobID jobId, Throwable
cause) {
return optionalJobInformation;
}
- //--
- // Leader contender
- //--
-
- /**
-* Callback method when current resourceManager is granted leadership.
-*
-* @param newLeaderSessionID unique leadershipID
-*/
- @Override
- public void grantLeadership(final UUID newLeaderSessionID) {
- runAsyncWithoutFencing(
- () -> {
- log.info("Dispatcher {} was granted leadership
with fencing token {}", getAddress(), newLeaderSessionID);
-
- final CompletableFuture>
recoveredJobsFuture = recoveryOperation.thenApplyAsync(
- FunctionUtils.uncheckedFunction(ignored
-> recoverJobs()),
- getRpcService().getExecutor());
-
- final CompletableFuture
fencingTokenFuture = recoveredJobsFuture.thenComposeAsync(
- (Collection recoveredJobs) ->
tryAcceptLeadershipAndRunJobs(newLeaderSessionID, recoveredJobs),
- getUnfencedMainThreadExecutor());
-
- final CompletableFuture
confirmationFuture = fencingTokenFuture.thenCombineAsync(
- recoveredJobsFuture,
-
BiFunctionWithException.unchecked((Boolean confirmLeadership,
Collection recoveredJobs) -> {
- if (confirmLeadership) {
-
leaderElectionService.confirmLeadership(newLeaderSessionID, getAddress());
- } else {
- for (JobGraph
recoveredJob : recoveredJobs) {
-
jobGraphStore.releaseJobGraph(recoveredJob.getJobID());
- }
- }
- return null;
- }),
- getRpcService().getExecutor());
-
- confirmationFuture.whenComplete(
- (Void ignored, Throwable throwable) -> {
- if (throwable != null) {
- onFatalError(
- new
DispatcherException(
-
String.format("Failed to take leadership with session id %s.",
newLeaderSessionID),
-
(ExceptionUtils.stripCompletionException(throwable;
- }
- });
-
- recoveryOperation = confirmationFuture;
- });
- }
-
- private CompletableFuture tryAcceptLeadershipAndRunJobs(UUID
newLeaderSessionID, Collection recoveredJobs) {
- final DispatcherId dispatcherId =
DispatcherId.fromUuid(newLeaderSessionID);
-
- if (leaderElectionService.hasLeadership(newLeaderSessionID)) {
- log.debug("Dispatcher {} accepted leadership with
fencing token {}. Start recovered jobs.", getAddress(), dispatcherId);
- setNewFencingToken(dispatcherId);
-
- Collection> runFutures = new
ArrayList<>(recoveredJobs.size());
-
- for (JobGraph recoveredJob : recoveredJobs) {
- final CompletableFuture runFuture =
waitForTerminatingJobManager(recoveredJob.getJobID(), recoveredJob,
this::runJob);
- runFutures.add(runFuture);
- }
-
- return
FutureUtils.waitForAll(runFutures).thenApply(ignored -> true);
- } else {
- log.debug("Dispatcher {} lost leadership before
accepting it. Stop recovering jobs for fencing token {}.", getAddress(),
dispatcherId);
- return CompletableFuture.completedFuture(false);
- }
- }
-
private CompletableFuture
[GitHub] [flink] TisonKun commented on a change in pull request #9832: [FLINK-11843] Bind lifespan of Dispatcher to leader session
TisonKun commented on a change in pull request #9832: [FLINK-11843] Bind
lifespan of Dispatcher to leader session
URL: https://github.com/apache/flink/pull/9832#discussion_r337546924
##
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java
##
@@ -197,19 +191,37 @@ public void onStart() throws Exception {
onFatalError(exception);
throw exception;
}
+
+ startRecoveredJobs();
Review comment:
I agree. I am rearranging the design document attached in FLINK-10333 for
emphasizing our proposal leader store based high-availability service, which
includes the removal of `JobGraphListener` since it is originally introduced
for patching coordinating concurrent modification.
I am going to start a discuss thread recently to see if our community is
glad to accept it. Does it sound good to you? If so, what is a good time to
start this thread?(I can finish the preparation in this week) I'm afraid that
the community is blocked by many ongoing threads but IMO
stability(high-availability) is critical component we'd better take more
priority with.
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[jira] [Closed] (FLINK-14245) Kafka010ProducerITCase>KafkaProducerTestBase.testOneToOneAtLeastOnceRegularSink fails on Travis
[
https://issues.apache.org/jira/browse/FLINK-14245?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Piotr Nowojski closed FLINK-14245.
--
Resolution: Fixed
This was probably a duplicated issue of
https://issues.apache.org/jira/browse/FLINK-14235
I hope this issue should be fixed by *c31e44e* (merged to the master branch).
Closing the ticket for now, please re-open if you get the same failure message
in the same test on a branch that includes this fix.
> Kafka010ProducerITCase>KafkaProducerTestBase.testOneToOneAtLeastOnceRegularSink
> fails on Travis
> ---
>
> Key: FLINK-14245
> URL: https://issues.apache.org/jira/browse/FLINK-14245
> Project: Flink
> Issue Type: Bug
> Components: Connectors / Kafka
>Affects Versions: 1.10.0
>Reporter: Till Rohrmann
>Priority: Critical
> Labels: test-stability
> Fix For: 1.10.0
>
>
> The
> {{Kafka010ProducerITCase>KafkaProducerTestBase.testOneToOneAtLeastOnceRegularSink}}
> fails on Travis with
> {code}
> Test
> testOneToOneAtLeastOnceRegularSink(org.apache.flink.streaming.connectors.kafka.Kafka010ProducerITCase)
> failed with:
> java.lang.AssertionError: Job should fail!
> at org.junit.Assert.fail(Assert.java:88)
> at
> org.apache.flink.streaming.connectors.kafka.KafkaProducerTestBase.testOneToOneAtLeastOnce(KafkaProducerTestBase.java:280)
> at
> org.apache.flink.streaming.connectors.kafka.KafkaProducerTestBase.testOneToOneAtLeastOnceRegularSink(KafkaProducerTestBase.java:206)
> 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.junit.runners.model.FrameworkMethod$1.runReflectiveCall(FrameworkMethod.java:50)
> at
> org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:12)
> at
> org.junit.runners.model.FrameworkMethod.invokeExplosively(FrameworkMethod.java:47)
> at
> org.junit.internal.runners.statements.InvokeMethod.evaluate(InvokeMethod.java:17)
> at org.junit.rules.TestWatcher$1.evaluate(TestWatcher.java:55)
> at org.junit.rules.RunRules.evaluate(RunRules.java:20)
> at org.junit.runners.ParentRunner.runLeaf(ParentRunner.java:325)
> at
> org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:78)
> at
> org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:57)
> at org.junit.runners.ParentRunner$3.run(ParentRunner.java:290)
> at org.junit.runners.ParentRunner$1.schedule(ParentRunner.java:71)
> at org.junit.runners.ParentRunner.runChildren(ParentRunner.java:288)
> at org.junit.runners.ParentRunner.access$000(ParentRunner.java:58)
> at org.junit.runners.ParentRunner$2.evaluate(ParentRunner.java:268)
> at
> org.junit.internal.runners.statements.RunBefores.evaluate(RunBefores.java:26)
> at
> org.junit.internal.runners.statements.RunAfters.evaluate(RunAfters.java:27)
> at org.junit.rules.ExternalResource$1.evaluate(ExternalResource.java:48)
> at org.junit.rules.ExternalResource$1.evaluate(ExternalResource.java:48)
> at org.junit.rules.RunRules.evaluate(RunRules.java:20)
> at org.junit.runners.ParentRunner.run(ParentRunner.java:363)
> at
> org.apache.maven.surefire.junit4.JUnit4Provider.execute(JUnit4Provider.java:365)
> at
> org.apache.maven.surefire.junit4.JUnit4Provider.executeWithRerun(JUnit4Provider.java:273)
> at
> org.apache.maven.surefire.junit4.JUnit4Provider.executeTestSet(JUnit4Provider.java:238)
> at
> org.apache.maven.surefire.junit4.JUnit4Provider.invoke(JUnit4Provider.java:159)
> at
> org.apache.maven.surefire.booter.ForkedBooter.invokeProviderInSameClassLoader(ForkedBooter.java:384)
> at
> org.apache.maven.surefire.booter.ForkedBooter.runSuitesInProcess(ForkedBooter.java:345)
> at
> org.apache.maven.surefire.booter.ForkedBooter.execute(ForkedBooter.java:126)
> at
> org.apache.maven.surefire.booter.ForkedBooter.main(ForkedBooter.java:418)
> {code}
> https://api.travis-ci.com/v3/job/239463674/log.txt
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[jira] [Closed] (FLINK-14224) Kafka010ProducerITCase>KafkaProducerTestBase.testOneToOneAtLeastOnceCustomOperator fails on Travis
[
https://issues.apache.org/jira/browse/FLINK-14224?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Piotr Nowojski closed FLINK-14224.
--
Resolution: Duplicate
This was probably a duplicated issue of
https://issues.apache.org/jira/browse/FLINK-14235
I hope this issue should be fixed by *c31e44e* (merged to the master branch).
Closing the ticket for now, please re-open if you get the same failure message
in the same test on a branch that includes this fix.
> Kafka010ProducerITCase>KafkaProducerTestBase.testOneToOneAtLeastOnceCustomOperator
> fails on Travis
> --
>
> Key: FLINK-14224
> URL: https://issues.apache.org/jira/browse/FLINK-14224
> Project: Flink
> Issue Type: Bug
> Components: Connectors / Kafka
>Affects Versions: 1.10.0
>Reporter: Till Rohrmann
>Priority: Blocker
> Labels: test-stability
> Fix For: 1.10.0
>
>
> The
> {{Kafka010ProducerITCase>KafkaProducerTestBase.testOneToOneAtLeastOnceCustomOperator}}
> fails on Travis with
> {code}
> Test
> testOneToOneAtLeastOnceCustomOperator(org.apache.flink.streaming.connectors.kafka.Kafka010ProducerITCase)
> failed with:
> java.lang.AssertionError: Job should fail!
> at org.junit.Assert.fail(Assert.java:88)
> at
> org.apache.flink.streaming.connectors.kafka.KafkaProducerTestBase.testOneToOneAtLeastOnce(KafkaProducerTestBase.java:280)
> at
> org.apache.flink.streaming.connectors.kafka.KafkaProducerTestBase.testOneToOneAtLeastOnceCustomOperator(KafkaProducerTestBase.java:214)
> 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.junit.runners.model.FrameworkMethod$1.runReflectiveCall(FrameworkMethod.java:50)
> at
> org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:12)
> at
> org.junit.runners.model.FrameworkMethod.invokeExplosively(FrameworkMethod.java:47)
> at
> org.junit.internal.runners.statements.InvokeMethod.evaluate(InvokeMethod.java:17)
> at org.junit.rules.TestWatcher$1.evaluate(TestWatcher.java:55)
> at org.junit.rules.RunRules.evaluate(RunRules.java:20)
> at org.junit.runners.ParentRunner.runLeaf(ParentRunner.java:325)
> at
> org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:78)
> at
> org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:57)
> at org.junit.runners.ParentRunner$3.run(ParentRunner.java:290)
> at org.junit.runners.ParentRunner$1.schedule(ParentRunner.java:71)
> at org.junit.runners.ParentRunner.runChildren(ParentRunner.java:288)
> at org.junit.runners.ParentRunner.access$000(ParentRunner.java:58)
> at org.junit.runners.ParentRunner$2.evaluate(ParentRunner.java:268)
> at
> org.junit.internal.runners.statements.RunBefores.evaluate(RunBefores.java:26)
> at
> org.junit.internal.runners.statements.RunAfters.evaluate(RunAfters.java:27)
> at org.junit.rules.ExternalResource$1.evaluate(ExternalResource.java:48)
> at org.junit.rules.ExternalResource$1.evaluate(ExternalResource.java:48)
> at org.junit.rules.RunRules.evaluate(RunRules.java:20)
> at org.junit.runners.ParentRunner.run(ParentRunner.java:363)
> at
> org.apache.maven.surefire.junit4.JUnit4Provider.execute(JUnit4Provider.java:365)
> at
> org.apache.maven.surefire.junit4.JUnit4Provider.executeWithRerun(JUnit4Provider.java:273)
> at
> org.apache.maven.surefire.junit4.JUnit4Provider.executeTestSet(JUnit4Provider.java:238)
> at
> org.apache.maven.surefire.junit4.JUnit4Provider.invoke(JUnit4Provider.java:159)
> at
> org.apache.maven.surefire.booter.ForkedBooter.invokeProviderInSameClassLoader(ForkedBooter.java:384)
> at
> org.apache.maven.surefire.booter.ForkedBooter.runSuitesInProcess(ForkedBooter.java:345)
> at
> org.apache.maven.surefire.booter.ForkedBooter.execute(ForkedBooter.java:126)
> at
> org.apache.maven.surefire.booter.ForkedBooter.main(ForkedBooter.java:418)
> {code}
> https://api.travis-ci.com/v3/job/238920411/log.txt
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[jira] [Assigned] (FLINK-14224) Kafka010ProducerITCase>KafkaProducerTestBase.testOneToOneAtLeastOnceCustomOperator fails on Travis
[
https://issues.apache.org/jira/browse/FLINK-14224?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Piotr Nowojski reassigned FLINK-14224:
--
Assignee: (was: Alex)
> Kafka010ProducerITCase>KafkaProducerTestBase.testOneToOneAtLeastOnceCustomOperator
> fails on Travis
> --
>
> Key: FLINK-14224
> URL: https://issues.apache.org/jira/browse/FLINK-14224
> Project: Flink
> Issue Type: Bug
> Components: Connectors / Kafka
>Affects Versions: 1.10.0
>Reporter: Till Rohrmann
>Priority: Blocker
> Labels: test-stability
> Fix For: 1.10.0
>
>
> The
> {{Kafka010ProducerITCase>KafkaProducerTestBase.testOneToOneAtLeastOnceCustomOperator}}
> fails on Travis with
> {code}
> Test
> testOneToOneAtLeastOnceCustomOperator(org.apache.flink.streaming.connectors.kafka.Kafka010ProducerITCase)
> failed with:
> java.lang.AssertionError: Job should fail!
> at org.junit.Assert.fail(Assert.java:88)
> at
> org.apache.flink.streaming.connectors.kafka.KafkaProducerTestBase.testOneToOneAtLeastOnce(KafkaProducerTestBase.java:280)
> at
> org.apache.flink.streaming.connectors.kafka.KafkaProducerTestBase.testOneToOneAtLeastOnceCustomOperator(KafkaProducerTestBase.java:214)
> 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.junit.runners.model.FrameworkMethod$1.runReflectiveCall(FrameworkMethod.java:50)
> at
> org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:12)
> at
> org.junit.runners.model.FrameworkMethod.invokeExplosively(FrameworkMethod.java:47)
> at
> org.junit.internal.runners.statements.InvokeMethod.evaluate(InvokeMethod.java:17)
> at org.junit.rules.TestWatcher$1.evaluate(TestWatcher.java:55)
> at org.junit.rules.RunRules.evaluate(RunRules.java:20)
> at org.junit.runners.ParentRunner.runLeaf(ParentRunner.java:325)
> at
> org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:78)
> at
> org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:57)
> at org.junit.runners.ParentRunner$3.run(ParentRunner.java:290)
> at org.junit.runners.ParentRunner$1.schedule(ParentRunner.java:71)
> at org.junit.runners.ParentRunner.runChildren(ParentRunner.java:288)
> at org.junit.runners.ParentRunner.access$000(ParentRunner.java:58)
> at org.junit.runners.ParentRunner$2.evaluate(ParentRunner.java:268)
> at
> org.junit.internal.runners.statements.RunBefores.evaluate(RunBefores.java:26)
> at
> org.junit.internal.runners.statements.RunAfters.evaluate(RunAfters.java:27)
> at org.junit.rules.ExternalResource$1.evaluate(ExternalResource.java:48)
> at org.junit.rules.ExternalResource$1.evaluate(ExternalResource.java:48)
> at org.junit.rules.RunRules.evaluate(RunRules.java:20)
> at org.junit.runners.ParentRunner.run(ParentRunner.java:363)
> at
> org.apache.maven.surefire.junit4.JUnit4Provider.execute(JUnit4Provider.java:365)
> at
> org.apache.maven.surefire.junit4.JUnit4Provider.executeWithRerun(JUnit4Provider.java:273)
> at
> org.apache.maven.surefire.junit4.JUnit4Provider.executeTestSet(JUnit4Provider.java:238)
> at
> org.apache.maven.surefire.junit4.JUnit4Provider.invoke(JUnit4Provider.java:159)
> at
> org.apache.maven.surefire.booter.ForkedBooter.invokeProviderInSameClassLoader(ForkedBooter.java:384)
> at
> org.apache.maven.surefire.booter.ForkedBooter.runSuitesInProcess(ForkedBooter.java:345)
> at
> org.apache.maven.surefire.booter.ForkedBooter.execute(ForkedBooter.java:126)
> at
> org.apache.maven.surefire.booter.ForkedBooter.main(ForkedBooter.java:418)
> {code}
> https://api.travis-ci.com/v3/job/238920411/log.txt
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[jira] [Closed] (FLINK-14235) Kafka010ProducerITCase>KafkaProducerTestBase.testOneToOneAtLeastOnceCustomOperator fails on travis
[
https://issues.apache.org/jira/browse/FLINK-14235?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Piotr Nowojski closed FLINK-14235.
--
Resolution: Fixed
I hope this issue should be fixed by *c31e44e* (merged to the master branch).
Closing the ticket for now, please re-open if you get the same failure message
in the same test on a branch that includes this fix.
Special thanks to [~AHeise] for analysing and finding the root cause.
> Kafka010ProducerITCase>KafkaProducerTestBase.testOneToOneAtLeastOnceCustomOperator
> fails on travis
> --
>
> Key: FLINK-14235
> URL: https://issues.apache.org/jira/browse/FLINK-14235
> Project: Flink
> Issue Type: Bug
> Components: Connectors / Kafka
>Affects Versions: 1.10.0
>Reporter: Piotr Nowojski
>Assignee: Piotr Nowojski
>Priority: Blocker
> Labels: pull-request-available, test-stability
> Fix For: 1.10.0
>
> Time Spent: 20m
> Remaining Estimate: 0h
>
> Fails with following message:
> {noformat}
> Expected to contain all of: <[0]>, but was: <[]>
> {noformat}
> with the following stack trace:
> {noformat}
> Test
> testOneToOneAtLeastOnceCustomOperator(org.apache.flink.streaming.connectors.kafka.Kafka010ProducerITCase)
> failed with:
> java.lang.AssertionError: Expected to contain all of: <[0]>, but was: <[]>
> at org.junit.Assert.fail(Assert.java:88)
> at
> org.apache.flink.streaming.connectors.kafka.KafkaTestBase.assertAtLeastOnceForTopic(KafkaTestBase.java:235)
> at
> org.apache.flink.streaming.connectors.kafka.KafkaProducerTestBase.testOneToOneAtLeastOnce(KafkaProducerTestBase.java:289)
> at
> org.apache.flink.streaming.connectors.kafka.KafkaProducerTestBase.testOneToOneAtLeastOnceCustomOperator(KafkaProducerTestBase.java:214)
> 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.junit.runners.model.FrameworkMethod$1.runReflectiveCall(FrameworkMethod.java:50)
> at
> org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:12)
> at
> org.junit.runners.model.FrameworkMethod.invokeExplosively(FrameworkMethod.java:47)
> at
> org.junit.internal.runners.statements.InvokeMethod.evaluate(InvokeMethod.java:17)
> at org.junit.rules.TestWatcher$1.evaluate(TestWatcher.java:55)
> at org.junit.rules.RunRules.evaluate(RunRules.java:20)
> at org.junit.runners.ParentRunner.runLeaf(ParentRunner.java:325)
> at
> org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:78)
> at
> org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:57)
> at org.junit.runners.ParentRunner$3.run(ParentRunner.java:290)
> at org.junit.runners.ParentRunner$1.schedule(ParentRunner.java:71)
> at org.junit.runners.ParentRunner.runChildren(ParentRunner.java:288)
> at org.junit.runners.ParentRunner.access$000(ParentRunner.java:58)
> at org.junit.runners.ParentRunner$2.evaluate(ParentRunner.java:268)
> at
> org.junit.internal.runners.statements.RunBefores.evaluate(RunBefores.java:26)
> at
> org.junit.internal.runners.statements.RunAfters.evaluate(RunAfters.java:27)
> at org.junit.rules.ExternalResource$1.evaluate(ExternalResource.java:48)
> at org.junit.rules.ExternalResource$1.evaluate(ExternalResource.java:48)
> at org.junit.rules.RunRules.evaluate(RunRules.java:20)
> at org.junit.runners.ParentRunner.run(ParentRunner.java:363)
> at
> org.apache.maven.surefire.junit4.JUnit4Provider.execute(JUnit4Provider.java:365)
> at
> org.apache.maven.surefire.junit4.JUnit4Provider.executeWithRerun(JUnit4Provider.java:273)
> at
> org.apache.maven.surefire.junit4.JUnit4Provider.executeTestSet(JUnit4Provider.java:238)
> at
> org.apache.maven.surefire.junit4.JUnit4Provider.invoke(JUnit4Provider.java:159)
> at
> org.apache.maven.surefire.booter.ForkedBooter.invokeProviderInSameClassLoader(ForkedBooter.java:384)
> at
> org.apache.maven.surefire.booter.ForkedBooter.runSuitesInProcess(ForkedBooter.java:345)
> at
> org.apache.maven.surefire.booter.ForkedBooter.execute(ForkedBooter.java:126)
> at
> org.apache.maven.surefire.booter.ForkedBooter.main(ForkedBooter.java:418)
> {noformat}
> https://api.travis-ci.com/v3/job/239300010/log.txt
> This might be related to https://issues.apache.org/jira/browse/FLINK-14224
> but the failure message is different (here it's data loss, there the
[jira] [Assigned] (FLINK-14235) Kafka010ProducerITCase>KafkaProducerTestBase.testOneToOneAtLeastOnceCustomOperator fails on travis
[
https://issues.apache.org/jira/browse/FLINK-14235?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Piotr Nowojski reassigned FLINK-14235:
--
Assignee: Piotr Nowojski
> Kafka010ProducerITCase>KafkaProducerTestBase.testOneToOneAtLeastOnceCustomOperator
> fails on travis
> --
>
> Key: FLINK-14235
> URL: https://issues.apache.org/jira/browse/FLINK-14235
> Project: Flink
> Issue Type: Bug
> Components: Connectors / Kafka
>Affects Versions: 1.10.0
>Reporter: Piotr Nowojski
>Assignee: Piotr Nowojski
>Priority: Blocker
> Labels: pull-request-available, test-stability
> Fix For: 1.10.0
>
> Time Spent: 20m
> Remaining Estimate: 0h
>
> Fails with following message:
> {noformat}
> Expected to contain all of: <[0]>, but was: <[]>
> {noformat}
> with the following stack trace:
> {noformat}
> Test
> testOneToOneAtLeastOnceCustomOperator(org.apache.flink.streaming.connectors.kafka.Kafka010ProducerITCase)
> failed with:
> java.lang.AssertionError: Expected to contain all of: <[0]>, but was: <[]>
> at org.junit.Assert.fail(Assert.java:88)
> at
> org.apache.flink.streaming.connectors.kafka.KafkaTestBase.assertAtLeastOnceForTopic(KafkaTestBase.java:235)
> at
> org.apache.flink.streaming.connectors.kafka.KafkaProducerTestBase.testOneToOneAtLeastOnce(KafkaProducerTestBase.java:289)
> at
> org.apache.flink.streaming.connectors.kafka.KafkaProducerTestBase.testOneToOneAtLeastOnceCustomOperator(KafkaProducerTestBase.java:214)
> 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.junit.runners.model.FrameworkMethod$1.runReflectiveCall(FrameworkMethod.java:50)
> at
> org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:12)
> at
> org.junit.runners.model.FrameworkMethod.invokeExplosively(FrameworkMethod.java:47)
> at
> org.junit.internal.runners.statements.InvokeMethod.evaluate(InvokeMethod.java:17)
> at org.junit.rules.TestWatcher$1.evaluate(TestWatcher.java:55)
> at org.junit.rules.RunRules.evaluate(RunRules.java:20)
> at org.junit.runners.ParentRunner.runLeaf(ParentRunner.java:325)
> at
> org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:78)
> at
> org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:57)
> at org.junit.runners.ParentRunner$3.run(ParentRunner.java:290)
> at org.junit.runners.ParentRunner$1.schedule(ParentRunner.java:71)
> at org.junit.runners.ParentRunner.runChildren(ParentRunner.java:288)
> at org.junit.runners.ParentRunner.access$000(ParentRunner.java:58)
> at org.junit.runners.ParentRunner$2.evaluate(ParentRunner.java:268)
> at
> org.junit.internal.runners.statements.RunBefores.evaluate(RunBefores.java:26)
> at
> org.junit.internal.runners.statements.RunAfters.evaluate(RunAfters.java:27)
> at org.junit.rules.ExternalResource$1.evaluate(ExternalResource.java:48)
> at org.junit.rules.ExternalResource$1.evaluate(ExternalResource.java:48)
> at org.junit.rules.RunRules.evaluate(RunRules.java:20)
> at org.junit.runners.ParentRunner.run(ParentRunner.java:363)
> at
> org.apache.maven.surefire.junit4.JUnit4Provider.execute(JUnit4Provider.java:365)
> at
> org.apache.maven.surefire.junit4.JUnit4Provider.executeWithRerun(JUnit4Provider.java:273)
> at
> org.apache.maven.surefire.junit4.JUnit4Provider.executeTestSet(JUnit4Provider.java:238)
> at
> org.apache.maven.surefire.junit4.JUnit4Provider.invoke(JUnit4Provider.java:159)
> at
> org.apache.maven.surefire.booter.ForkedBooter.invokeProviderInSameClassLoader(ForkedBooter.java:384)
> at
> org.apache.maven.surefire.booter.ForkedBooter.runSuitesInProcess(ForkedBooter.java:345)
> at
> org.apache.maven.surefire.booter.ForkedBooter.execute(ForkedBooter.java:126)
> at
> org.apache.maven.surefire.booter.ForkedBooter.main(ForkedBooter.java:418)
> {noformat}
> https://api.travis-ci.com/v3/job/239300010/log.txt
> This might be related to https://issues.apache.org/jira/browse/FLINK-14224
> but the failure message is different (here it's data loss, there the job has
> failed)
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[GitHub] [flink] TisonKun commented on a change in pull request #9832: [FLINK-11843] Bind lifespan of Dispatcher to leader session
TisonKun commented on a change in pull request #9832: [FLINK-11843] Bind
lifespan of Dispatcher to leader session
URL: https://github.com/apache/flink/pull/9832#discussion_r337541083
##
File path:
flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/runner/DispatcherLeaderProcessImplTest.java
##
@@ -239,6 +246,109 @@ public void
closeAsync_duringJobRecovery_preventsDispatcherServiceCreation() thr
}
}
+ @Test
+ public void onRemovedJobGraph_cancelsRunningJob() throws Exception {
Review comment:
interesting to try it out :-)
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[jira] [Commented] (FLINK-14224) Kafka010ProducerITCase>KafkaProducerTestBase.testOneToOneAtLeastOnceCustomOperator fails on Travis
[
https://issues.apache.org/jira/browse/FLINK-14224?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16957105#comment-16957105
]
Yun Tang commented on FLINK-14224:
--
Another instance: https://api.travis-ci.com/v3/job/248212043/log.txt
> Kafka010ProducerITCase>KafkaProducerTestBase.testOneToOneAtLeastOnceCustomOperator
> fails on Travis
> --
>
> Key: FLINK-14224
> URL: https://issues.apache.org/jira/browse/FLINK-14224
> Project: Flink
> Issue Type: Bug
> Components: Connectors / Kafka
>Affects Versions: 1.10.0
>Reporter: Till Rohrmann
>Assignee: Alex
>Priority: Blocker
> Labels: test-stability
> Fix For: 1.10.0
>
>
> The
> {{Kafka010ProducerITCase>KafkaProducerTestBase.testOneToOneAtLeastOnceCustomOperator}}
> fails on Travis with
> {code}
> Test
> testOneToOneAtLeastOnceCustomOperator(org.apache.flink.streaming.connectors.kafka.Kafka010ProducerITCase)
> failed with:
> java.lang.AssertionError: Job should fail!
> at org.junit.Assert.fail(Assert.java:88)
> at
> org.apache.flink.streaming.connectors.kafka.KafkaProducerTestBase.testOneToOneAtLeastOnce(KafkaProducerTestBase.java:280)
> at
> org.apache.flink.streaming.connectors.kafka.KafkaProducerTestBase.testOneToOneAtLeastOnceCustomOperator(KafkaProducerTestBase.java:214)
> 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.junit.runners.model.FrameworkMethod$1.runReflectiveCall(FrameworkMethod.java:50)
> at
> org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:12)
> at
> org.junit.runners.model.FrameworkMethod.invokeExplosively(FrameworkMethod.java:47)
> at
> org.junit.internal.runners.statements.InvokeMethod.evaluate(InvokeMethod.java:17)
> at org.junit.rules.TestWatcher$1.evaluate(TestWatcher.java:55)
> at org.junit.rules.RunRules.evaluate(RunRules.java:20)
> at org.junit.runners.ParentRunner.runLeaf(ParentRunner.java:325)
> at
> org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:78)
> at
> org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:57)
> at org.junit.runners.ParentRunner$3.run(ParentRunner.java:290)
> at org.junit.runners.ParentRunner$1.schedule(ParentRunner.java:71)
> at org.junit.runners.ParentRunner.runChildren(ParentRunner.java:288)
> at org.junit.runners.ParentRunner.access$000(ParentRunner.java:58)
> at org.junit.runners.ParentRunner$2.evaluate(ParentRunner.java:268)
> at
> org.junit.internal.runners.statements.RunBefores.evaluate(RunBefores.java:26)
> at
> org.junit.internal.runners.statements.RunAfters.evaluate(RunAfters.java:27)
> at org.junit.rules.ExternalResource$1.evaluate(ExternalResource.java:48)
> at org.junit.rules.ExternalResource$1.evaluate(ExternalResource.java:48)
> at org.junit.rules.RunRules.evaluate(RunRules.java:20)
> at org.junit.runners.ParentRunner.run(ParentRunner.java:363)
> at
> org.apache.maven.surefire.junit4.JUnit4Provider.execute(JUnit4Provider.java:365)
> at
> org.apache.maven.surefire.junit4.JUnit4Provider.executeWithRerun(JUnit4Provider.java:273)
> at
> org.apache.maven.surefire.junit4.JUnit4Provider.executeTestSet(JUnit4Provider.java:238)
> at
> org.apache.maven.surefire.junit4.JUnit4Provider.invoke(JUnit4Provider.java:159)
> at
> org.apache.maven.surefire.booter.ForkedBooter.invokeProviderInSameClassLoader(ForkedBooter.java:384)
> at
> org.apache.maven.surefire.booter.ForkedBooter.runSuitesInProcess(ForkedBooter.java:345)
> at
> org.apache.maven.surefire.booter.ForkedBooter.execute(ForkedBooter.java:126)
> at
> org.apache.maven.surefire.booter.ForkedBooter.main(ForkedBooter.java:418)
> {code}
> https://api.travis-ci.com/v3/job/238920411/log.txt
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[GitHub] [flink] pnowojski commented on issue #9959: [FLINK-14235][kafka, tests] Change source in at-least-once test from finite to infinite
pnowojski commented on issue #9959: [FLINK-14235][kafka,tests] Change source in at-least-once test from finite to infinite URL: https://github.com/apache/flink/pull/9959#issuecomment-544982034 Thanks for the quick review @AHeise. Merging. I will close all of the three linked tickets and let's hope nobody will re-open them. This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] pnowojski merged pull request #9959: [FLINK-14235][kafka, tests] Change source in at-least-once test from finite to infinite
pnowojski merged pull request #9959: [FLINK-14235][kafka,tests] Change source in at-least-once test from finite to infinite URL: https://github.com/apache/flink/pull/9959 This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #9966: [FLINK-14053] [blink-planner] DenseRankAggFunction.accumulateExpressions. it should be thinki…
flinkbot edited a comment on issue #9966: [FLINK-14053] [blink-planner] DenseRankAggFunction.accumulateExpressions. it should be thinki… URL: https://github.com/apache/flink/pull/9966#issuecomment-544859291 ## CI report: * 6d09ab06b8bffd326034d8872cd85810bc09b164 : FAILURE [Build](https://travis-ci.com/flink-ci/flink/builds/132954193) * ec066d37d9bf1a5d120231dbbd73b4b196333691 : FAILURE [Build](https://travis-ci.com/flink-ci/flink/builds/132989763) Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] flinkbot commented on issue #9969: [FLINK-14022][table-planner][table-planner-blink] Add validation check for places where Python ScalarFunction cannot be used
flinkbot commented on issue #9969: [FLINK-14022][table-planner][table-planner-blink] Add validation check for places where Python ScalarFunction cannot be used URL: https://github.com/apache/flink/pull/9969#issuecomment-54497 ## CI report: * f482bed6b11fcc6790ee1373be618388c13e1c05 : UNKNOWN Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] flinkbot commented on issue #9970: [FLINK-14395] Refactor ES 7 connectors to make them keep consistency with es 6 connectors
flinkbot commented on issue #9970: [FLINK-14395] Refactor ES 7 connectors to make them keep consistency with es 6 connectors URL: https://github.com/apache/flink/pull/9970#issuecomment-544980056 ## CI report: * a2b9906d887c4feba3f1777957af4d3027116cb0 : UNKNOWN Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #9950: [FLINK-14464][runtime] Introduce the AbstractUserClassPathJobGraphRetriever
flinkbot edited a comment on issue #9950: [FLINK-14464][runtime] Introduce the AbstractUserClassPathJobGraphRetriever URL: https://github.com/apache/flink/pull/9950#issuecomment-544414033 ## CI report: * decba8623c001e8bbe5dc797de2bc421b2e216ef : SUCCESS [Build](https://travis-ci.com/flink-ci/flink/builds/132777340) * 6c7ffb9f9049016ce874b9790cd51ed337cfec00 : SUCCESS [Build](https://travis-ci.com/flink-ci/flink/builds/132974469) * f381f5765ae1a654d545aff4c679eaf9e4812ddc : SUCCESS [Build](https://travis-ci.com/flink-ci/flink/builds/132985132) Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #9928: [FLINK-12122] Add support for spreading slots out across all TaskExecutors
flinkbot edited a comment on issue #9928: [FLINK-12122] Add support for spreading slots out across all TaskExecutors URL: https://github.com/apache/flink/pull/9928#issuecomment-543204040 ## CI report: * 1f0eed38301790afdbbd981d2a9853fa240e10df : FAILURE [Build](https://travis-ci.com/flink-ci/flink/builds/132351947) * 32fe55877e2118f81d8e31d4f9b0449e99ecb185 : FAILURE [Build](https://travis-ci.com/flink-ci/flink/builds/132357162) * 7a70e355da5c4ed900f766772eadb7628a46f140 : FAILURE [Build](https://travis-ci.com/flink-ci/flink/builds/132386132) * 02ba0e625ac179168f4db9a50eed61deff1ca197 : SUCCESS [Build](https://travis-ci.com/flink-ci/flink/builds/132989732) Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #9881: [FLINK-14377] Parse Executor-relevant ProgramOptions to ConfigOptions
flinkbot edited a comment on issue #9881: [FLINK-14377] Parse Executor-relevant ProgramOptions to ConfigOptions URL: https://github.com/apache/flink/pull/9881#issuecomment-541072480 ## CI report: * 03fb6e64db35f290c9a1d5552aa34dfe5ccf9cd8 : SUCCESS [Build](https://travis-ci.com/flink-ci/flink/builds/131535420) * eb165ca8d15ebe68afde907b9f52de962e293a2f : SUCCESS [Build](https://travis-ci.com/flink-ci/flink/builds/131751271) * 0d3f28ea5a5aa583f5f169c365a2b4ab4ffa7cab : SUCCESS [Build](https://travis-ci.com/flink-ci/flink/builds/131793914) * bc1f0676e81681c1b0c42a7cf0cbc03d0ec47051 : SUCCESS [Build](https://travis-ci.com/flink-ci/flink/builds/131965643) * c4b9fee1eaefd15cd0e9a7c7acaae67347b80041 : PENDING [Build](https://travis-ci.com/flink-ci/flink/builds/133000714) Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #9300: [FLINK-13513][ml] Add the FlatMapper and related classes for later al…
flinkbot edited a comment on issue #9300: [FLINK-13513][ml] Add the FlatMapper and related classes for later al… URL: https://github.com/apache/flink/pull/9300#issuecomment-516849331 ## CI report: * 9b5b5ff5df053498e491d43a04f44d5ba452579c : SUCCESS [Build](https://travis-ci.com/flink-ci/flink/builds/121415160) * eb6cf333f4331ed20d7e22a056cbd3c9b61f31f8 : FAILURE [Build](https://travis-ci.com/flink-ci/flink/builds/124204150) * 6440190df24a227f96e2b917acecccee04ab981b : FAILURE [Build](https://travis-ci.com/flink-ci/flink/builds/128458494) * de471a0446423d026ae75476f1e4892126668a40 : SUCCESS [Build](https://travis-ci.com/flink-ci/flink/builds/128501969) * c377de8dd42673a3bd229e9e05e28cbd1c1863d7 : SUCCESS [Build](https://travis-ci.com/flink-ci/flink/builds/129638886) * 5dde9734d42c1f427ad574a4d5fdbd4bd4f35fdd : SUCCESS [Build](https://travis-ci.com/flink-ci/flink/builds/132989695) Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #9881: [FLINK-14377] Parse Executor-relevant ProgramOptions to ConfigOptions
flinkbot edited a comment on issue #9881: [FLINK-14377] Parse Executor-relevant ProgramOptions to ConfigOptions URL: https://github.com/apache/flink/pull/9881#issuecomment-541072480 ## CI report: * 03fb6e64db35f290c9a1d5552aa34dfe5ccf9cd8 : SUCCESS [Build](https://travis-ci.com/flink-ci/flink/builds/131535420) * eb165ca8d15ebe68afde907b9f52de962e293a2f : SUCCESS [Build](https://travis-ci.com/flink-ci/flink/builds/131751271) * 0d3f28ea5a5aa583f5f169c365a2b4ab4ffa7cab : SUCCESS [Build](https://travis-ci.com/flink-ci/flink/builds/131793914) * bc1f0676e81681c1b0c42a7cf0cbc03d0ec47051 : SUCCESS [Build](https://travis-ci.com/flink-ci/flink/builds/131965643) * c4b9fee1eaefd15cd0e9a7c7acaae67347b80041 : UNKNOWN Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #9747: [FLINK-13985] Use unsafe memory for managed memory
flinkbot edited a comment on issue #9747: [FLINK-13985] Use unsafe memory for managed memory URL: https://github.com/apache/flink/pull/9747#issuecomment-534087338 ## CI report: * 1967191401dfac1a56eb3d82a0e97bd080e4c588 : SUCCESS [Build](https://travis-ci.com/flink-ci/flink/builds/128743926) * 79f81fd5897e2f42a4918178bda23cbcc9174514 : SUCCESS [Build](https://travis-ci.com/flink-ci/flink/builds/131974566) * 92e38d91c935017fab82f6a53a2b4bcbe528376f : SUCCESS [Build](https://travis-ci.com/flink-ci/flink/builds/132123636) * addcfd190caf98fc8bbd64cb65b0938b4069 : SUCCESS [Build](https://travis-ci.com/flink-ci/flink/builds/132989666) * 866835a644e4027c3c334561f1754c37fee0b12d : PENDING [Build](https://travis-ci.com/flink-ci/flink/builds/132998270) Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[jira] [Commented] (FLINK-13025) Elasticsearch 7.x support
[ https://issues.apache.org/jira/browse/FLINK-13025?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16957054#comment-16957054 ] vinoyang commented on FLINK-13025: -- [~lilyevsky] It seems you can get the artifact in the maven central repo after releasing 1.10. Additionally, we still have some little issue need to fix. Please see FLINK-14395, I have opened a PR, will fix it soon. > Elasticsearch 7.x support > - > > Key: FLINK-13025 > URL: https://issues.apache.org/jira/browse/FLINK-13025 > Project: Flink > Issue Type: New Feature > Components: Connectors / ElasticSearch >Affects Versions: 1.8.0 >Reporter: Keegan Standifer >Assignee: vinoyang >Priority: Major > Labels: pull-request-available > Fix For: 1.10.0 > > Attachments: flink-connector-elasticsearch7_2.12-1.10-SNAPSHOT.jar > > Time Spent: 20m > Remaining Estimate: 0h > > Elasticsearch 7.0.0 was released in April of 2019: > [https://www.elastic.co/blog/elasticsearch-7-0-0-released] > The latest elasticsearch connector is > [flink-connector-elasticsearch6|https://github.com/apache/flink/tree/master/flink-connectors/flink-connector-elasticsearch6] -- This message was sent by Atlassian Jira (v8.3.4#803005)
[GitHub] [flink] tillrohrmann commented on a change in pull request #9832: [FLINK-11843] Bind lifespan of Dispatcher to leader session
tillrohrmann commented on a change in pull request #9832: [FLINK-11843] Bind
lifespan of Dispatcher to leader session
URL: https://github.com/apache/flink/pull/9832#discussion_r337511483
##
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/runner/AbstractDispatcherLeaderProcess.java
##
@@ -0,0 +1,255 @@
+/*
+ * 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.dispatcher.runner;
+
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.runtime.clusterframework.ApplicationStatus;
+import org.apache.flink.runtime.concurrent.FutureUtils;
+import org.apache.flink.runtime.dispatcher.DispatcherGateway;
+import org.apache.flink.runtime.dispatcher.DispatcherId;
+import org.apache.flink.runtime.jobgraph.JobGraph;
+import org.apache.flink.runtime.jobmanager.JobGraphWriter;
+import org.apache.flink.runtime.rpc.FatalErrorHandler;
+import org.apache.flink.runtime.webmonitor.RestfulGateway;
+import org.apache.flink.util.AutoCloseableAsync;
+import org.apache.flink.util.Preconditions;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import java.util.Collection;
+import java.util.Optional;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.function.Predicate;
+import java.util.function.Supplier;
+
+@SuppressWarnings("checkstyle:EmptyLineSeparator")
Review comment:
Will remove this line when merging.
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[jira] [Commented] (FLINK-13025) Elasticsearch 7.x support
[ https://issues.apache.org/jira/browse/FLINK-13025?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16957049#comment-16957049 ] Leonid Ilyevsky commented on FLINK-13025: - Thanks everybody. When can we see this artifact in central repo? > Elasticsearch 7.x support > - > > Key: FLINK-13025 > URL: https://issues.apache.org/jira/browse/FLINK-13025 > Project: Flink > Issue Type: New Feature > Components: Connectors / ElasticSearch >Affects Versions: 1.8.0 >Reporter: Keegan Standifer >Assignee: vinoyang >Priority: Major > Labels: pull-request-available > Fix For: 1.10.0 > > Attachments: flink-connector-elasticsearch7_2.12-1.10-SNAPSHOT.jar > > Time Spent: 20m > Remaining Estimate: 0h > > Elasticsearch 7.0.0 was released in April of 2019: > [https://www.elastic.co/blog/elasticsearch-7-0-0-released] > The latest elasticsearch connector is > [flink-connector-elasticsearch6|https://github.com/apache/flink/tree/master/flink-connectors/flink-connector-elasticsearch6] -- This message was sent by Atlassian Jira (v8.3.4#803005)
[GitHub] [flink] flinkbot edited a comment on issue #9747: [FLINK-13985] Use unsafe memory for managed memory
flinkbot edited a comment on issue #9747: [FLINK-13985] Use unsafe memory for managed memory URL: https://github.com/apache/flink/pull/9747#issuecomment-534087338 ## CI report: * 1967191401dfac1a56eb3d82a0e97bd080e4c588 : SUCCESS [Build](https://travis-ci.com/flink-ci/flink/builds/128743926) * 79f81fd5897e2f42a4918178bda23cbcc9174514 : SUCCESS [Build](https://travis-ci.com/flink-ci/flink/builds/131974566) * 92e38d91c935017fab82f6a53a2b4bcbe528376f : SUCCESS [Build](https://travis-ci.com/flink-ci/flink/builds/132123636) * addcfd190caf98fc8bbd64cb65b0938b4069 : PENDING [Build](https://travis-ci.com/flink-ci/flink/builds/132989666) * 866835a644e4027c3c334561f1754c37fee0b12d : UNKNOWN Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] flinkbot commented on issue #9970: [FLINK-14395] Refactor ES 7 connectors to make them keep consistency with es 6 connectors
flinkbot commented on issue #9970: [FLINK-14395] Refactor ES 7 connectors to make them keep consistency with es 6 connectors URL: https://github.com/apache/flink/pull/9970#issuecomment-544951741 Thanks a lot for your contribution to the Apache Flink project. I'm the @flinkbot. I help the community to review your pull request. We will use this comment to track the progress of the review. ## Automated Checks Last check on commit 3f08a10766658606e458773682c3be93d108b071 (Tue Oct 22 13:04:26 UTC 2019) **Warnings:** * No documentation files were touched! Remember to keep the Flink docs up to date! * **This pull request references an unassigned [Jira ticket](https://issues.apache.org/jira/browse/FLINK-14395).** According to the [code contribution guide](https://flink.apache.org/contributing/contribute-code.html), tickets need to be assigned before starting with the implementation work. Mention the bot in a comment to re-run the automated checks. ## Review Progress * ❓ 1. The [description] looks good. * ❓ 2. There is [consensus] that the contribution should go into to Flink. * ❓ 3. Needs [attention] from. * ❓ 4. The change fits into the overall [architecture]. * ❓ 5. Overall code [quality] is good. Please see the [Pull Request Review Guide](https://flink.apache.org/contributing/reviewing-prs.html) for a full explanation of the review process. The Bot is tracking the review progress through labels. Labels are applied according to the order of the review items. For consensus, approval by a Flink committer of PMC member is required Bot commands The @flinkbot bot supports the following commands: - `@flinkbot approve description` to approve one or more aspects (aspects: `description`, `consensus`, `architecture` and `quality`) - `@flinkbot approve all` to approve all aspects - `@flinkbot approve-until architecture` to approve everything until `architecture` - `@flinkbot attention @username1 [@username2 ..]` to require somebody's attention - `@flinkbot disapprove architecture` to remove an approval you gave earlier This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #9928: [FLINK-12122] Add support for spreading slots out across all TaskExecutors
flinkbot edited a comment on issue #9928: [FLINK-12122] Add support for spreading slots out across all TaskExecutors URL: https://github.com/apache/flink/pull/9928#issuecomment-543204040 ## CI report: * 1f0eed38301790afdbbd981d2a9853fa240e10df : FAILURE [Build](https://travis-ci.com/flink-ci/flink/builds/132351947) * 32fe55877e2118f81d8e31d4f9b0449e99ecb185 : FAILURE [Build](https://travis-ci.com/flink-ci/flink/builds/132357162) * 7a70e355da5c4ed900f766772eadb7628a46f140 : FAILURE [Build](https://travis-ci.com/flink-ci/flink/builds/132386132) * 02ba0e625ac179168f4db9a50eed61deff1ca197 : PENDING [Build](https://travis-ci.com/flink-ci/flink/builds/132989732) Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] yanghua opened a new pull request #9970: [FLINK-14395] Refactor ES 7 connectors to make them keep consistency with es 6 connectors
yanghua opened a new pull request #9970: [FLINK-14395] Refactor ES 7 connectors to make them keep consistency with es 6 connectors URL: https://github.com/apache/flink/pull/9970 ## What is the purpose of the change *This pull request refactors ES 7 connectors to make them keep consistency with es 6 connectors* ## Brief change log - *Moved table-specific code and test from sql connector to general connector* ## Verifying this change This change is a trivial rework / code cleanup without any test coverage. ## Does this pull request potentially affect one of the following parts: - Dependencies (does it add or upgrade a dependency): (yes / **no**) - The public API, i.e., is any changed class annotated with `@Public(Evolving)`: (yes / **no**) - The serializers: (yes / **no** / don't know) - The runtime per-record code paths (performance sensitive): (yes / **no** / don't know) - Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Yarn/Mesos, ZooKeeper: (yes / **no** / don't know) - The S3 file system connector: (yes / **no** / don't know) ## Documentation - Does this pull request introduce a new feature? (yes / **no**) - If yes, how is the feature documented? (not applicable / docs / JavaDocs / **not documented**) This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] flinkbot commented on issue #9969: [FLINK-14022][table-planner][table-planner-blink] Add validation check for places where Python ScalarFunction cannot be used
flinkbot commented on issue #9969: [FLINK-14022][table-planner][table-planner-blink] Add validation check for places where Python ScalarFunction cannot be used URL: https://github.com/apache/flink/pull/9969#issuecomment-544948951 Thanks a lot for your contribution to the Apache Flink project. I'm the @flinkbot. I help the community to review your pull request. We will use this comment to track the progress of the review. ## Automated Checks Last check on commit f482bed6b11fcc6790ee1373be618388c13e1c05 (Tue Oct 22 12:57:27 UTC 2019) **Warnings:** * No documentation files were touched! Remember to keep the Flink docs up to date! Mention the bot in a comment to re-run the automated checks. ## Review Progress * ❓ 1. The [description] looks good. * ❓ 2. There is [consensus] that the contribution should go into to Flink. * ❓ 3. Needs [attention] from. * ❓ 4. The change fits into the overall [architecture]. * ❓ 5. Overall code [quality] is good. Please see the [Pull Request Review Guide](https://flink.apache.org/contributing/reviewing-prs.html) for a full explanation of the review process. The Bot is tracking the review progress through labels. Labels are applied according to the order of the review items. For consensus, approval by a Flink committer of PMC member is required Bot commands The @flinkbot bot supports the following commands: - `@flinkbot approve description` to approve one or more aspects (aspects: `description`, `consensus`, `architecture` and `quality`) - `@flinkbot approve all` to approve all aspects - `@flinkbot approve-until architecture` to approve everything until `architecture` - `@flinkbot attention @username1 [@username2 ..]` to require somebody's attention - `@flinkbot disapprove architecture` to remove an approval you gave earlier This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #9300: [FLINK-13513][ml] Add the FlatMapper and related classes for later al…
flinkbot edited a comment on issue #9300: [FLINK-13513][ml] Add the FlatMapper and related classes for later al… URL: https://github.com/apache/flink/pull/9300#issuecomment-516849331 ## CI report: * 9b5b5ff5df053498e491d43a04f44d5ba452579c : SUCCESS [Build](https://travis-ci.com/flink-ci/flink/builds/121415160) * eb6cf333f4331ed20d7e22a056cbd3c9b61f31f8 : FAILURE [Build](https://travis-ci.com/flink-ci/flink/builds/124204150) * 6440190df24a227f96e2b917acecccee04ab981b : FAILURE [Build](https://travis-ci.com/flink-ci/flink/builds/128458494) * de471a0446423d026ae75476f1e4892126668a40 : SUCCESS [Build](https://travis-ci.com/flink-ci/flink/builds/128501969) * c377de8dd42673a3bd229e9e05e28cbd1c1863d7 : SUCCESS [Build](https://travis-ci.com/flink-ci/flink/builds/129638886) * 5dde9734d42c1f427ad574a4d5fdbd4bd4f35fdd : PENDING [Build](https://travis-ci.com/flink-ci/flink/builds/132989695) Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[jira] [Updated] (FLINK-14022) Add validation check for places where Python ScalarFunction cannot be used
[ https://issues.apache.org/jira/browse/FLINK-14022?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] ASF GitHub Bot updated FLINK-14022: --- Labels: pull-request-available (was: ) > Add validation check for places where Python ScalarFunction cannot be used > -- > > Key: FLINK-14022 > URL: https://issues.apache.org/jira/browse/FLINK-14022 > Project: Flink > Issue Type: Sub-task > Components: API / Python >Reporter: Dian Fu >Assignee: Huang Xingbo >Priority: Major > Labels: pull-request-available > Fix For: 1.10.0 > > > Currently, there are places where Python ScalarFunction could not be used, > for example: > # Python UDF could not be used in MatchRecognize > # Python UDFs could not be used in Join condition which take the columns > from both the left table and the right table as inputs > We should add validation check for places where it’s not supported. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[GitHub] [flink] HuangXingBo opened a new pull request #9969: [FLINK-14022][table-planner][table-planner-blink] Add validation check for places where Python ScalarFunction cannot be used
HuangXingBo opened a new pull request #9969: [FLINK-14022][table-planner][table-planner-blink] Add validation check for places where Python ScalarFunction cannot be used URL: https://github.com/apache/flink/pull/9969 ## What is the purpose of the change *Currently, there are places where Python ScalarFunction could not be used, for example:* - *Python UDF could not be used in MatchRecognize* - *Python UDFs could not be used in Join condition which take the columns from both the left table and the right table as inputs* We should add validation check for places where it’s not supported. ## Brief change log This change added tests and can be verified as follows: * add join and matchRecognize tests* ## Does this pull request potentially affect one of the following parts: - Dependencies (does it add or upgrade a dependency): (no) - The public API, i.e., is any changed class annotated with `@Public(Evolving)`: (no) - The serializers: (no) - The runtime per-record code paths (performance sensitive): (no) - Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Yarn/Mesos, ZooKeeper: (no) - The S3 file system connector: (no) ## Documentation - Does this pull request introduce a new feature? (no) - If yes, how is the feature documented? (not applicable) This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[jira] [Resolved] (FLINK-14447) Network metrics doc table render confusion
[ https://issues.apache.org/jira/browse/FLINK-14447?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Andrey Zagrebin resolved FLINK-14447. - Release Note: Merged into master by 56126bd43f6539dea39419e8f81779419ccea1c4 Merged into release-1.9 by 644810014ecf7487137c06c4a4583bede855ac94 Resolution: Fixed > Network metrics doc table render confusion > -- > > Key: FLINK-14447 > URL: https://issues.apache.org/jira/browse/FLINK-14447 > Project: Flink > Issue Type: Bug > Components: Documentation, Runtime / Metrics >Affects Versions: 1.9.0, 1.10.0 >Reporter: vinoyang >Assignee: vinoyang >Priority: Major > Labels: pull-request-available > Fix For: 1.10.0, 1.9.2 > > Time Spent: 20m > Remaining Estimate: 0h > > More detail: > https://ci.apache.org/projects/flink/flink-docs-master/monitoring/metrics.html#network-deprecated-use-default-shuffle-service-metrics -- This message was sent by Atlassian Jira (v8.3.4#803005)
[GitHub] [flink] flinkbot edited a comment on issue #9747: [FLINK-13985] Use unsafe memory for managed memory
flinkbot edited a comment on issue #9747: [FLINK-13985] Use unsafe memory for managed memory URL: https://github.com/apache/flink/pull/9747#issuecomment-534087338 ## CI report: * 1967191401dfac1a56eb3d82a0e97bd080e4c588 : SUCCESS [Build](https://travis-ci.com/flink-ci/flink/builds/128743926) * 79f81fd5897e2f42a4918178bda23cbcc9174514 : SUCCESS [Build](https://travis-ci.com/flink-ci/flink/builds/131974566) * 92e38d91c935017fab82f6a53a2b4bcbe528376f : SUCCESS [Build](https://travis-ci.com/flink-ci/flink/builds/132123636) * addcfd190caf98fc8bbd64cb65b0938b4069 : PENDING [Build](https://travis-ci.com/flink-ci/flink/builds/132989666) Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #9966: [FLINK-14053] [blink-planner] DenseRankAggFunction.accumulateExpressions. it should be thinki…
flinkbot edited a comment on issue #9966: [FLINK-14053] [blink-planner] DenseRankAggFunction.accumulateExpressions. it should be thinki… URL: https://github.com/apache/flink/pull/9966#issuecomment-544859291 ## CI report: * 6d09ab06b8bffd326034d8872cd85810bc09b164 : FAILURE [Build](https://travis-ci.com/flink-ci/flink/builds/132954193) * ec066d37d9bf1a5d120231dbbd73b4b196333691 : UNKNOWN Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #9950: [FLINK-14464][runtime] Introduce the AbstractUserClassPathJobGraphRetriever
flinkbot edited a comment on issue #9950: [FLINK-14464][runtime] Introduce the AbstractUserClassPathJobGraphRetriever URL: https://github.com/apache/flink/pull/9950#issuecomment-544414033 ## CI report: * decba8623c001e8bbe5dc797de2bc421b2e216ef : SUCCESS [Build](https://travis-ci.com/flink-ci/flink/builds/132777340) * 6c7ffb9f9049016ce874b9790cd51ed337cfec00 : SUCCESS [Build](https://travis-ci.com/flink-ci/flink/builds/132974469) * f381f5765ae1a654d545aff4c679eaf9e4812ddc : PENDING [Build](https://travis-ci.com/flink-ci/flink/builds/132985132) Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #9928: [FLINK-12122] Add support for spreading slots out across all TaskExecutors
flinkbot edited a comment on issue #9928: [FLINK-12122] Add support for spreading slots out across all TaskExecutors URL: https://github.com/apache/flink/pull/9928#issuecomment-543204040 ## CI report: * 1f0eed38301790afdbbd981d2a9853fa240e10df : FAILURE [Build](https://travis-ci.com/flink-ci/flink/builds/132351947) * 32fe55877e2118f81d8e31d4f9b0449e99ecb185 : FAILURE [Build](https://travis-ci.com/flink-ci/flink/builds/132357162) * 7a70e355da5c4ed900f766772eadb7628a46f140 : FAILURE [Build](https://travis-ci.com/flink-ci/flink/builds/132386132) * 02ba0e625ac179168f4db9a50eed61deff1ca197 : UNKNOWN Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #9300: [FLINK-13513][ml] Add the FlatMapper and related classes for later al…
flinkbot edited a comment on issue #9300: [FLINK-13513][ml] Add the FlatMapper and related classes for later al… URL: https://github.com/apache/flink/pull/9300#issuecomment-516849331 ## CI report: * 9b5b5ff5df053498e491d43a04f44d5ba452579c : SUCCESS [Build](https://travis-ci.com/flink-ci/flink/builds/121415160) * eb6cf333f4331ed20d7e22a056cbd3c9b61f31f8 : FAILURE [Build](https://travis-ci.com/flink-ci/flink/builds/124204150) * 6440190df24a227f96e2b917acecccee04ab981b : FAILURE [Build](https://travis-ci.com/flink-ci/flink/builds/128458494) * de471a0446423d026ae75476f1e4892126668a40 : SUCCESS [Build](https://travis-ci.com/flink-ci/flink/builds/128501969) * c377de8dd42673a3bd229e9e05e28cbd1c1863d7 : SUCCESS [Build](https://travis-ci.com/flink-ci/flink/builds/129638886) * 5dde9734d42c1f427ad574a4d5fdbd4bd4f35fdd : UNKNOWN Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[jira] [Updated] (FLINK-14447) Network metrics doc table render confusion
[ https://issues.apache.org/jira/browse/FLINK-14447?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Andrey Zagrebin updated FLINK-14447: Fix Version/s: 1.9.2 1.10.0 > Network metrics doc table render confusion > -- > > Key: FLINK-14447 > URL: https://issues.apache.org/jira/browse/FLINK-14447 > Project: Flink > Issue Type: Bug > Components: Documentation, Runtime / Metrics >Affects Versions: 1.9.0, 1.10.0 >Reporter: vinoyang >Assignee: vinoyang >Priority: Major > Labels: pull-request-available > Fix For: 1.10.0, 1.9.2 > > Time Spent: 20m > Remaining Estimate: 0h > > More detail: > https://ci.apache.org/projects/flink/flink-docs-master/monitoring/metrics.html#network-deprecated-use-default-shuffle-service-metrics -- This message was sent by Atlassian Jira (v8.3.4#803005)
[GitHub] [flink] flinkbot edited a comment on issue #9747: [FLINK-13985] Use unsafe memory for managed memory
flinkbot edited a comment on issue #9747: [FLINK-13985] Use unsafe memory for managed memory URL: https://github.com/apache/flink/pull/9747#issuecomment-534087338 ## CI report: * 1967191401dfac1a56eb3d82a0e97bd080e4c588 : SUCCESS [Build](https://travis-ci.com/flink-ci/flink/builds/128743926) * 79f81fd5897e2f42a4918178bda23cbcc9174514 : SUCCESS [Build](https://travis-ci.com/flink-ci/flink/builds/131974566) * 92e38d91c935017fab82f6a53a2b4bcbe528376f : SUCCESS [Build](https://travis-ci.com/flink-ci/flink/builds/132123636) * addcfd190caf98fc8bbd64cb65b0938b4069 : UNKNOWN Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] azagrebin merged pull request #9951: [FLINK-14447] Network metrics doc table render confusion
azagrebin merged pull request #9951: [FLINK-14447] Network metrics doc table render confusion URL: https://github.com/apache/flink/pull/9951 This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] xuyang1706 commented on issue #9300: [FLINK-13513][ml] Add the FlatMapper and related classes for later al…
xuyang1706 commented on issue #9300: [FLINK-13513][ml] Add the FlatMapper and related classes for later al… URL: https://github.com/apache/flink/pull/9300#issuecomment-544931164 > @xuyang1706 Thanks for the patch. The structure and semantic in the patch seems a little confusing to me. I created a PR against your PR with my suggestions. The PR might be a little messy at this point. But should be clean once you rebase the `flatmapper` branch on master. > > Please let me know what do you think. Thanks. Thanks for your comments and discussion, @becketqin. We refactored the codes, please look it if you have time. Thanks, -Xu This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #9950: [FLINK-14464][runtime] Introduce the AbstractUserClassPathJobGraphRetriever
flinkbot edited a comment on issue #9950: [FLINK-14464][runtime] Introduce the AbstractUserClassPathJobGraphRetriever URL: https://github.com/apache/flink/pull/9950#issuecomment-544414033 ## CI report: * decba8623c001e8bbe5dc797de2bc421b2e216ef : SUCCESS [Build](https://travis-ci.com/flink-ci/flink/builds/132777340) * 6c7ffb9f9049016ce874b9790cd51ed337cfec00 : SUCCESS [Build](https://travis-ci.com/flink-ci/flink/builds/132974469) * f381f5765ae1a654d545aff4c679eaf9e4812ddc : UNKNOWN Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] xuyang1706 commented on a change in pull request #9300: [FLINK-13513][ml] Add the FlatMapper and related classes for later al…
xuyang1706 commented on a change in pull request #9300: [FLINK-13513][ml] Add
the FlatMapper and related classes for later al…
URL: https://github.com/apache/flink/pull/9300#discussion_r337475511
##
File path:
flink-ml-parent/flink-ml-lib/src/main/java/org/apache/flink/ml/common/mapper/FlatMapper.java
##
@@ -0,0 +1,108 @@
+/*
+ * 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.ml.common.mapper;
+
+import org.apache.flink.ml.api.misc.param.Params;
+import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.Collector;
+
+import java.io.Serializable;
+
+/**
+ * Abstract class for flatMappers.
+ * FlatMapper maps a row to zero, one or multiple rows.
+ */
+public abstract class FlatMapper implements Serializable {
+
+ /**
+* schema of the input.
+*/
+ protected TableSchema dataSchema;
+
+ /**
+* params used for FlatMapper.
+* User can set the params before that the FlatMapper is executed.
+*/
+ protected Params params;
+
+ public FlatMapper(TableSchema dataSchema, Params params) {
+ this.dataSchema = dataSchema;
+ this.params = (null == params) ? new Params() : params.clone();
+ }
+
+ /**
+* The core method of the FlatMapper.
+* Takes a row from the input and maps it to multiple rows.
+*
+* @param rowThe input row.
+* @param output The collector for returning the result values.
+* @throws Exception This method may throw exceptions. Throwing an
exception will cause the operation
+* to fail.
+*/
+ public abstract void flatMap(Row row, Collector output) throws
Exception;
+
+ /**
+* Get the table schema(includes column names and types) of the
calculation result.
+*
+* @return the table schema of output Row type data
+*/
+ public abstract TableSchema getOutputSchema();
+
+ /**
+* Generate new instance of given FlatMapper class.
+*
+* @param flatMapperClassName Name of the FlatMapper class
+* @param dataSchema The schema of the input data represented
in Table format.
+* @param params The parameters for the instance
construction.
+* @return new instance of given FlatMapper class
+* @throws Exception if flatMapperClass is not the class of {@link
FlatMapper}
+*/
+ public static FlatMapper of(
+ String flatMapperClassName,
+ TableSchema dataSchema,
+ Params params) throws Exception {
+
+ return of(Class.forName(flatMapperClassName), dataSchema,
params);
+ }
+
+ /**
+* Generate new instance of given FlatMapper class.
+*
+* @param flatMapperClass FlatMapper class of the new instance
+* @param dataSchema The schema of the input data represented in
Table format.
+* @param params the parameters for the instance construction.
+* @return new instance of given FlatMapper class
+* @throws Exception if flatMapperClass is not the class of {@link
FlatMapper}
+*/
+ public static FlatMapper of(
+ Class flatMapperClass,
+ TableSchema dataSchema,
+ Params params) throws Exception {
+
+ if (FlatMapper.class.isAssignableFrom(flatMapperClass)) {
+ return (FlatMapper)
flatMapperClass.getConstructor(TableSchema.class, Params.class)
Review comment:
Thanks, changed.
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[jira] [Closed] (FLINK-13984) Separate on-heap and off-heap managed memory pools
[
https://issues.apache.org/jira/browse/FLINK-13984?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Andrey Zagrebin closed FLINK-13984.
---
Release Note: The managed memory segment preallocation has been completely
removed (Configuration option 'taskmanager.memory.preallocate'). It should not
break the existing setups but setting this option will have no effect which can
lead to a slight performance degradation as the segments will be allocated only
in the ad-hoc manner during the execution.
Resolution: Done
Merged into master by 76349cf80ff5170835de9315420dae082425a74a
> Separate on-heap and off-heap managed memory pools
> --
>
> Key: FLINK-13984
> URL: https://issues.apache.org/jira/browse/FLINK-13984
> Project: Flink
> Issue Type: Sub-task
> Components: Runtime / Task
>Reporter: Xintong Song
>Assignee: Andrey Zagrebin
>Priority: Major
> Labels: pull-request-available
> Fix For: 1.10.0
>
> Time Spent: 10m
> Remaining Estimate: 0h
>
> * Update {{MemoryManager}} to have two separated pools.
> * Extend {{MemoryManager}} interfaces to specify which pool to allocate
> memory from.
> Implement this step in common code paths for the legacy / new mode. For the
> legacy mode, depending to the configured memory type, we can set one of the
> two pools to the managed memory size and always allocate from this pool,
> leaving the other pool empty.
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[jira] [Closed] (FLINK-14399) Add memory chunk reservation API to MemoryManager
[ https://issues.apache.org/jira/browse/FLINK-14399?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Andrey Zagrebin closed FLINK-14399. --- Fix Version/s: 1.10.0 Resolution: Done Merged into master by 036505f2afcde1ca4f0922cddafc3fe796ce45cb > Add memory chunk reservation API to MemoryManager > - > > Key: FLINK-14399 > URL: https://issues.apache.org/jira/browse/FLINK-14399 > Project: Flink > Issue Type: Sub-task > Components: Runtime / Task >Reporter: Andrey Zagrebin >Assignee: Andrey Zagrebin >Priority: Major > Fix For: 1.10.0 > > > MemoryManager allocates paged segments from the provided memory pools of > different types (on-/off-heap). Additionally, it can manage reservation and > release of arbitrarily sized chunks of memory from the same memory pools > respecting their overall limit. The way, how the memory is allocated, used > and freed, is then up to the memory user. MemoryManager is just a > book-keeping and limit checking component. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[GitHub] [flink] guoweiM commented on issue #9950: [FLINK-14464][runtime] Introduce the AbstractUserClassPathJobGraphRetriever
guoweiM commented on issue #9950: [FLINK-14464][runtime] Introduce the AbstractUserClassPathJobGraphRetriever URL: https://github.com/apache/flink/pull/9950#issuecomment-544930313 Thanks @zhuzhurk for reviewing. @flinkbot attention @tillrohrmann Could you have a look at this pr. Thanks. This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] xuyang1706 commented on a change in pull request #9300: [FLINK-13513][ml] Add the FlatMapper and related classes for later al…
xuyang1706 commented on a change in pull request #9300: [FLINK-13513][ml] Add
the FlatMapper and related classes for later al…
URL: https://github.com/apache/flink/pull/9300#discussion_r337474707
##
File path:
flink-ml-parent/flink-ml-lib/src/main/java/org/apache/flink/ml/common/mapper/Mapper.java
##
@@ -0,0 +1,49 @@
+/*
+ * 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.ml.common.mapper;
+
+import org.apache.flink.ml.api.misc.param.Params;
+import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.Collector;
+
+/**
+ * Abstract class for mappers.
+ */
+public abstract class Mapper extends FlatMapper implements MapOperable {
Review comment:
Thanks, refactored the definition of Mapper.
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] tillrohrmann commented on issue #9928: [FLINK-12122] Add support for spreading slots out across all TaskExecutors
tillrohrmann commented on issue #9928: [FLINK-12122] Add support for spreading slots out across all TaskExecutors URL: https://github.com/apache/flink/pull/9928#issuecomment-544929438 Thanks a lot for your review @xintongsong. I've addressed most of you comments. Concerning the configuration of this feature my reasoning was the following: I'm not entirely sure whether spread out for standalone and bin packing for Yarn is always the best option. Consider for example a workload which rather benefits from co-locating operators on the same `TaskExecutor` because of the reduced network I/O. These jobs would then degrade when being executed on a standalone cluster with slot spreading. Moreover, I would like to avoid to introduce too much magic. In the past we have seen that implicit behaviour can be hard to understand if not properly documented. Usually, I think that making things explicit and consistent across different setups is easier for users to understand. Since I don't see a superior option here, I believe that this option needs to be configurable. Concerning the config option itself: I opted for a boolean option because currently it is the simplest option for our use case and I cannot foresee future option values. In the past we have seen that optimizing/designing for future additions usually does not pay off because sometimes these additions won't come. The important bit here would be that we don't block the addition of more strategies in the future. I believe that this is the case. A potential strategy would be to deprecate `EVENLY_SPREAD_OUT_SLOTS_STRATEGY` and to use this value if the newly introduced configuration value has not been specified. This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[jira] [Updated] (FLINK-13984) Separate on-heap and off-heap managed memory pools
[
https://issues.apache.org/jira/browse/FLINK-13984?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Andrey Zagrebin updated FLINK-13984:
Fix Version/s: 1.10.0
> Separate on-heap and off-heap managed memory pools
> --
>
> Key: FLINK-13984
> URL: https://issues.apache.org/jira/browse/FLINK-13984
> Project: Flink
> Issue Type: Sub-task
> Components: Runtime / Task
>Reporter: Xintong Song
>Assignee: Andrey Zagrebin
>Priority: Major
> Labels: pull-request-available
> Fix For: 1.10.0
>
> Time Spent: 10m
> Remaining Estimate: 0h
>
> * Update {{MemoryManager}} to have two separated pools.
> * Extend {{MemoryManager}} interfaces to specify which pool to allocate
> memory from.
> Implement this step in common code paths for the legacy / new mode. For the
> legacy mode, depending to the configured memory type, we can set one of the
> two pools to the managed memory size and always allocate from this pool,
> leaving the other pool empty.
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[GitHub] [flink] tillrohrmann commented on a change in pull request #9928: [FLINK-12122] Add support for spreading slots out across all TaskExecutors
tillrohrmann commented on a change in pull request #9928: [FLINK-12122] Add
support for spreading slots out across all TaskExecutors
URL: https://github.com/apache/flink/pull/9928#discussion_r337474651
##
File path:
flink-core/src/main/java/org/apache/flink/configuration/ClusterOptions.java
##
@@ -50,4 +53,13 @@
.key("cluster.services.shutdown-timeout")
.defaultValue(3L)
.withDescription("The shutdown timeout for cluster services
like executors in milliseconds.");
+
+ public static final ConfigOption
EVENLY_SPREAD_OUT_SLOTS_STRATEGY = ConfigOptions
Review comment:
I've responded to your suggestion here:
https://github.com/apache/flink/pull/9928#issuecomment-544929438
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] xuyang1706 commented on a change in pull request #9300: [FLINK-13513][ml] Add the FlatMapper and related classes for later al…
xuyang1706 commented on a change in pull request #9300: [FLINK-13513][ml] Add
the FlatMapper and related classes for later al…
URL: https://github.com/apache/flink/pull/9300#discussion_r337473932
##
File path:
flink-ml-parent/flink-ml-lib/src/main/java/org/apache/flink/ml/common/mapper/FlatModelMapper.java
##
@@ -0,0 +1,138 @@
+/*
+ * 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.ml.common.mapper;
+
+import org.apache.flink.ml.api.misc.param.Params;
+import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.types.Row;
+
+import java.util.List;
+
+/**
+ * Abstract class for flatMappers with model.
+ *
+ * The general process of transform the input use machine learning model is:
+ *
+ * 1. load the model into memory.
+ * 2. process the input using the model.
+ *
+ * So, different from the {@link FlatMapper}, this class has a new abstract
method
+ * named {@link #loadModel(List)} that load the model and transform it to the
+ * memory structure.
+ *
+ * The model is the machine learning model that use the Table as
+ * its representation(serialized to Table from the memory
+ * or deserialized from Table to memory).
+ */
+public abstract class FlatModelMapper extends FlatMapper {
+
+ /**
+* schema of the model with Table type.
+*/
+ protected TableSchema modelSchema;
+
+ public FlatModelMapper(TableSchema modelSchema, TableSchema dataSchema,
Params params) {
+ super(dataSchema, params);
+ this.modelSchema = modelSchema;
+ }
+
+ /**
+* Load model from the list of Row type data.
+*
+* @param modelRows the list of Row type data
+*/
+ public abstract void loadModel(List modelRows);
+
+ /**
+* Generate new instance of given FlatModelMapper class without model
data.
+* The instance can not deal with real data, but it could be used to
get the output result schema.
+*
+* @param flatModelMapperClassName Name of the FlatModelMapper class
+* @param modelScheme The model scheme represented in
Table format.
+* @param dataSchema The schema of the input data
represented in Table format.
+* @param params The parameters for the instance
construction.
+* @return The object of {@link FlatModelMapper}
+* @throws Exception if flatModelMapperClass is not the class of {@link
FlatModelMapper}
+*/
+ public static FlatModelMapper of(
Review comment:
Thanks, changed the static method.
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[jira] [Updated] (FLINK-13984) Separate on-heap and off-heap managed memory pools
[
https://issues.apache.org/jira/browse/FLINK-13984?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
ASF GitHub Bot updated FLINK-13984:
---
Labels: pull-request-available (was: )
> Separate on-heap and off-heap managed memory pools
> --
>
> Key: FLINK-13984
> URL: https://issues.apache.org/jira/browse/FLINK-13984
> Project: Flink
> Issue Type: Sub-task
> Components: Runtime / Task
>Reporter: Xintong Song
>Assignee: Andrey Zagrebin
>Priority: Major
> Labels: pull-request-available
>
> * Update {{MemoryManager}} to have two separated pools.
> * Extend {{MemoryManager}} interfaces to specify which pool to allocate
> memory from.
> Implement this step in common code paths for the legacy / new mode. For the
> legacy mode, depending to the configured memory type, we can set one of the
> two pools to the managed memory size and always allocate from this pool,
> leaving the other pool empty.
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[GitHub] [flink] xuyang1706 commented on a change in pull request #9300: [FLINK-13513][ml] Add the FlatMapper and related classes for later al…
xuyang1706 commented on a change in pull request #9300: [FLINK-13513][ml] Add
the FlatMapper and related classes for later al…
URL: https://github.com/apache/flink/pull/9300#discussion_r337473854
##
File path:
flink-ml-parent/flink-ml-lib/src/main/java/org/apache/flink/ml/common/mapper/FlatMapper.java
##
@@ -0,0 +1,108 @@
+/*
+ * 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.ml.common.mapper;
+
+import org.apache.flink.ml.api.misc.param.Params;
+import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.Collector;
+
+import java.io.Serializable;
+
+/**
+ * Abstract class for flatMappers.
+ * FlatMapper maps a row to zero, one or multiple rows.
+ */
+public abstract class FlatMapper implements Serializable {
+
+ /**
+* schema of the input.
+*/
+ protected TableSchema dataSchema;
+
+ /**
+* params used for FlatMapper.
+* User can set the params before that the FlatMapper is executed.
+*/
+ protected Params params;
+
+ public FlatMapper(TableSchema dataSchema, Params params) {
+ this.dataSchema = dataSchema;
+ this.params = (null == params) ? new Params() : params.clone();
+ }
+
+ /**
+* The core method of the FlatMapper.
+* Takes a row from the input and maps it to multiple rows.
+*
+* @param rowThe input row.
+* @param output The collector for returning the result values.
+* @throws Exception This method may throw exceptions. Throwing an
exception will cause the operation
+* to fail.
+*/
+ public abstract void flatMap(Row row, Collector output) throws
Exception;
+
+ /**
+* Get the table schema(includes column names and types) of the
calculation result.
+*
+* @return the table schema of output Row type data
+*/
+ public abstract TableSchema getOutputSchema();
+
+ /**
+* Generate new instance of given FlatMapper class.
+*
+* @param flatMapperClassName Name of the FlatMapper class
+* @param dataSchema The schema of the input data represented
in Table format.
+* @param params The parameters for the instance
construction.
+* @return new instance of given FlatMapper class
+* @throws Exception if flatMapperClass is not the class of {@link
FlatMapper}
+*/
+ public static FlatMapper of(
Review comment:
Thanks, changed the static method.
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] zhuzhurk commented on issue #9950: [FLINK-14464][runtime] Introduce the AbstractUserClassPathJobGraphRetriever
zhuzhurk commented on issue #9950: [FLINK-14464][runtime] Introduce the AbstractUserClassPathJobGraphRetriever URL: https://github.com/apache/flink/pull/9950#issuecomment-544928450 Thanks @guoweiM for addressing all the comments. It looks to me now. +1 to merge it. This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] azagrebin merged pull request #9693: [FLINK-13984] Separate on-heap and off-heap managed memory pools
azagrebin merged pull request #9693: [FLINK-13984] Separate on-heap and off-heap managed memory pools URL: https://github.com/apache/flink/pull/9693 This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] guoweiM commented on a change in pull request #9950: [FLINK-14464][runtime] Introduce the AbstractUserClassPathJobGraphRetriever
guoweiM commented on a change in pull request #9950: [FLINK-14464][runtime]
Introduce the AbstractUserClassPathJobGraphRetriever
URL: https://github.com/apache/flink/pull/9950#discussion_r337472932
##
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/component/AbstractUserClassPathJobGraphRetriever.java
##
@@ -0,0 +1,117 @@
+/*
+ * 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.entrypoint.component;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.net.URL;
+import java.nio.file.FileVisitOption;
+import java.nio.file.FileVisitResult;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.nio.file.SimpleFileVisitor;
+import java.nio.file.attribute.BasicFileAttributes;
+import java.util.Collections;
+import java.util.EnumSet;
+import java.util.LinkedList;
+import java.util.List;
+
+/**
+ * Abstract class for the JobGraphRetriever, which wants to get classpath
user's code depends on.
+ */
+public abstract class AbstractUserClassPathJobGraphRetriever implements
JobGraphRetriever {
+
+ protected static final Logger LOG =
LoggerFactory.getLogger(AbstractUserClassPathJobGraphRetriever.class);
Review comment:
I think it is ok. `ClassPathJobGraphRetriever` and `FileJobGraphRetriever`
will use this `LOG` in the following commits.
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] guoweiM commented on a change in pull request #9950: [FLINK-14464][runtime] Introduce the AbstractUserClassPathJobGraphRetriever
guoweiM commented on a change in pull request #9950: [FLINK-14464][runtime]
Introduce the AbstractUserClassPathJobGraphRetriever
URL: https://github.com/apache/flink/pull/9950#discussion_r337472932
##
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/component/AbstractUserClassPathJobGraphRetriever.java
##
@@ -0,0 +1,117 @@
+/*
+ * 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.entrypoint.component;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.net.URL;
+import java.nio.file.FileVisitOption;
+import java.nio.file.FileVisitResult;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.nio.file.SimpleFileVisitor;
+import java.nio.file.attribute.BasicFileAttributes;
+import java.util.Collections;
+import java.util.EnumSet;
+import java.util.LinkedList;
+import java.util.List;
+
+/**
+ * Abstract class for the JobGraphRetriever, which wants to get classpath
user's code depends on.
+ */
+public abstract class AbstractUserClassPathJobGraphRetriever implements
JobGraphRetriever {
+
+ protected static final Logger LOG =
LoggerFactory.getLogger(AbstractUserClassPathJobGraphRetriever.class);
Review comment:
I think it is ok. `ClassPathJobGraphRetriever` and `FileJobGraphRetriever`
will use this `LOG` in the following comments.
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #9542: [FLINK-13873][metrics] Change the column family as tags for influxdb …
flinkbot edited a comment on issue #9542: [FLINK-13873][metrics] Change the column family as tags for influxdb … URL: https://github.com/apache/flink/pull/9542#issuecomment-525276537 ## CI report: * e8636926351f3d406962dcadba275e20e49aff39 : SUCCESS [Build](https://travis-ci.com/flink-ci/flink/builds/124736151) * d23ea97e8419bbacea0698b3ba82a459d940cf38 : CANCELED [Build](https://travis-ci.com/flink-ci/flink/builds/128606376) * 05977cd49eb306d768668be4e8cb31034343df02 : FAILURE [Build](https://travis-ci.com/flink-ci/flink/builds/128606947) * 057d453e5e00656bcfcb87d1d69172f614b2d11f : CANCELED [Build](https://travis-ci.com/flink-ci/flink/builds/128681895) * 5001042b5fc5202da14c06e2d21faf1427f50a66 : FAILURE [Build](https://travis-ci.com/flink-ci/flink/builds/128683546) * 3a1e39e889daf0fca82c54982911ded35df6cb77 : SUCCESS [Build](https://travis-ci.com/flink-ci/flink/builds/128692102) * 2745a3cb02e601a1a26360e9d6b3f0af5428c66a : SUCCESS [Build](https://travis-ci.com/flink-ci/flink/builds/129027621) * e108aa24e739d6f48f63efc560c3fc049b509860 : FAILURE [Build](https://travis-ci.com/flink-ci/flink/builds/131172797) * 24a270d468dea677379713d5cf402ea453d9f222 : FAILURE [Build](https://travis-ci.com/flink-ci/flink/builds/131246042) * acf1c2b9add8c3b903a8485ed41c9f0b18d97729 : SUCCESS [Build](https://travis-ci.com/flink-ci/flink/builds/131260202) * 2f5f28b6bc2c24e81b330a3ad65cde2a23a1af95 : FAILURE [Build](https://travis-ci.com/flink-ci/flink/builds/132564124) * 1d5388df33162240b94852f742ce6d7d7a468be4 : SUCCESS [Build](https://travis-ci.com/flink-ci/flink/builds/132974400) Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] zhuzhurk commented on a change in pull request #9950: [FLINK-14464][runtime] Introduce the AbstractUserClassPathJobGraphRetriever
zhuzhurk commented on a change in pull request #9950: [FLINK-14464][runtime]
Introduce the AbstractUserClassPathJobGraphRetriever
URL: https://github.com/apache/flink/pull/9950#discussion_r337468217
##
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/component/AbstractUserClassPathJobGraphRetriever.java
##
@@ -0,0 +1,117 @@
+/*
+ * 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.entrypoint.component;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.net.URL;
+import java.nio.file.FileVisitOption;
+import java.nio.file.FileVisitResult;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.nio.file.SimpleFileVisitor;
+import java.nio.file.attribute.BasicFileAttributes;
+import java.util.Collections;
+import java.util.EnumSet;
+import java.util.LinkedList;
+import java.util.List;
+
+/**
+ * Abstract class for the JobGraphRetriever, which wants to get classpath
user's code depends on.
+ */
+public abstract class AbstractUserClassPathJobGraphRetriever implements
JobGraphRetriever {
+
+ protected static final Logger LOG =
LoggerFactory.getLogger(AbstractUserClassPathJobGraphRetriever.class);
Review comment:
It can be private since it is static now. And future subclass like
`ClassPathJobGraphRetriever` has its own logger.
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] tillrohrmann commented on a change in pull request #9928: [FLINK-12122] Add support for spreading slots out across all TaskExecutors
tillrohrmann commented on a change in pull request #9928: [FLINK-12122] Add
support for spreading slots out across all TaskExecutors
URL: https://github.com/apache/flink/pull/9928#discussion_r337467084
##
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/EvenlySpreadOutLocationPreferenceSlotSelectionStrategy.java
##
@@ -0,0 +1,44 @@
+/*
+ * 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.jobmaster.slotpool;
+
+import org.apache.flink.runtime.clusterframework.types.ResourceProfile;
+import org.apache.flink.runtime.jobmanager.scheduler.Locality;
+
+import javax.annotation.Nonnull;
+
+import java.util.Collection;
+import java.util.Comparator;
+import java.util.Optional;
+
+class EvenlySpreadOutLocationPreferenceSlotSelectionStrategy extends
LocationPreferenceSlotSelectionStrategy {
+ @Nonnull
+ @Override
+ protected Optional
selectWithoutLocationPreference(@Nonnull Collection
availableSlots, @Nonnull ResourceProfile resourceProfile) {
+ return availableSlots.stream()
+ .filter(slotInfoAndResources ->
slotInfoAndResources.getRemainingResources().isMatching(resourceProfile))
+
.min(Comparator.comparing(SlotInfoAndResources::getTaskExecutorUtilization))
+ .map(slotInfoAndResources ->
SlotInfoAndLocality.of(slotInfoAndResources.getSlotInfo(),
Locality.UNCONSTRAINED));
+ }
+
+ @Override
+ protected double calculateCandidateScore(int localWeigh, int
hostLocalWeigh, double taskExecutorUtilization) {
+ return localWeigh * 20 + hostLocalWeigh * 2 -
taskExecutorUtilization;
Review comment:
I'll add a comment for this.
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #9968: [hotfix][typo] fix typo in YarnClusterDescriptor
flinkbot edited a comment on issue #9968: [hotfix][typo] fix typo in YarnClusterDescriptor URL: https://github.com/apache/flink/pull/9968#issuecomment-544895954 ## CI report: * 76cbd6febfa330bfecab18176279e059241443b7 : SUCCESS [Build](https://travis-ci.com/flink-ci/flink/builds/132970317) Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] wangyang0918 commented on a change in pull request #9965: [FLINK-10935][kubernetes]Implement KubeClient with Faric8 Kubernetes clients
wangyang0918 commented on a change in pull request #9965:
[FLINK-10935][kubernetes]Implement KubeClient with Faric8 Kubernetes clients
URL: https://github.com/apache/flink/pull/9965#discussion_r337465776
##
File path:
flink-kubernetes/src/main/java/org/apache/flink/kubernetes/configuration/KubernetesConfigOptions.java
##
@@ -0,0 +1,115 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.kubernetes.configuration;
+
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.kubernetes.cli.KubernetesCliOptions;
+
+import static org.apache.flink.configuration.ConfigOptions.key;
+
+/**
+ * This class holds configuration constants used by Flink's kubernetes runners.
+ */
+public class KubernetesConfigOptions {
+
+ public static final ConfigOption REST_SERVICE_EXPOSED_TYPE =
+ key("kubernetes.rest-service.exposed.type")
+ .defaultValue(ServiceExposedType.LoadBalancer.toString())
+ .withDescription("It could be
ClusterIP/NodePort/LoadBalancer(default). When set to ClusterIP, the rest
service" +
+ "will not be created.");
+
+ public static final ConfigOption JOB_MANAGER_SERVICE_ACCOUNT =
+ key("kubernetes.jobmanager.service-account")
+ .defaultValue("default")
+ .withDescription("Service account that is used by jobmanager
within kubernetes cluster. " +
+ "The job manager uses this service account when
requesting taskmanager pods from the API server.");
+
+ public static final ConfigOption JOB_MANAGER_CPU =
+ key("kubernetes.jobmanager.cpu")
+ .defaultValue(1.0)
+ .withDescription("The number of cpu used by job manager");
+
+ public static final ConfigOption TASK_MANAGER_CPU =
+ key("kubernetes.taskmanager.cpu")
+ .defaultValue(1.0)
+ .withDescription("The number of cpu used by task manager");
+
+ public static final ConfigOption CONTAINER_IMAGE_PULL_POLICY =
+ key("kubernetes.container.image.pullPolicy")
+ .defaultValue("Always")
+ .withDescription("Kubernetes image pull policy. Valid values
are Always, Never, and IfNotPresent.");
+
+ public static final ConfigOption KUBE_CONFIG_FILE =
+ key("kubernetes.config.file")
+ .noDefaultValue()
+ .withDescription("The kubernetes config file will be used to
create the client. The default " +
+ "is located at ~/.kube/config");
+
+ public static final ConfigOption NAME_SPACE =
Review comment:
Nice catch. I will fix it.
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] tillrohrmann commented on a change in pull request #9928: [FLINK-12122] Add support for spreading slots out across all TaskExecutors
tillrohrmann commented on a change in pull request #9928: [FLINK-12122] Add
support for spreading slots out across all TaskExecutors
URL: https://github.com/apache/flink/pull/9928#discussion_r337464754
##
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/SlotSharingManager.java
##
@@ -185,18 +186,55 @@ MultiTaskSlot createRootSlot(
return resolvedRootSlots
.values()
.stream()
- .flatMap((Map map)
-> map.values().stream())
-
.filter(validMultiTaskSlotAndDoesNotContain(groupId))
- .map((MultiTaskSlot multiTaskSlot) -> {
- SlotInfo slotInfo =
multiTaskSlot.getSlotContextFuture().join();
+ .flatMap((Map map)
-> filterUsedAndReleasingMultiTaskSlots(map, groupId))
+ .map((MultiTaskSlotInfo multiTaskSlotInfo) -> {
+ SlotInfo slotInfo =
multiTaskSlotInfo.getSlotInfo();
return new
SlotSelectionStrategy.SlotInfoAndResources(
- slotInfo,
-
slotInfo.getResourceProfile().subtract(multiTaskSlot.getReservedResources()));
+ slotInfo,
+
slotInfo.getResourceProfile().subtract(multiTaskSlotInfo.getReservedResources()),
+
multiTaskSlotInfo.getTaskExecutorUtilization());
}).collect(Collectors.toList());
}
+ private Stream
filterUsedAndReleasingMultiTaskSlots(Map
taskExecutorSlots, AbstractID groupId) {
Review comment:
what about `createValidMultiTaskSlotInfos`?
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] zhuzhurk commented on a change in pull request #9950: [FLINK-14464][runtime] Introduce the AbstractUserClassPathJobGraphRetriever
zhuzhurk commented on a change in pull request #9950: [FLINK-14464][runtime]
Introduce the AbstractUserClassPathJobGraphRetriever
URL: https://github.com/apache/flink/pull/9950#discussion_r337462709
##
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/component/AbstractUserClassPathJobGraphRetriever.java
##
@@ -0,0 +1,116 @@
+/*
+ * 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.entrypoint.component;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.net.URL;
+import java.nio.file.FileVisitOption;
+import java.nio.file.FileVisitResult;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.nio.file.SimpleFileVisitor;
+import java.nio.file.attribute.BasicFileAttributes;
+import java.util.Collections;
+import java.util.EnumSet;
+import java.util.LinkedList;
+import java.util.List;
+
+/**
+ * Abstract class for the JobGraphRetriever, which wants to get classpath
user's code depends on.
+ */
+public abstract class AbstractUserClassPathJobGraphRetriever implements
JobGraphRetriever {
+
+ protected static final Logger LOG =
LoggerFactory.getLogger(AbstractUserClassPathJobGraphRetriever.class);
+
+ /** The directory contains all the jars, which user code depends on. */
+ @Nullable
+ private final String jobDir;
+
+ private List userClassPaths;
+
+ public AbstractUserClassPathJobGraphRetriever(String jobDir) {
+ this.jobDir = jobDir;
+ }
+
+ public List getUserClassPaths() throws IOException {
+ if (userClassPaths == null) {
+ userClassPaths = getRelativeJarsURLFromDir(jobDir);
+ }
+ return userClassPaths;
+ }
+
+ /**
+* Scan all the jar files in the {@code dir} and return all these jar
files' relative URLs to "user.dir".
+* @param dir the dir needed to scan the jar files
+* @return the jar files' relative URLs
+* @throws IOException
+*/
+ private List getRelativeJarsURLFromDir(String dir) throws
IOException {
+
+ if (dir == null) {
+ return Collections.emptyList();
+ }
+
+ final List jarURLs = new LinkedList<>();
+ if (!Files.exists(Paths.get(dir))) {
+ throw new IllegalArgumentException("the job dir " + dir
+ " dose not exists.");
+ }
+ if (!Files.isDirectory(Paths.get(dir))) {
+ throw new IllegalArgumentException("the job dir " + dir
+ " is not a directory.");
+ }
+
+ Path dirPath;
+ if (Paths.get(dir).isAbsolute()) {
+ dirPath =
Paths.get(System.getProperty("user.dir")).relativize(Paths.get(dir));
+ } else {
+ dirPath = Paths.get(dir);
+ }
+ Files.walkFileTree(
+ dirPath,
+ EnumSet.of(FileVisitOption.FOLLOW_LINKS),
+ Integer.MAX_VALUE,
+ new SimpleFileVisitor() {
+
+ @Override
+ public FileVisitResult visitFile(java.nio.file.Path
file, BasicFileAttributes attrs)
+ throws IOException {
+ FileVisitResult fileVisitResult =
super.visitFile(file, attrs);
+ if
(file.getFileName().toString().endsWith(".jar")) {
+ LOG.info("add " + file.toString() + "
to user classpath");
+ jarURLs.add(
Review comment:
bad indentation.
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git
[GitHub] [flink] flinkbot edited a comment on issue #9832: [FLINK-11843] Bind lifespan of Dispatcher to leader session
flinkbot edited a comment on issue #9832: [FLINK-11843] Bind lifespan of Dispatcher to leader session URL: https://github.com/apache/flink/pull/9832#issuecomment-537049332 ## CI report: * abaae048fef753455970fac9d6ab421b660b0536 : UNKNOWN * b96c63552ccd322adae7a41a410615e95b538ece : SUCCESS [Build](https://travis-ci.com/flink-ci/flink/builds/129867595) * 2c95a3939dbf0259d694af6c69451f0ede3c3891 : SUCCESS [Build](https://travis-ci.com/flink-ci/flink/builds/130066310) * 6e030add922011ea54178690f171911d0139f14b : SUCCESS [Build](https://travis-ci.com/flink-ci/flink/builds/130858235) * 271703eda6f6c55b1641a54206109ef659f62854 : SUCCESS [Build](https://travis-ci.com/flink-ci/flink/builds/131454171) * 730c3ff9eef5f41f917095eca2f0f5d1e90aea53 : FAILURE [Build](https://travis-ci.com/flink-ci/flink/builds/132974433) Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[jira] [Commented] (FLINK-14429) Wrong app final status when running batch job on yarn with non-detached mode
[
https://issues.apache.org/jira/browse/FLINK-14429?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16956946#comment-16956946
]
liupengcheng commented on FLINK-14429:
--
Thanks [~trohrmann] [~aljoscha] , I got it!
>Conceptually, I think the client should not tell the cluster with which return
>code to terminate. This seems like a wrong separation of concerns. The cluster
>>should make this decision instead.
I agree with it, the cluster should make this decision itself. BTW, this make
it more similar with spark exit behavior. This is good for design, but
sometimes our users always complains that the app final status is not
consistent with the actual SUCCESS of their job in yarn-client/non-detached
mode.
> Wrong app final status when running batch job on yarn with non-detached mode
>
>
> Key: FLINK-14429
> URL: https://issues.apache.org/jira/browse/FLINK-14429
> Project: Flink
> Issue Type: Bug
> Components: Deployment / YARN
>Affects Versions: 1.9.0
>Reporter: liupengcheng
>Priority: Minor
> Labels: pull-request-available
> Attachments: image-2019-10-17-16-47-47-038.png
>
> Time Spent: 10m
> Remaining Estimate: 0h
>
> Recently, we found that the app final status is not correct when an
> application failed when running batch job on yarn with non-detached mode, It
> reported SUCCEEDED but FAILED is what we expected.
> !image-2019-10-17-16-47-47-038.png!
>
> But the logs and client reported error and job failed(It's caused by OOM):
> {code:java}
> 2019-10-10 14:36:21,797 INFO
> org.apache.flink.runtime.executiongraph.ExecutionGraph- Job TeraSort
> (d82cbfaae905c695597083b1476e51b8) switched from state FAILING to FAILED.
> org.apache.flink.runtime.io.network.partition.consumer.PartitionConnectionException:
> Connection for partition
> a254412fc7464cd4e0fe04ab9e3a6309@8d5afff58c86dd7f5bc78946f0101699 not
> reachable.
> at
> org.apache.flink.runtime.io.network.partition.consumer.RemoteInputChannel.requestSubpartition(RemoteInputChannel.java:168)
> at
> org.apache.flink.runtime.io.network.partition.consumer.SingleInputGate.requestPartitions(SingleInputGate.java:237)
> at
> org.apache.flink.runtime.io.network.partition.consumer.SingleInputGate.setup(SingleInputGate.java:215)
> at
> org.apache.flink.runtime.taskmanager.InputGateWithMetrics.setup(InputGateWithMetrics.java:65)
> at
> org.apache.flink.runtime.taskmanager.Task.setupPartitionsAndGates(Task.java:866)
> at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:621)
> at org.apache.flink.runtime.taskmanager.Task.run(Task.java:530)
> at java.lang.Thread.run(Thread.java:748)
> Caused by: java.io.IOException: Connecting the channel failed: Connecting to
> remote task manager + 'zjy-hadoop-prc-st164.bj/10.152.47.8:45704' has failed.
> This might indicate that the remote task manager has been lost.
> at
> org.apache.flink.runtime.io.network.netty.PartitionRequestClientFactory$ConnectingChannel.waitForChannel(PartitionRequestClientFactory.java:197)
> at
> org.apache.flink.runtime.io.network.netty.PartitionRequestClientFactory$ConnectingChannel.access$000(PartitionRequestClientFactory.java:134)
> at
> org.apache.flink.runtime.io.network.netty.PartitionRequestClientFactory.createPartitionRequestClient(PartitionRequestClientFactory.java:86)
> at
> org.apache.flink.runtime.io.network.netty.NettyConnectionManager.createPartitionRequestClient(NettyConnectionManager.java:68)
> at
> org.apache.flink.runtime.io.network.partition.consumer.RemoteInputChannel.requestSubpartition(RemoteInputChannel.java:165)
> ... 7 more
> Caused by:
> org.apache.flink.runtime.io.network.netty.exception.RemoteTransportException:
> Connecting to remote task manager +
> 'zjy-hadoop-prc-st164.bj/10.152.47.8:45704' has failed. This might indicate
> that the remote task manager has been lost.
> at
> org.apache.flink.runtime.io.network.netty.PartitionRequestClientFactory$ConnectingChannel.operationComplete(PartitionRequestClientFactory.java:220)
> at
> org.apache.flink.runtime.io.network.netty.PartitionRequestClientFactory$ConnectingChannel.operationComplete(PartitionRequestClientFactory.java:134)
> at
> org.apache.flink.shaded.netty4.io.netty.util.concurrent.DefaultPromise.notifyListener0(DefaultPromise.java:511)
> at
> org.apache.flink.shaded.netty4.io.netty.util.concurrent.DefaultPromise.notifyListeners0(DefaultPromise.java:504)
> at
> org.apache.flink.shaded.netty4.io.netty.util.concurrent.DefaultPromise.notifyListenersNow(DefaultPromise.java:483)
> at
>
[GitHub] [flink] zhuzhurk commented on a change in pull request #9950: [FLINK-14464][runtime] Introduce the AbstractUserClassPathJobGraphRetriever
zhuzhurk commented on a change in pull request #9950: [FLINK-14464][runtime]
Introduce the AbstractUserClassPathJobGraphRetriever
URL: https://github.com/apache/flink/pull/9950#discussion_r337459475
##
File path:
flink-runtime/src/test/java/org/apache/flink/runtime/entrypoint/component/AbstractUserClassPathJobGraphRetrieverTest.java
##
@@ -0,0 +1,148 @@
+/*
+ * 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.entrypoint.component;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.jobgraph.JobGraph;
+import org.apache.flink.util.TestLogger;
+
+import org.apache.commons.collections.CollectionUtils;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+import java.io.File;
+import java.io.IOException;
+import java.net.URL;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+/**
+ * Tests for the {@link AbstractUserClassPathJobGraphRetriever}.
+ */
+public class AbstractUserClassPathJobGraphRetrieverTest extends TestLogger {
+
+ @Rule
+ public final TemporaryFolder temporaryFolder = new TemporaryFolder();
+
+ /**
+* Test class.
Review comment:
It's better to remove this useless comment.
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] zhuzhurk commented on a change in pull request #9950: [FLINK-14464][runtime] Introduce the AbstractUserClassPathJobGraphRetriever
zhuzhurk commented on a change in pull request #9950: [FLINK-14464][runtime]
Introduce the AbstractUserClassPathJobGraphRetriever
URL: https://github.com/apache/flink/pull/9950#discussion_r337458869
##
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/component/AbstractUserClassPathJobGraphRetriever.java
##
@@ -0,0 +1,115 @@
+/*
+ * 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.entrypoint.component;
+
+import org.apache.flink.core.fs.Path;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import java.io.File;
+import java.io.IOException;
+import java.net.URL;
+import java.nio.file.FileVisitOption;
+import java.nio.file.FileVisitResult;
+import java.nio.file.Files;
+import java.nio.file.SimpleFileVisitor;
+import java.nio.file.attribute.BasicFileAttributes;
+import java.util.Collections;
+import java.util.EnumSet;
+import java.util.LinkedList;
+import java.util.List;
+
+/**
+ * Abstract class for the JobGraphRetriever, which wants to get classpath
user's code depends on.
+ */
+
+public abstract class AbstractUserClassPathJobGraphRetriever implements
JobGraphRetriever {
+
+ protected static final Logger LOG =
LoggerFactory.getLogger(AbstractUserClassPathJobGraphRetriever.class);
+
+ public static final String DEFAULT_JOB_DIR = "job";
+
+ /** The directory contains all the jars, which user code depends on. */
+ @Nullable
+ private final String jobDir;
+
+ private List userClassPaths;
+
+ public AbstractUserClassPathJobGraphRetriever(String jobDir) {
+ this.jobDir = jobDir;
+ }
+
+ public List getUserClassPaths() throws IOException {
+ if (userClassPaths == null) {
+ userClassPaths = scanJarsInJobClassDir(jobDir);
+ }
+ return userClassPaths;
+ }
+
+ private List scanJarsInJobClassDir(String dir) throws IOException {
+
+ if (dir == null) {
+ return Collections.emptyList();
+ }
+
+ final File dirFile = new File(new Path(dir).toString());
+ final List jarURLs = new LinkedList<>();
+
+ if (!dirFile.exists()) {
+ LOG.warn("the job dir " + dirFile + " dose not
exists.");
+ return Collections.emptyList();
+ }
+ if (!dirFile.isDirectory()) {
+ LOG.warn("the job dir " + dirFile + " is not a
directory.");
+ return Collections.emptyList();
+ }
+
+ Files.walkFileTree(dirFile.toPath(),
+ EnumSet.of(FileVisitOption.FOLLOW_LINKS),
+ Integer.MAX_VALUE,
+ new SimpleFileVisitor(){
+
+ @Override
+ public FileVisitResult visitFile(java.nio.file.Path
file, BasicFileAttributes attrs)
+ throws IOException {
+ FileVisitResult fileVisitResult =
super.visitFile(file, attrs);
+ if
(file.getFileName().toString().endsWith(".jar")) {
+ LOG.info("add " + file.toString() + "
to user classpath");
+ if (file.isAbsolute()) {
+
jarURLs.add(file.toUri().toURL());
+ } else {
+ jarURLs.add(
+ new URL(new
URL(file.getFileName().toUri().getScheme() + ":"), file.toString())
Review comment:
> MalformedURLException - if no protocol is specified, or an unknown
protocol is found, or spec is null.
Ok. Looks the main reason is the protocol needs to be specified to avoid
`MalformedURLException`.
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific
[GitHub] [flink] flinkbot edited a comment on issue #9950: [FLINK-14464][runtime] Introduce the AbstractUserClassPathJobGraphRetriever
flinkbot edited a comment on issue #9950: [FLINK-14464][runtime] Introduce the AbstractUserClassPathJobGraphRetriever URL: https://github.com/apache/flink/pull/9950#issuecomment-544414033 ## CI report: * decba8623c001e8bbe5dc797de2bc421b2e216ef : SUCCESS [Build](https://travis-ci.com/flink-ci/flink/builds/132777340) * 6c7ffb9f9049016ce874b9790cd51ed337cfec00 : PENDING [Build](https://travis-ci.com/flink-ci/flink/builds/132974469) Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] wangyang0918 commented on a change in pull request #9965: [FLINK-10935][kubernetes]Implement KubeClient with Faric8 Kubernetes clients
wangyang0918 commented on a change in pull request #9965:
[FLINK-10935][kubernetes]Implement KubeClient with Faric8 Kubernetes clients
URL: https://github.com/apache/flink/pull/9965#discussion_r337454573
##
File path:
flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/decorators/FlinkMasterDeploymentDecorator.java
##
@@ -0,0 +1,180 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.kubernetes.kubeclient.decorators;
+
+import org.apache.flink.client.cli.CliFrontend;
+import org.apache.flink.client.deployment.ClusterSpecification;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.JobManagerOptions;
+import org.apache.flink.configuration.RestOptions;
+import org.apache.flink.kubernetes.configuration.KubernetesConfigOptions;
+import org.apache.flink.kubernetes.kubeclient.resources.FlinkDeployment;
+import org.apache.flink.kubernetes.utils.Constants;
+import org.apache.flink.kubernetes.utils.KubernetesUtils;
+import org.apache.flink.util.Preconditions;
+
+import io.fabric8.kubernetes.api.model.ConfigMapVolumeSourceBuilder;
+import io.fabric8.kubernetes.api.model.Container;
+import io.fabric8.kubernetes.api.model.ContainerBuilder;
+import io.fabric8.kubernetes.api.model.ContainerPortBuilder;
+import io.fabric8.kubernetes.api.model.KeyToPath;
+import io.fabric8.kubernetes.api.model.PodSpec;
+import io.fabric8.kubernetes.api.model.PodSpecBuilder;
+import io.fabric8.kubernetes.api.model.Quantity;
+import io.fabric8.kubernetes.api.model.ResourceRequirementsBuilder;
+import io.fabric8.kubernetes.api.model.Volume;
+import io.fabric8.kubernetes.api.model.VolumeMountBuilder;
+import io.fabric8.kubernetes.api.model.apps.Deployment;
+import io.fabric8.kubernetes.api.model.apps.DeploymentSpecBuilder;
+
+import java.io.File;
+import java.util.Arrays;
+import java.util.Map;
+
+import static
org.apache.flink.configuration.GlobalConfiguration.FLINK_CONF_FILENAME;
+import static org.apache.flink.kubernetes.utils.Constants.BLOB_SERVER_PORT;
+import static
org.apache.flink.kubernetes.utils.Constants.CONFIG_FILE_LOG4J_NAME;
+import static
org.apache.flink.kubernetes.utils.Constants.CONFIG_FILE_LOGBACK_NAME;
+import static org.apache.flink.kubernetes.utils.Constants.CONFIG_MAP_PREFIX;
+import static org.apache.flink.kubernetes.utils.Constants.FLINK_CONF_VOLUME;
+
+/**
+ * Flink master specific deployment configuration.
+ * */
+public class FlinkMasterDeploymentDecorator extends Decorator {
+
+ private static final String CONTAINER_NAME = "flink-job-manager";
+
+ private final ClusterSpecification clusterSpecification;
+
+ public static final ConfigOption ENTRY_POINT_CLASS =
ConfigOptions
+ .key("kubernetes.internal.jobmanager.entrypoint.class")
+ .noDefaultValue()
+ .withDescription("The entrypoint class for jobmanager. It will
be set in kubernetesClusterDescriptor.");
+
+ public static final ConfigOption ENTRY_POINT_CLASS_ARGS =
ConfigOptions
+ .key("kubernetes.internal.jobmanager.entrypoint.class.args")
+ .noDefaultValue()
+ .withDescription("The args of entrypoint class for jobmanager.
It will be set in FlinkKubernetesCustomCli.");
+
+ public FlinkMasterDeploymentDecorator(ClusterSpecification
clusterSpecification) {
+ this.clusterSpecification = clusterSpecification;
+ }
+
+ @Override
+ protected Deployment doDecorate(Deployment deployment, Configuration
flinkConfig) {
+ String clusterId =
flinkConfig.getString(KubernetesConfigOptions.CLUSTER_ID);
+ Preconditions.checkNotNull(clusterId, "ClusterId must be
specified!");
+
+ String mainClass = flinkConfig.getString(ENTRY_POINT_CLASS);
+ Preconditions.checkNotNull(mainClass, "Main class must be
specified!");
+
+ Map labels = LabelBuilder
+ .withExist(deployment.getMetadata().getLabels())
+ .withJobManagerComponent()
+
[GitHub] [flink] wangyang0918 commented on a change in pull request #9965: [FLINK-10935][kubernetes]Implement KubeClient with Faric8 Kubernetes clients
wangyang0918 commented on a change in pull request #9965:
[FLINK-10935][kubernetes]Implement KubeClient with Faric8 Kubernetes clients
URL: https://github.com/apache/flink/pull/9965#discussion_r337454017
##
File path:
flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/decorators/TaskManagerPodDecorator.java
##
@@ -0,0 +1,136 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.kubernetes.kubeclient.decorators;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.kubernetes.configuration.KubernetesConfigOptions;
+import org.apache.flink.kubernetes.kubeclient.TaskManagerPodParameter;
+import org.apache.flink.kubernetes.kubeclient.resources.FlinkPod;
+import org.apache.flink.kubernetes.utils.Constants;
+import org.apache.flink.util.Preconditions;
+
+import io.fabric8.kubernetes.api.model.ConfigMapVolumeSourceBuilder;
+import io.fabric8.kubernetes.api.model.Container;
+import io.fabric8.kubernetes.api.model.ContainerBuilder;
+import io.fabric8.kubernetes.api.model.ContainerPortBuilder;
+import io.fabric8.kubernetes.api.model.EnvVar;
+import io.fabric8.kubernetes.api.model.KeyToPath;
+import io.fabric8.kubernetes.api.model.Pod;
+import io.fabric8.kubernetes.api.model.PodSpecBuilder;
+import io.fabric8.kubernetes.api.model.Quantity;
+import io.fabric8.kubernetes.api.model.ResourceRequirementsBuilder;
+import io.fabric8.kubernetes.api.model.Volume;
+import io.fabric8.kubernetes.api.model.VolumeMountBuilder;
+
+import java.io.File;
+import java.util.Arrays;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import static
org.apache.flink.configuration.GlobalConfiguration.FLINK_CONF_FILENAME;
+import static
org.apache.flink.kubernetes.utils.Constants.CONFIG_FILE_LOG4J_NAME;
+import static
org.apache.flink.kubernetes.utils.Constants.CONFIG_FILE_LOGBACK_NAME;
+import static org.apache.flink.kubernetes.utils.Constants.CONFIG_MAP_PREFIX;
+import static org.apache.flink.kubernetes.utils.Constants.FLINK_CONF_VOLUME;
+
+/**
+ * Task manager specific pod configuration.
+ * */
+public class TaskManagerPodDecorator extends Decorator {
+
+ private static final String CONTAINER_NAME = "flink-task-manager";
+
+ private final TaskManagerPodParameter parameter;
+
+ public TaskManagerPodDecorator(TaskManagerPodParameter parameters) {
+ Preconditions.checkNotNull(parameters);
+ this.parameter = parameters;
+ }
+
+ @Override
+ protected Pod doDecorate(Pod pod, Configuration flinkConfig) {
+
+ String clusterId =
flinkConfig.getString(KubernetesConfigOptions.CLUSTER_ID);
+ Preconditions.checkNotNull(clusterId, "ClusterId must be
specified!");
+
+ Map labels =
org.apache.flink.kubernetes.kubeclient.decorators.LabelBuilder
+ .withExist(pod.getMetadata().getLabels())
+ .withTaskManagerComponent()
+ .toLabels();
+
+ pod.getMetadata().setLabels(labels);
+ pod.getMetadata().setName(this.parameter.getPodName());
+
+ Volume configMapVolume = new Volume();
+ configMapVolume.setName(FLINK_CONF_VOLUME);
+ configMapVolume.setConfigMap(new ConfigMapVolumeSourceBuilder()
+ .withName(CONFIG_MAP_PREFIX + clusterId)
+ .withItems(Arrays.asList(
+ new KeyToPath(FLINK_CONF_FILENAME, null,
FLINK_CONF_FILENAME),
+ new KeyToPath(CONFIG_FILE_LOG4J_NAME, null,
CONFIG_FILE_LOG4J_NAME),
+ new KeyToPath(CONFIG_FILE_LOGBACK_NAME, null,
CONFIG_FILE_LOGBACK_NAME)))
+ .build());
+
+ pod.setSpec(new PodSpecBuilder()
+ .withVolumes(configMapVolume)
+ .withContainers(createTaskManagerContainer(flinkConfig))
+ .build());
+ return pod;
+ }
+
+ private Container createTaskManagerContainer(Configuration flinkConfig)
{
+ Quantity taskManagerCpuQuantity = new
[GitHub] [flink] flinkbot edited a comment on issue #9832: [FLINK-11843] Bind lifespan of Dispatcher to leader session
flinkbot edited a comment on issue #9832: [FLINK-11843] Bind lifespan of Dispatcher to leader session URL: https://github.com/apache/flink/pull/9832#issuecomment-537049332 ## CI report: * abaae048fef753455970fac9d6ab421b660b0536 : UNKNOWN * b96c63552ccd322adae7a41a410615e95b538ece : SUCCESS [Build](https://travis-ci.com/flink-ci/flink/builds/129867595) * 2c95a3939dbf0259d694af6c69451f0ede3c3891 : SUCCESS [Build](https://travis-ci.com/flink-ci/flink/builds/130066310) * 6e030add922011ea54178690f171911d0139f14b : SUCCESS [Build](https://travis-ci.com/flink-ci/flink/builds/130858235) * 271703eda6f6c55b1641a54206109ef659f62854 : SUCCESS [Build](https://travis-ci.com/flink-ci/flink/builds/131454171) * 730c3ff9eef5f41f917095eca2f0f5d1e90aea53 : PENDING [Build](https://travis-ci.com/flink-ci/flink/builds/132974433) Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #9542: [FLINK-13873][metrics] Change the column family as tags for influxdb …
flinkbot edited a comment on issue #9542: [FLINK-13873][metrics] Change the column family as tags for influxdb … URL: https://github.com/apache/flink/pull/9542#issuecomment-525276537 ## CI report: * e8636926351f3d406962dcadba275e20e49aff39 : SUCCESS [Build](https://travis-ci.com/flink-ci/flink/builds/124736151) * d23ea97e8419bbacea0698b3ba82a459d940cf38 : CANCELED [Build](https://travis-ci.com/flink-ci/flink/builds/128606376) * 05977cd49eb306d768668be4e8cb31034343df02 : FAILURE [Build](https://travis-ci.com/flink-ci/flink/builds/128606947) * 057d453e5e00656bcfcb87d1d69172f614b2d11f : CANCELED [Build](https://travis-ci.com/flink-ci/flink/builds/128681895) * 5001042b5fc5202da14c06e2d21faf1427f50a66 : FAILURE [Build](https://travis-ci.com/flink-ci/flink/builds/128683546) * 3a1e39e889daf0fca82c54982911ded35df6cb77 : SUCCESS [Build](https://travis-ci.com/flink-ci/flink/builds/128692102) * 2745a3cb02e601a1a26360e9d6b3f0af5428c66a : SUCCESS [Build](https://travis-ci.com/flink-ci/flink/builds/129027621) * e108aa24e739d6f48f63efc560c3fc049b509860 : FAILURE [Build](https://travis-ci.com/flink-ci/flink/builds/131172797) * 24a270d468dea677379713d5cf402ea453d9f222 : FAILURE [Build](https://travis-ci.com/flink-ci/flink/builds/131246042) * acf1c2b9add8c3b903a8485ed41c9f0b18d97729 : SUCCESS [Build](https://travis-ci.com/flink-ci/flink/builds/131260202) * 2f5f28b6bc2c24e81b330a3ad65cde2a23a1af95 : FAILURE [Build](https://travis-ci.com/flink-ci/flink/builds/132564124) * 1d5388df33162240b94852f742ce6d7d7a468be4 : PENDING [Build](https://travis-ci.com/flink-ci/flink/builds/132974400) Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #9255: [FLINK-13034] Introduce isEmpty method for MapState
flinkbot edited a comment on issue #9255: [FLINK-13034] Introduce isEmpty method for MapState URL: https://github.com/apache/flink/pull/9255#issuecomment-515870480 ## CI report: * de863750b7262d091d2e7dd82767b8f9b2e2c52a : SUCCESS [Build](https://travis-ci.com/flink-ci/flink/builds/121015859) * ca4f7d21ca127be3710c8d2b249bf29bef8d2d5d : FAILURE [Build](https://travis-ci.com/flink-ci/flink/builds/129050799) * 7a533f9e7472c73bbef42da35599df295c4f7666 : FAILURE [Build](https://travis-ci.com/flink-ci/flink/builds/132351883) * 53539a3865d910c65e07430bc5bc9f3bbe8e9074 : FAILURE [Build](https://travis-ci.com/flink-ci/flink/builds/132966708) Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] liuyongvs commented on issue #9966: [FLINK-14053] [blink-planner] DenseRankAggFunction.accumulateExpressions. it should be thinki…
liuyongvs commented on issue #9966: [FLINK-14053] [blink-planner] DenseRankAggFunction.accumulateExpressions. it should be thinki… URL: https://github.com/apache/flink/pull/9966#issuecomment-544906868 > Your base branch seems to be wrong, please use latest master branch and create a fix based on that. Thanks, i found the conflict when i commited because of my personal respository, which it not the latest. This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] liuyongvs commented on a change in pull request #9966: [FLINK-14053] [blink-planner] DenseRankAggFunction.accumulateExpressions. it should be thinki…
liuyongvs commented on a change in pull request #9966: [FLINK-14053]
[blink-planner] DenseRankAggFunction.accumulateExpressions. it should be thinki…
URL: https://github.com/apache/flink/pull/9966#discussion_r337447303
##
File path:
flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/functions/aggfunctions/DenseRankAggFunction.java
##
@@ -67,8 +65,8 @@ public DenseRankAggFunction(InternalType[] orderKeyTypes) {
@Override
public Expression[] accumulateExpressions() {
Expression[] accExpressions = new Expression[1 +
operands().length];
- // sequence = if (lastValues equalTo orderKeys) sequence else
sequence + 1
- accExpressions[0] = ifThenElse(orderKeyEqualsExpression(),
sequence, plus(sequence, literal(1L)));
+ // sequence = if (lastValues equalTo orderKeys and sequence !=
0) sequence else sequence + 1
+ accExpressions[0] = ifThenElse(and(orderKeyEqualsExpression(),
not(equalTo(sequence, literal(0L, sequence, plus(sequence, literal(1L)));
Review comment:
you scan see the unit test i commit. it will be zero rank
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] liuyongvs commented on a change in pull request #9966: [FLINK-14053] [blink-planner] DenseRankAggFunction.accumulateExpressions. it should be thinki…
liuyongvs commented on a change in pull request #9966: [FLINK-14053] [blink-planner] DenseRankAggFunction.accumulateExpressions. it should be thinki… URL: https://github.com/apache/flink/pull/9966#discussion_r337446771 ## File path: flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/functions/aggfunctions/DenseRankAggFunction.java ## @@ -23,9 +23,7 @@ import org.apache.flink.table.type.InternalType; import org.apache.flink.table.type.InternalTypes; -import static org.apache.flink.table.expressions.ExpressionBuilder.ifThenElse; -import static org.apache.flink.table.expressions.ExpressionBuilder.literal; -import static org.apache.flink.table.expressions.ExpressionBuilder.plus; +import static org.apache.flink.table.expressions.ExpressionBuilder.*; Review comment: got it This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #9964: test Travis arm ci
flinkbot edited a comment on issue #9964: test Travis arm ci URL: https://github.com/apache/flink/pull/9964#issuecomment-544799364 ## CI report: * 754182e6b5101abec8af40e3465a9d43c685d63c : FAILURE [Build](https://travis-ci.com/flink-ci/flink/builds/132933631) * ea99f7d1b0cc855256cb6da75c2f88069ca46bde : FAILURE [Build](https://travis-ci.com/flink-ci/flink/builds/132941984) * 0d38d29dd05ebd7bb47b749461f07f4bf49392ae : UNKNOWN * 6c5f2484a87ce787885467ef9cbaface77be1857 : CANCELED [Build](https://travis-ci.com/flink-ci/flink/builds/132944338) * f4569df2d248991a31658aa3d3f1ed5b175a47af : FAILURE [Build](https://travis-ci.com/flink-ci/flink/builds/132954158) Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #9967: [FLINK-14453][runtime] Support building pipelined regions from base topology
flinkbot edited a comment on issue #9967: [FLINK-14453][runtime] Support building pipelined regions from base topology URL: https://github.com/apache/flink/pull/9967#issuecomment-544879098 ## CI report: * cdf5f2c32d7ac87b172bc42cb868865e6da2e537 : SUCCESS [Build](https://travis-ci.com/flink-ci/flink/builds/132962326) Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #9968: [hotfix][typo] fix typo in YarnClusterDescriptor
flinkbot edited a comment on issue #9968: [hotfix][typo] fix typo in YarnClusterDescriptor URL: https://github.com/apache/flink/pull/9968#issuecomment-544895954 ## CI report: * 76cbd6febfa330bfecab18176279e059241443b7 : PENDING [Build](https://travis-ci.com/flink-ci/flink/builds/132970317) Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #9965: [FLINK-10935][kubernetes]Implement KubeClient with Faric8 Kubernetes clients
flinkbot edited a comment on issue #9965: [FLINK-10935][kubernetes]Implement KubeClient with Faric8 Kubernetes clients URL: https://github.com/apache/flink/pull/9965#issuecomment-544813931 ## CI report: * 6f90b457e56a0a8cb45d63c1b05b47d2e38030a1 : SUCCESS [Build](https://travis-ci.com/flink-ci/flink/builds/132938440) * 86aa5ce8f77faf233c51a7231b3f71e518fd6c92 : SUCCESS [Build](https://travis-ci.com/flink-ci/flink/builds/132962300) Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] wangxiyuan closed pull request #9964: test Travis arm ci
wangxiyuan closed pull request #9964: test Travis arm ci URL: https://github.com/apache/flink/pull/9964 This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #9950: [FLINK-14464][runtime] Introduce the AbstractUserClassPathJobGraphRetriever
flinkbot edited a comment on issue #9950: [FLINK-14464][runtime] Introduce the AbstractUserClassPathJobGraphRetriever URL: https://github.com/apache/flink/pull/9950#issuecomment-544414033 ## CI report: * decba8623c001e8bbe5dc797de2bc421b2e216ef : SUCCESS [Build](https://travis-ci.com/flink-ci/flink/builds/132777340) * 6c7ffb9f9049016ce874b9790cd51ed337cfec00 : UNKNOWN Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[jira] [Commented] (FLINK-9953) Active Kubernetes integration
[
https://issues.apache.org/jira/browse/FLINK-9953?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16956925#comment-16956925
]
Yang Wang commented on FLINK-9953:
--
It is reasonable for me that we only support session mode in the first MVP. And
i have almost completed the basic implementation. I will appreciate it if
[~felixzheng] or someone from the community could help to review the PRs.
[https://github.com/apache/flink/pull/9957]
[https://github.com/apache/flink/pull/9965]
> Active Kubernetes integration
> -
>
> Key: FLINK-9953
> URL: https://issues.apache.org/jira/browse/FLINK-9953
> Project: Flink
> Issue Type: New Feature
> Components: Runtime / Coordination
>Reporter: Till Rohrmann
>Assignee: Yang Wang
>Priority: Major
> Labels: pull-request-available
> Time Spent: 10m
> Remaining Estimate: 0h
>
> This is the umbrella issue tracking Flink's active Kubernetes integration.
> Active means in this context that the {{ResourceManager}} can talk to
> Kubernetes to launch new pods similar to Flink's Yarn and Mesos integration.
> Phase1 implementation will have complete functions to make flink running on
> kubernetes. Phrase2 is mainly focused on production optimization, including
> k8s native high-availability, storage, network, log collector and etc.
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[GitHub] [flink] tillrohrmann commented on a change in pull request #9928: [FLINK-12122] Add support for spreading slots out across all TaskExecutors
tillrohrmann commented on a change in pull request #9928: [FLINK-12122] Add
support for spreading slots out across all TaskExecutors
URL: https://github.com/apache/flink/pull/9928#discussion_r337441021
##
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/LeastUtilizationSlotMatchingStrategy.java
##
@@ -0,0 +1,64 @@
+/*
+ * 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.runtime.clusterframework.types.ResourceProfile;
+import org.apache.flink.runtime.instance.InstanceID;
+
+import java.util.Collection;
+import java.util.Comparator;
+import java.util.Map;
+import java.util.Optional;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+/**
+ * {@link SlotMatchingStrategy} which picks a matching slot from a TaskExecutor
+ * with the least utilization.
+ */
+public enum LeastUtilizationSlotMatchingStrategy implements
SlotMatchingStrategy {
+ INSTANCE;
+
+ @Override
+ public Optional
findMatchingSlot(
+ ResourceProfile requestedProfile,
+ Collection freeSlots,
+ Function
numberRegisteredSlotsLookup) {
+ final Map numSlotsPerTaskExecutor =
freeSlots.stream()
+ .collect(Collectors.groupingBy(
+ TaskManagerSlotInformation::getInstanceId,
+ Collectors.reducing(0, i -> 1, Integer::sum)));
+
+ return freeSlots.stream()
+ .filter(taskManagerSlot ->
taskManagerSlot.isMatchingRequirement(requestedProfile))
+ .min(Comparator.comparingDouble(taskManagerSlot ->
calculateUtilization(taskManagerSlot.getInstanceId(),
numberRegisteredSlotsLookup, numSlotsPerTaskExecutor)));
+ }
+
+ private static double calculateUtilization(InstanceID instanceId,
Function numberRegisteredSlotsLookup,
Map numSlotsPerTaskExecutor) {
+ final int numberRegisteredSlots =
numberRegisteredSlotsLookup.apply(instanceId);
+
+ if (numberRegisteredSlots == 0) {
+ return Double.MAX_VALUE;
+ }
Review comment:
True. Will update the PR.
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #9832: [FLINK-11843] Bind lifespan of Dispatcher to leader session
flinkbot edited a comment on issue #9832: [FLINK-11843] Bind lifespan of Dispatcher to leader session URL: https://github.com/apache/flink/pull/9832#issuecomment-537049332 ## CI report: * abaae048fef753455970fac9d6ab421b660b0536 : UNKNOWN * b96c63552ccd322adae7a41a410615e95b538ece : SUCCESS [Build](https://travis-ci.com/flink-ci/flink/builds/129867595) * 2c95a3939dbf0259d694af6c69451f0ede3c3891 : SUCCESS [Build](https://travis-ci.com/flink-ci/flink/builds/130066310) * 6e030add922011ea54178690f171911d0139f14b : SUCCESS [Build](https://travis-ci.com/flink-ci/flink/builds/130858235) * 271703eda6f6c55b1641a54206109ef659f62854 : SUCCESS [Build](https://travis-ci.com/flink-ci/flink/builds/131454171) * 730c3ff9eef5f41f917095eca2f0f5d1e90aea53 : UNKNOWN Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] tillrohrmann commented on a change in pull request #9928: [FLINK-12122] Add support for spreading slots out across all TaskExecutors
tillrohrmann commented on a change in pull request #9928: [FLINK-12122] Add
support for spreading slots out across all TaskExecutors
URL: https://github.com/apache/flink/pull/9928#discussion_r337440872
##
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/LeastUtilizationSlotMatchingStrategy.java
##
@@ -0,0 +1,64 @@
+/*
+ * 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.runtime.clusterframework.types.ResourceProfile;
+import org.apache.flink.runtime.instance.InstanceID;
+
+import java.util.Collection;
+import java.util.Comparator;
+import java.util.Map;
+import java.util.Optional;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+/**
+ * {@link SlotMatchingStrategy} which picks a matching slot from a TaskExecutor
+ * with the least utilization.
+ */
+public enum LeastUtilizationSlotMatchingStrategy implements
SlotMatchingStrategy {
+ INSTANCE;
+
+ @Override
+ public Optional
findMatchingSlot(
+ ResourceProfile requestedProfile,
+ Collection freeSlots,
+ Function
numberRegisteredSlotsLookup) {
+ final Map numSlotsPerTaskExecutor =
freeSlots.stream()
+ .collect(Collectors.groupingBy(
+ TaskManagerSlotInformation::getInstanceId,
+ Collectors.reducing(0, i -> 1, Integer::sum)));
+
+ return freeSlots.stream()
+ .filter(taskManagerSlot ->
taskManagerSlot.isMatchingRequirement(requestedProfile))
+ .min(Comparator.comparingDouble(taskManagerSlot ->
calculateUtilization(taskManagerSlot.getInstanceId(),
numberRegisteredSlotsLookup, numSlotsPerTaskExecutor)));
Review comment:
I would leave this as a future optimization if we see that this becomes a
problem.
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] MalcolmSanders commented on a change in pull request #9965: [FLINK-10935][kubernetes]Implement KubeClient with Faric8 Kubernetes clients
MalcolmSanders commented on a change in pull request #9965:
[FLINK-10935][kubernetes]Implement KubeClient with Faric8 Kubernetes clients
URL: https://github.com/apache/flink/pull/9965#discussion_r337433178
##
File path:
flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/decorators/FlinkMasterDeploymentDecorator.java
##
@@ -0,0 +1,180 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.kubernetes.kubeclient.decorators;
+
+import org.apache.flink.client.cli.CliFrontend;
+import org.apache.flink.client.deployment.ClusterSpecification;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.JobManagerOptions;
+import org.apache.flink.configuration.RestOptions;
+import org.apache.flink.kubernetes.configuration.KubernetesConfigOptions;
+import org.apache.flink.kubernetes.kubeclient.resources.FlinkDeployment;
+import org.apache.flink.kubernetes.utils.Constants;
+import org.apache.flink.kubernetes.utils.KubernetesUtils;
+import org.apache.flink.util.Preconditions;
+
+import io.fabric8.kubernetes.api.model.ConfigMapVolumeSourceBuilder;
+import io.fabric8.kubernetes.api.model.Container;
+import io.fabric8.kubernetes.api.model.ContainerBuilder;
+import io.fabric8.kubernetes.api.model.ContainerPortBuilder;
+import io.fabric8.kubernetes.api.model.KeyToPath;
+import io.fabric8.kubernetes.api.model.PodSpec;
+import io.fabric8.kubernetes.api.model.PodSpecBuilder;
+import io.fabric8.kubernetes.api.model.Quantity;
+import io.fabric8.kubernetes.api.model.ResourceRequirementsBuilder;
+import io.fabric8.kubernetes.api.model.Volume;
+import io.fabric8.kubernetes.api.model.VolumeMountBuilder;
+import io.fabric8.kubernetes.api.model.apps.Deployment;
+import io.fabric8.kubernetes.api.model.apps.DeploymentSpecBuilder;
+
+import java.io.File;
+import java.util.Arrays;
+import java.util.Map;
+
+import static
org.apache.flink.configuration.GlobalConfiguration.FLINK_CONF_FILENAME;
+import static org.apache.flink.kubernetes.utils.Constants.BLOB_SERVER_PORT;
+import static
org.apache.flink.kubernetes.utils.Constants.CONFIG_FILE_LOG4J_NAME;
+import static
org.apache.flink.kubernetes.utils.Constants.CONFIG_FILE_LOGBACK_NAME;
+import static org.apache.flink.kubernetes.utils.Constants.CONFIG_MAP_PREFIX;
+import static org.apache.flink.kubernetes.utils.Constants.FLINK_CONF_VOLUME;
+
+/**
+ * Flink master specific deployment configuration.
+ * */
+public class FlinkMasterDeploymentDecorator extends Decorator {
+
+ private static final String CONTAINER_NAME = "flink-job-manager";
+
+ private final ClusterSpecification clusterSpecification;
+
+ public static final ConfigOption ENTRY_POINT_CLASS =
ConfigOptions
+ .key("kubernetes.internal.jobmanager.entrypoint.class")
+ .noDefaultValue()
+ .withDescription("The entrypoint class for jobmanager. It will
be set in kubernetesClusterDescriptor.");
+
+ public static final ConfigOption ENTRY_POINT_CLASS_ARGS =
ConfigOptions
+ .key("kubernetes.internal.jobmanager.entrypoint.class.args")
+ .noDefaultValue()
+ .withDescription("The args of entrypoint class for jobmanager.
It will be set in FlinkKubernetesCustomCli.");
+
+ public FlinkMasterDeploymentDecorator(ClusterSpecification
clusterSpecification) {
+ this.clusterSpecification = clusterSpecification;
+ }
+
+ @Override
+ protected Deployment doDecorate(Deployment deployment, Configuration
flinkConfig) {
+ String clusterId =
flinkConfig.getString(KubernetesConfigOptions.CLUSTER_ID);
+ Preconditions.checkNotNull(clusterId, "ClusterId must be
specified!");
+
+ String mainClass = flinkConfig.getString(ENTRY_POINT_CLASS);
+ Preconditions.checkNotNull(mainClass, "Main class must be
specified!");
+
+ Map labels = LabelBuilder
+ .withExist(deployment.getMetadata().getLabels())
+ .withJobManagerComponent()
+
[GitHub] [flink] flinkbot edited a comment on issue #9542: [FLINK-13873][metrics] Change the column family as tags for influxdb …
flinkbot edited a comment on issue #9542: [FLINK-13873][metrics] Change the column family as tags for influxdb … URL: https://github.com/apache/flink/pull/9542#issuecomment-525276537 ## CI report: * e8636926351f3d406962dcadba275e20e49aff39 : SUCCESS [Build](https://travis-ci.com/flink-ci/flink/builds/124736151) * d23ea97e8419bbacea0698b3ba82a459d940cf38 : CANCELED [Build](https://travis-ci.com/flink-ci/flink/builds/128606376) * 05977cd49eb306d768668be4e8cb31034343df02 : FAILURE [Build](https://travis-ci.com/flink-ci/flink/builds/128606947) * 057d453e5e00656bcfcb87d1d69172f614b2d11f : CANCELED [Build](https://travis-ci.com/flink-ci/flink/builds/128681895) * 5001042b5fc5202da14c06e2d21faf1427f50a66 : FAILURE [Build](https://travis-ci.com/flink-ci/flink/builds/128683546) * 3a1e39e889daf0fca82c54982911ded35df6cb77 : SUCCESS [Build](https://travis-ci.com/flink-ci/flink/builds/128692102) * 2745a3cb02e601a1a26360e9d6b3f0af5428c66a : SUCCESS [Build](https://travis-ci.com/flink-ci/flink/builds/129027621) * e108aa24e739d6f48f63efc560c3fc049b509860 : FAILURE [Build](https://travis-ci.com/flink-ci/flink/builds/131172797) * 24a270d468dea677379713d5cf402ea453d9f222 : FAILURE [Build](https://travis-ci.com/flink-ci/flink/builds/131246042) * acf1c2b9add8c3b903a8485ed41c9f0b18d97729 : SUCCESS [Build](https://travis-ci.com/flink-ci/flink/builds/131260202) * 2f5f28b6bc2c24e81b330a3ad65cde2a23a1af95 : FAILURE [Build](https://travis-ci.com/flink-ci/flink/builds/132564124) * 1d5388df33162240b94852f742ce6d7d7a468be4 : UNKNOWN Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] MalcolmSanders commented on a change in pull request #9965: [FLINK-10935][kubernetes]Implement KubeClient with Faric8 Kubernetes clients
MalcolmSanders commented on a change in pull request #9965: [FLINK-10935][kubernetes]Implement KubeClient with Faric8 Kubernetes clients URL: https://github.com/apache/flink/pull/9965#discussion_r337362157 ## File path: flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/fabric8/FlinkConfigMap.java ## @@ -0,0 +1,34 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.kubernetes.kubeclient.fabric8; Review comment: what's the purpose of add "fabric8" package ? This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] MalcolmSanders commented on a change in pull request #9965: [FLINK-10935][kubernetes]Implement KubeClient with Faric8 Kubernetes clients
MalcolmSanders commented on a change in pull request #9965:
[FLINK-10935][kubernetes]Implement KubeClient with Faric8 Kubernetes clients
URL: https://github.com/apache/flink/pull/9965#discussion_r337436697
##
File path:
flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/decorators/TaskManagerPodDecorator.java
##
@@ -0,0 +1,136 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.kubernetes.kubeclient.decorators;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.kubernetes.configuration.KubernetesConfigOptions;
+import org.apache.flink.kubernetes.kubeclient.TaskManagerPodParameter;
+import org.apache.flink.kubernetes.kubeclient.resources.FlinkPod;
+import org.apache.flink.kubernetes.utils.Constants;
+import org.apache.flink.util.Preconditions;
+
+import io.fabric8.kubernetes.api.model.ConfigMapVolumeSourceBuilder;
+import io.fabric8.kubernetes.api.model.Container;
+import io.fabric8.kubernetes.api.model.ContainerBuilder;
+import io.fabric8.kubernetes.api.model.ContainerPortBuilder;
+import io.fabric8.kubernetes.api.model.EnvVar;
+import io.fabric8.kubernetes.api.model.KeyToPath;
+import io.fabric8.kubernetes.api.model.Pod;
+import io.fabric8.kubernetes.api.model.PodSpecBuilder;
+import io.fabric8.kubernetes.api.model.Quantity;
+import io.fabric8.kubernetes.api.model.ResourceRequirementsBuilder;
+import io.fabric8.kubernetes.api.model.Volume;
+import io.fabric8.kubernetes.api.model.VolumeMountBuilder;
+
+import java.io.File;
+import java.util.Arrays;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import static
org.apache.flink.configuration.GlobalConfiguration.FLINK_CONF_FILENAME;
+import static
org.apache.flink.kubernetes.utils.Constants.CONFIG_FILE_LOG4J_NAME;
+import static
org.apache.flink.kubernetes.utils.Constants.CONFIG_FILE_LOGBACK_NAME;
+import static org.apache.flink.kubernetes.utils.Constants.CONFIG_MAP_PREFIX;
+import static org.apache.flink.kubernetes.utils.Constants.FLINK_CONF_VOLUME;
+
+/**
+ * Task manager specific pod configuration.
+ * */
+public class TaskManagerPodDecorator extends Decorator {
+
+ private static final String CONTAINER_NAME = "flink-task-manager";
+
+ private final TaskManagerPodParameter parameter;
+
+ public TaskManagerPodDecorator(TaskManagerPodParameter parameters) {
+ Preconditions.checkNotNull(parameters);
+ this.parameter = parameters;
+ }
+
+ @Override
+ protected Pod doDecorate(Pod pod, Configuration flinkConfig) {
+
+ String clusterId =
flinkConfig.getString(KubernetesConfigOptions.CLUSTER_ID);
+ Preconditions.checkNotNull(clusterId, "ClusterId must be
specified!");
+
+ Map labels =
org.apache.flink.kubernetes.kubeclient.decorators.LabelBuilder
+ .withExist(pod.getMetadata().getLabels())
+ .withTaskManagerComponent()
+ .toLabels();
+
+ pod.getMetadata().setLabels(labels);
+ pod.getMetadata().setName(this.parameter.getPodName());
+
+ Volume configMapVolume = new Volume();
+ configMapVolume.setName(FLINK_CONF_VOLUME);
+ configMapVolume.setConfigMap(new ConfigMapVolumeSourceBuilder()
+ .withName(CONFIG_MAP_PREFIX + clusterId)
+ .withItems(Arrays.asList(
+ new KeyToPath(FLINK_CONF_FILENAME, null,
FLINK_CONF_FILENAME),
+ new KeyToPath(CONFIG_FILE_LOG4J_NAME, null,
CONFIG_FILE_LOG4J_NAME),
+ new KeyToPath(CONFIG_FILE_LOGBACK_NAME, null,
CONFIG_FILE_LOGBACK_NAME)))
+ .build());
+
+ pod.setSpec(new PodSpecBuilder()
+ .withVolumes(configMapVolume)
+ .withContainers(createTaskManagerContainer(flinkConfig))
+ .build());
+ return pod;
+ }
+
+ private Container createTaskManagerContainer(Configuration flinkConfig)
{
+ Quantity taskManagerCpuQuantity = new
[GitHub] [flink] MalcolmSanders commented on a change in pull request #9965: [FLINK-10935][kubernetes]Implement KubeClient with Faric8 Kubernetes clients
MalcolmSanders commented on a change in pull request #9965:
[FLINK-10935][kubernetes]Implement KubeClient with Faric8 Kubernetes clients
URL: https://github.com/apache/flink/pull/9965#discussion_r337428992
##
File path: flink-kubernetes/pom.xml
##
@@ -0,0 +1,208 @@
+
+
+
+http://maven.apache.org/POM/4.0.0;
+xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance;
+xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
http://maven.apache.org/xsd/maven-4.0.0.xsd;>
+ 4.0.0
+
+ org.apache.flink
+ flink-parent
+ 1.10-SNAPSHOT
+ ..
+
+
+ flink-kubernetes_${scala.binary.version}
+ flink-kubernetes
+ jar
+
+
+ 4.5.2
+
Review comment:
move this property to root/pom.xm just like hadoop.version ?
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] MalcolmSanders commented on a change in pull request #9965: [FLINK-10935][kubernetes]Implement KubeClient with Faric8 Kubernetes clients
MalcolmSanders commented on a change in pull request #9965:
[FLINK-10935][kubernetes]Implement KubeClient with Faric8 Kubernetes clients
URL: https://github.com/apache/flink/pull/9965#discussion_r337435108
##
File path:
flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/decorators/TaskManagerPodDecorator.java
##
@@ -0,0 +1,136 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.kubernetes.kubeclient.decorators;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.kubernetes.configuration.KubernetesConfigOptions;
+import org.apache.flink.kubernetes.kubeclient.TaskManagerPodParameter;
+import org.apache.flink.kubernetes.kubeclient.resources.FlinkPod;
+import org.apache.flink.kubernetes.utils.Constants;
+import org.apache.flink.util.Preconditions;
+
+import io.fabric8.kubernetes.api.model.ConfigMapVolumeSourceBuilder;
+import io.fabric8.kubernetes.api.model.Container;
+import io.fabric8.kubernetes.api.model.ContainerBuilder;
+import io.fabric8.kubernetes.api.model.ContainerPortBuilder;
+import io.fabric8.kubernetes.api.model.EnvVar;
+import io.fabric8.kubernetes.api.model.KeyToPath;
+import io.fabric8.kubernetes.api.model.Pod;
+import io.fabric8.kubernetes.api.model.PodSpecBuilder;
+import io.fabric8.kubernetes.api.model.Quantity;
+import io.fabric8.kubernetes.api.model.ResourceRequirementsBuilder;
+import io.fabric8.kubernetes.api.model.Volume;
+import io.fabric8.kubernetes.api.model.VolumeMountBuilder;
+
+import java.io.File;
+import java.util.Arrays;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import static
org.apache.flink.configuration.GlobalConfiguration.FLINK_CONF_FILENAME;
+import static
org.apache.flink.kubernetes.utils.Constants.CONFIG_FILE_LOG4J_NAME;
+import static
org.apache.flink.kubernetes.utils.Constants.CONFIG_FILE_LOGBACK_NAME;
+import static org.apache.flink.kubernetes.utils.Constants.CONFIG_MAP_PREFIX;
+import static org.apache.flink.kubernetes.utils.Constants.FLINK_CONF_VOLUME;
+
+/**
+ * Task manager specific pod configuration.
+ * */
+public class TaskManagerPodDecorator extends Decorator {
+
+ private static final String CONTAINER_NAME = "flink-task-manager";
+
+ private final TaskManagerPodParameter parameter;
+
+ public TaskManagerPodDecorator(TaskManagerPodParameter parameters) {
+ Preconditions.checkNotNull(parameters);
+ this.parameter = parameters;
+ }
+
+ @Override
+ protected Pod doDecorate(Pod pod, Configuration flinkConfig) {
+
+ String clusterId =
flinkConfig.getString(KubernetesConfigOptions.CLUSTER_ID);
+ Preconditions.checkNotNull(clusterId, "ClusterId must be
specified!");
+
+ Map labels =
org.apache.flink.kubernetes.kubeclient.decorators.LabelBuilder
+ .withExist(pod.getMetadata().getLabels())
+ .withTaskManagerComponent()
+ .toLabels();
+
+ pod.getMetadata().setLabels(labels);
+ pod.getMetadata().setName(this.parameter.getPodName());
+
+ Volume configMapVolume = new Volume();
+ configMapVolume.setName(FLINK_CONF_VOLUME);
+ configMapVolume.setConfigMap(new ConfigMapVolumeSourceBuilder()
+ .withName(CONFIG_MAP_PREFIX + clusterId)
+ .withItems(Arrays.asList(
+ new KeyToPath(FLINK_CONF_FILENAME, null,
FLINK_CONF_FILENAME),
+ new KeyToPath(CONFIG_FILE_LOG4J_NAME, null,
CONFIG_FILE_LOG4J_NAME),
+ new KeyToPath(CONFIG_FILE_LOGBACK_NAME, null,
CONFIG_FILE_LOGBACK_NAME)))
+ .build());
Review comment:
seems duplicated code reference to FlinkMasterDeploymentDecorator
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
[GitHub] [flink] MalcolmSanders commented on a change in pull request #9965: [FLINK-10935][kubernetes]Implement KubeClient with Faric8 Kubernetes clients
MalcolmSanders commented on a change in pull request #9965:
[FLINK-10935][kubernetes]Implement KubeClient with Faric8 Kubernetes clients
URL: https://github.com/apache/flink/pull/9965#discussion_r337430586
##
File path:
flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/Fabric8FlinkKubeClient.java
##
@@ -0,0 +1,325 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.kubernetes.kubeclient;
+
+import org.apache.flink.client.deployment.ClusterSpecification;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.RestOptions;
+import org.apache.flink.kubernetes.configuration.KubernetesConfigOptions;
+import org.apache.flink.kubernetes.kubeclient.decorators.ConfigMapDecorator;
+import org.apache.flink.kubernetes.kubeclient.decorators.Decorator;
+import
org.apache.flink.kubernetes.kubeclient.decorators.FlinkMasterDeploymentDecorator;
+import org.apache.flink.kubernetes.kubeclient.decorators.InitializerDecorator;
+import
org.apache.flink.kubernetes.kubeclient.decorators.OwnerReferenceDecorator;
+import org.apache.flink.kubernetes.kubeclient.decorators.ServiceDecorator;
+import
org.apache.flink.kubernetes.kubeclient.decorators.TaskManagerPodDecorator;
+import org.apache.flink.kubernetes.kubeclient.resources.ActionWatcher;
+import org.apache.flink.kubernetes.kubeclient.resources.FlinkConfigMap;
+import org.apache.flink.kubernetes.kubeclient.resources.FlinkDeployment;
+import org.apache.flink.kubernetes.kubeclient.resources.FlinkPod;
+import org.apache.flink.kubernetes.kubeclient.resources.FlinkService;
+import org.apache.flink.kubernetes.utils.Constants;
+
+import io.fabric8.kubernetes.api.model.ConfigMap;
+import io.fabric8.kubernetes.api.model.Pod;
+import io.fabric8.kubernetes.api.model.Service;
+import io.fabric8.kubernetes.api.model.ServicePort;
+import io.fabric8.kubernetes.api.model.apps.Deployment;
+import io.fabric8.kubernetes.client.KubernetesClient;
+import io.fabric8.kubernetes.client.KubernetesClientException;
+import io.fabric8.kubernetes.client.Watch;
+import io.fabric8.kubernetes.client.Watcher;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * The implementation of {@link FlinkKubeClient}.
+ * */
+public class Fabric8FlinkKubeClient implements FlinkKubeClient {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(Fabric8FlinkKubeClient.class);
+
+ private final Configuration flinkConfig;
+
+ private final KubernetesClient internalClient;
+
+ private List> configMapDecorators;
+
+ private List>
internalServiceDecorators;
+ private List> restServiceDecorators;
+
+ private List>
flinkMasterDeploymentDecorators;
+
+ private List> taskManagerPodDecorators;
+
+ @Nonnull
+ private final String clusterId;
+
+ @Nonnull
+ private final String nameSpace;
+
+ public Fabric8FlinkKubeClient(Configuration flinkConfig,
KubernetesClient client) {
+ this.internalClient = client;
+ this.flinkConfig = flinkConfig;
+ this.configMapDecorators = new ArrayList<>();
+ this.internalServiceDecorators = new ArrayList<>();
+ this.restServiceDecorators = new ArrayList<>();
+ this.flinkMasterDeploymentDecorators = new ArrayList<>();
+ this.taskManagerPodDecorators = new ArrayList<>();
+
+ this.clusterId =
flinkConfig.getString(KubernetesConfigOptions.CLUSTER_ID);
+ this.nameSpace =
flinkConfig.getString(KubernetesConfigOptions.NAME_SPACE);
+
+ initialize();
+ }
+
+ private void initialize() {
+ this.configMapDecorators.add(new
InitializerDecorator<>(Constants.CONFIG_MAP_PREFIX + clusterId));
+ this.configMapDecorators.add(new OwnerReferenceDecorator<>());
[GitHub] [flink] MalcolmSanders commented on a change in pull request #9965: [FLINK-10935][kubernetes]Implement KubeClient with Faric8 Kubernetes clients
MalcolmSanders commented on a change in pull request #9965:
[FLINK-10935][kubernetes]Implement KubeClient with Faric8 Kubernetes clients
URL: https://github.com/apache/flink/pull/9965#discussion_r337429424
##
File path:
flink-kubernetes/src/main/java/org/apache/flink/kubernetes/configuration/KubernetesConfigOptions.java
##
@@ -0,0 +1,115 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.kubernetes.configuration;
+
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.kubernetes.cli.KubernetesCliOptions;
+
+import static org.apache.flink.configuration.ConfigOptions.key;
+
+/**
+ * This class holds configuration constants used by Flink's kubernetes runners.
+ */
+public class KubernetesConfigOptions {
+
+ public static final ConfigOption REST_SERVICE_EXPOSED_TYPE =
+ key("kubernetes.rest-service.exposed.type")
+ .defaultValue(ServiceExposedType.LoadBalancer.toString())
+ .withDescription("It could be
ClusterIP/NodePort/LoadBalancer(default). When set to ClusterIP, the rest
service" +
+ "will not be created.");
+
+ public static final ConfigOption JOB_MANAGER_SERVICE_ACCOUNT =
+ key("kubernetes.jobmanager.service-account")
+ .defaultValue("default")
+ .withDescription("Service account that is used by jobmanager
within kubernetes cluster. " +
+ "The job manager uses this service account when
requesting taskmanager pods from the API server.");
+
+ public static final ConfigOption JOB_MANAGER_CPU =
+ key("kubernetes.jobmanager.cpu")
+ .defaultValue(1.0)
+ .withDescription("The number of cpu used by job manager");
+
+ public static final ConfigOption TASK_MANAGER_CPU =
+ key("kubernetes.taskmanager.cpu")
+ .defaultValue(1.0)
+ .withDescription("The number of cpu used by task manager");
+
+ public static final ConfigOption CONTAINER_IMAGE_PULL_POLICY =
+ key("kubernetes.container.image.pullPolicy")
+ .defaultValue("Always")
+ .withDescription("Kubernetes image pull policy. Valid values
are Always, Never, and IfNotPresent.");
+
+ public static final ConfigOption KUBE_CONFIG_FILE =
+ key("kubernetes.config.file")
+ .noDefaultValue()
+ .withDescription("The kubernetes config file will be used to
create the client. The default " +
+ "is located at ~/.kube/config");
+
+ public static final ConfigOption NAME_SPACE =
Review comment:
namespace is a word, maybe we don't need to seperate this word ?
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] tillrohrmann commented on a change in pull request #9928: [FLINK-12122] Add support for spreading slots out across all TaskExecutors
tillrohrmann commented on a change in pull request #9928: [FLINK-12122] Add
support for spreading slots out across all TaskExecutors
URL: https://github.com/apache/flink/pull/9928#discussion_r337438335
##
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/SlotManagerImpl.java
##
@@ -124,12 +126,14 @@
private boolean failUnfulfillableRequest = true;
public SlotManagerImpl(
+ SlotMatchingStrategy slotMatchingStrategy,
ScheduledExecutor scheduledExecutor,
Time taskManagerRequestTimeout,
Time slotRequestTimeout,
Time taskManagerTimeout,
boolean waitResultConsumedBeforeRelease) {
+ this.slotMatchingStrategy = slotMatchingStrategy;
Review comment:
True, I will add it.
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot commented on issue #9968: [hotfix][typo] fix typo in YarnClusterDescriptor
flinkbot commented on issue #9968: [hotfix][typo] fix typo in YarnClusterDescriptor URL: https://github.com/apache/flink/pull/9968#issuecomment-544895954 ## CI report: * 76cbd6febfa330bfecab18176279e059241443b7 : UNKNOWN Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #9965: [FLINK-10935][kubernetes]Implement KubeClient with Faric8 Kubernetes clients
flinkbot edited a comment on issue #9965: [FLINK-10935][kubernetes]Implement KubeClient with Faric8 Kubernetes clients URL: https://github.com/apache/flink/pull/9965#issuecomment-544813931 ## CI report: * 6f90b457e56a0a8cb45d63c1b05b47d2e38030a1 : SUCCESS [Build](https://travis-ci.com/flink-ci/flink/builds/132938440) * 86aa5ce8f77faf233c51a7231b3f71e518fd6c92 : PENDING [Build](https://travis-ci.com/flink-ci/flink/builds/132962300) Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
