[jira] [Created] (FLINK-14067) Decouple PlanExecutor from JSON plan generation
Aljoscha Krettek created FLINK-14067: Summary: Decouple PlanExecutor from JSON plan generation Key: FLINK-14067 URL: https://issues.apache.org/jira/browse/FLINK-14067 Project: Flink Issue Type: Sub-task Components: API / DataSet, Client / Job Submission Reporter: Aljoscha Krettek Assignee: Aljoscha Krettek {{PlanExecutor}} has a method {{getOptimizerPlanAsJSON()}} that is used by DataSet environments to get a JSON version of the execution plan. To ease future work and to make it more maintainable we should get rid of that method and instead have a dedicated utility for generating JSON plans that the environments can use. (The only reason this method is on the executor is because only {{flink-clients}} via {{flink-optimizer}} has the required components to derive a JSON plan.) -- This message was sent by Atlassian Jira (v8.3.2#803003)
[jira] [Created] (FLINK-14130) Remove ClusterClient.run() methods
Aljoscha Krettek created FLINK-14130: Summary: Remove ClusterClient.run() methods Key: FLINK-14130 URL: https://issues.apache.org/jira/browse/FLINK-14130 Project: Flink Issue Type: Sub-task Components: Client / Job Submission Reporter: Aljoscha Krettek {{ClusterClient}} is an internal interface of the {{flink-clients}} package. It should only be concerned with submitting {{JobGraphs}} to a cluster, which is what {{submitJob()}} does. The {{run()}} methods are concerned with unpacking programs or job-with-jars and at the end use {{submitJob()}} in some way, they should reside in some other component. The only valid remaining run method is {{run(PackagedProgram prog, int parallelism)}}, this could be in {{PackagedProgramUtils}}. The other {{run()}} methods are actually only used in one test: {{ClientTest.shouldSubmitToJobClient()}}. I don't think that test is valid anymore, it evolved for a very long time and now doesn't test what it was supposed to test once. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Created] (FLINK-14250) Introduce separate "bin/flink runpy" and remove python support from "bin/flink run"
Aljoscha Krettek created FLINK-14250: Summary: Introduce separate "bin/flink runpy" and remove python support from "bin/flink run" Key: FLINK-14250 URL: https://issues.apache.org/jira/browse/FLINK-14250 Project: Flink Issue Type: Sub-task Components: Client / Job Submission Reporter: Aljoscha Krettek Currently, "bin/flink run" supports both Java and Python programs and there is quite some complexity in command line parsing and validation and the code is spread across different classes. I think if we had a separate "bin/flink runpy" then we could simplify the parsing quite a bit and the usage help of each command would only show those options that are relevant for the given use case. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Created] (FLINK-14290) Decouple plan translation from job execution/ClusterClient
Aljoscha Krettek created FLINK-14290: Summary: Decouple plan translation from job execution/ClusterClient Key: FLINK-14290 URL: https://issues.apache.org/jira/browse/FLINK-14290 Project: Flink Issue Type: Sub-task Components: Client / Job Submission Reporter: Aljoscha Krettek Assignee: Aljoscha Krettek -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Created] (FLINK-14391) Remove FlinkPlan Interface
Aljoscha Krettek created FLINK-14391: Summary: Remove FlinkPlan Interface Key: FLINK-14391 URL: https://issues.apache.org/jira/browse/FLINK-14391 Project: Flink Issue Type: Sub-task Components: API / DataSet, API / DataStream Reporter: Aljoscha Krettek In the process of implementing FLINK-14290 [~tison] noticed that we can remove {{FlinkPlan}} after introducing {{Pipeline}} and changing the translation logic. It was only introduced as a somewhat hacky workaround for the fact that {{OptimizedPlan}} and {{StreamGraph}} don't have a common base class/interface. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Created] (FLINK-14412) Rename ML Pipeline to MLPipeline
Aljoscha Krettek created FLINK-14412: Summary: Rename ML Pipeline to MLPipeline Key: FLINK-14412 URL: https://issues.apache.org/jira/browse/FLINK-14412 Project: Flink Issue Type: Bug Components: Library / Machine Learning Reporter: Aljoscha Krettek In FLINK-14290 we introduced a {{Pipeline}} interface in {{flink-core}} as the common interface of Flink Jobs/Pipelines. Unfortunately, this name clashes with {{Pipeline}} in the ML package. My suggestion is to rename {{Pipeline}} in the ML package to {{MLPipeline}}. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Created] (FLINK-14840) Use new Executor interface in SQL cli
Aljoscha Krettek created FLINK-14840: Summary: Use new Executor interface in SQL cli Key: FLINK-14840 URL: https://issues.apache.org/jira/browse/FLINK-14840 Project: Flink Issue Type: Sub-task Components: Table SQL / Client Reporter: Aljoscha Krettek Assignee: Aljoscha Krettek Currently, the SQL cli has custom code for job deployment in {{ProgramDeployer}}. We should replace this by using the newly introduced {{Executor}}. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Created] (FLINK-14854) Add executeAsync() method to environments
Aljoscha Krettek created FLINK-14854: Summary: Add executeAsync() method to environments Key: FLINK-14854 URL: https://issues.apache.org/jira/browse/FLINK-14854 Project: Flink Issue Type: Sub-task Components: API / DataSet, API / DataStream Reporter: Aljoscha Krettek The new {{executeAsync()}} method should return a {{JobClient}}. This exposes the new executor/job client work on the user API. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Created] (FLINK-14968) Kerberized YARN on Docker test (custom fs plugin) fails on Travis
Aljoscha Krettek created FLINK-14968: Summary: Kerberized YARN on Docker test (custom fs plugin) fails on Travis Key: FLINK-14968 URL: https://issues.apache.org/jira/browse/FLINK-14968 Project: Flink Issue Type: Bug Components: Deployment / YARN, Tests Affects Versions: 1.10.0 Reporter: Gary Yao Assignee: Aljoscha Krettek Fix For: 1.10.0 https://api.travis-ci.org/v3/job/612782888/log.txt -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Created] (FLINK-14992) Add job submission listener to execution environments
Aljoscha Krettek created FLINK-14992: Summary: Add job submission listener to execution environments Key: FLINK-14992 URL: https://issues.apache.org/jira/browse/FLINK-14992 Project: Flink Issue Type: Sub-task Components: API / DataSet, API / DataStream Reporter: Aljoscha Krettek Fix For: 1.10.0 We should add a way of registering listeners that are notified when a job is submitted for execution using an environment. This is useful for cases where a framework, for example the Zeppelin Notebook, creates an environment for the user, the user can submit jobs, but the framework needs a handle to the job in order to manage it. This can be as simple as {code} interface JobSubmissionListener { void notify(JobClient jobClient) } {code} with a method {{registerJobSubmissionListener(JobSubmissionListener)}} on the environments. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Created] (FLINK-15007) Flink on YARN does not request required TaskExecutors in some cases
Aljoscha Krettek created FLINK-15007: Summary: Flink on YARN does not request required TaskExecutors in some cases Key: FLINK-15007 URL: https://issues.apache.org/jira/browse/FLINK-15007 Project: Flink Issue Type: Bug Components: Runtime / Coordination, Runtime / Task Affects Versions: 1.10.0 Reporter: Aljoscha Krettek Fix For: 1.10.0 This was discovered while debugging FLINK-14834. In some cases Flink does not request new {{TaskExecutors}} even though new slots are requested. You can see this in some of the logs attached to FLINK-14834. You can reproduce this using https://github.com/aljoscha/docker-hadoop-cluster to bring up a YARN cluster and then running a compiled Flink in there. When you run {code} bin/flink run -m yarn-cluster -p 3 -yjm 1200 -ytm 1200 /root/DualInputWordCount.jar --input hdfs:///wc-in-1 --input hdfs:///wc-in-2 --output hdfs:///wc-out && hdfs dfs -rm -r /wc-out {code} the job waits and eventually fails because it does not have enough slots. (You can see in the log that 3 new slots are requested but only 2 {{TaskExecutors}} are requested. When you run {code} bin/flink run -m yarn-cluster -p 3 -yjm 1224 -ytm 1224 /root/DualInputWordCount.jar --input hdfs:///wc-in-1 --input hdfs:///wc-in-2 --output hdfs:///wc-out && hdfs dfs -rm -r /wc-out {code} runs successfully. This is the {{git bisect}} log that identifies the first faulty commit ([https://github.com/apache/flink/commit/9fed0ddc5bc015f98246a2d8d9adbe5fb2b91ba4|http://example.com]): {code} git bisect start # good: [09f2f43a1d73c76bf4d3f4a1205269eb860deb14] [FLINK-14154][ml] Add the class for multivariate Gaussian Distribution. git bisect good 09f2f43a1d73c76bf4d3f4a1205269eb860deb14 # bad: [85905f80e9711967711c2992612dccdd2cc211ac] [FLINK-14834][tests] Disable flaky yarn_kerberos_docker (default input) test git bisect bad 85905f80e9711967711c2992612dccdd2cc211ac # good: [c9c8a29b1b2e4f2886fba1524432f9788b564e61] [FLINK-14759][coordination] Remove unused class TaskManagerCliOptions git bisect good c9c8a29b1b2e4f2886fba1524432f9788b564e61 # good: [c9c8a29b1b2e4f2886fba1524432f9788b564e61] [FLINK-14759][coordination] Remove unused class TaskManagerCliOptions git bisect good c9c8a29b1b2e4f2886fba1524432f9788b564e61 # bad: [ae539c97c858b94e0e2504b54a8517ac1383482a] [hotfix][runtime] Check managed memory fraction range when setting it into StreamConfig git bisect bad ae539c97c858b94e0e2504b54a8517ac1383482a # good: [01d6972ab267807b8afccb09a45c454fa76d6c4b] [hotfix] Refactor out slots creation from the TaskSlotTable constructor git bisect good 01d6972ab267807b8afccb09a45c454fa76d6c4b # bad: [d32e1d00854e24bc4bb3aad6d866c2d709acd993] [FLINK-14594][core] Change Resource to use BigDecimal as its value git bisect bad d32e1d00854e24bc4bb3aad6d866c2d709acd993 # bad: [25f87ec208a642283e995811d809632129ca289a] [FLINK-11935][table-planner] Fix cast timestamp/date to string to avoid Gregorian cutover git bisect bad 25f87ec208a642283e995811d809632129ca289a # bad: [21c6b85a6f5991aabcbcd41fedc860d662d478fb] [FLINK-14842][table] add logging for loaded modules and functions git bisect bad 21c6b85a6f5991aabcbcd41fedc860d662d478fb # bad: [48986aa0b89de731d1b9136b59d409933cc15408] [hotfix] Remove unnecessary comments about memory size calculation before network init git bisect bad 48986aa0b89de731d1b9136b59d409933cc15408 # bad: [4c4652efa43ed8ab456f5f63c89b57d8c4a621f8] [hotfix] Remove unused number of slots in MemoryManager git bisect bad 4c4652efa43ed8ab456f5f63c89b57d8c4a621f8 # bad: [9fed0ddc5bc015f98246a2d8d9adbe5fb2b91ba4] [FLINK-14400] Shrink the scope of MemoryManager from TaskExecutor to slot git bisect bad 9fed0ddc5bc015f98246a2d8d9adbe5fb2b91ba4 # first bad commit: [9fed0ddc5bc015f98246a2d8d9adbe5fb2b91ba4] [FLINK-14400] Shrink the scope of MemoryManager from TaskExecutor to slot {code} -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Created] (FLINK-15013) Flink (on YARN) sometimes needs too many slots
Aljoscha Krettek created FLINK-15013: Summary: Flink (on YARN) sometimes needs too many slots Key: FLINK-15013 URL: https://issues.apache.org/jira/browse/FLINK-15013 Project: Flink Issue Type: Bug Components: Runtime / Coordination Affects Versions: 1.10.0 Reporter: Aljoscha Krettek Fix For: 1.10.0 Attachments: DualInputWordCount.jar *THIS IS DIFFERENT FROM FLINK-15007, even though the text looks almost the same.* This was discovered while debugging FLINK-14834. In some cases a Flink needs needs more slots to execute than expected. You can see this in some of the logs attached to FLINK-14834. You can reproduce this using [https://github.com/aljoscha/docker-hadoop-cluster] to bring up a YARN cluster and then running a compiled Flink in there. When you run {code:java} bin/flink run -m yarn-cluster -p 3 -yjm 1224 -ytm 1224 /root/DualInputWordCount.jar --input hdfs:///wc-in-1 --output hdfs:///wc-out && hdfs dfs -rm -r /wc-out {code} and check the logs afterwards you will sometimes see three "Requesting new slot..." statements and sometimes you will see four. This is the {{git bisect}} log that identifies the first faulty commit ([https://github.com/apache/flink/commit/2ab8b61f2f22f1a1ce7f92cd6b8dd32d2c0c227d|https://github.com/apache/flink/commit/2ab8b61f2f22f1a1ce7f92cd6b8dd32d2c0c227d]): {code:java} git bisect start # good: [09f2f43a1d73c76bf4d3f4a1205269eb860deb14] [FLINK-14154][ml] Add the class for multivariate Gaussian Distribution. git bisect good 09f2f43a1d73c76bf4d3f4a1205269eb860deb14 # bad: [01d6972ab267807b8afccb09a45c454fa76d6c4b] [hotfix] Refactor out slots creation from the TaskSlotTable constructor git bisect bad 01d6972ab267807b8afccb09a45c454fa76d6c4b # bad: [7a61c582c7213f123e10de4fd11a13d96425fd77] [hotfix] Fix wrong Java doc comment of BroadcastStateBootstrapFunction.Context git bisect bad 7a61c582c7213f123e10de4fd11a13d96425fd77 # good: [edeec8d7420185d1c49b2739827bd921d2c2d485] [hotfix][runtime] Replace all occurrences of letter to mail to unify wording of variables and documentation. git bisect good edeec8d7420185d1c49b2739827bd921d2c2d485 # bad: [1b4ebce86b71d56f44185f1cb83d9a3b51de13df] [FLINK-14262][table-planner-blink] support referencing function with fully/partially qualified names in blink git bisect bad 1b4ebce86b71d56f44185f1cb83d9a3b51de13df # good: [25a3d9138cd5e39fc786315682586b75d8ac86ea] [hotfix] Move TaskManagerSlot to o.a.f.runtime.resourcemanager.slotmanager git bisect good 25a3d9138cd5e39fc786315682586b75d8ac86ea # good: [362d7670593adc2e4b20650c8854398727d8102b] [FLINK-12122] Calculate TaskExecutorUtilization when listing available slots git bisect good 362d7670593adc2e4b20650c8854398727d8102b # bad: [7e8218515baf630e668348a68ff051dfa49c90c3] [FLINK-13969][Checkpointing] Do not allow trigger new checkpoitn after stop the coordinator git bisect bad 7e8218515baf630e668348a68ff051dfa49c90c3 # bad: [269e7f007e855c2bdedf8bad64ef13f516a608a6] [FLINK-12122] Choose SlotSelectionStrategy based on ClusterOptions#EVENLY_SPREAD_OUT_SLOTS_STRATEGY git bisect bad 269e7f007e855c2bdedf8bad64ef13f516a608a6 # bad: [2ab8b61f2f22f1a1ce7f92cd6b8dd32d2c0c227d] [FLINK-12122] Add EvenlySpreadOutLocationPreferenceSlotSelectionStrategy git bisect bad 2ab8b61f2f22f1a1ce7f92cd6b8dd32d2c0c227d # first bad commit: [2ab8b61f2f22f1a1ce7f92cd6b8dd32d2c0c227d] [FLINK-12122] Add EvenlySpreadOutLocationPreferenceSlotSelectionStrategy {code} I'm using the streaming WordCount example that I modified to have two "inputs", similar to how the WordCount example is used in the YARN/kerberos/Docker test. Instead of using the input once we use it like this: {code} text = env.readTextFile(params.get("input")).union(env.readTextFile(params.get("input"))); {code} to create two inputs from the same path. A jar is attached. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Created] (FLINK-15116) Make JobClient stateless, remove AutoCloseable
Aljoscha Krettek created FLINK-15116: Summary: Make JobClient stateless, remove AutoCloseable Key: FLINK-15116 URL: https://issues.apache.org/jira/browse/FLINK-15116 Project: Flink Issue Type: Sub-task Components: API / DataSet, API / DataStream Reporter: Aljoscha Krettek Currently, {{JobClient}} is {{AutoCloseable}} and we require users to close the {{JobClient}} that they get as a result from {{executeAsync()}}. This is problematic because users can simply ignore the result of {{executeAsync()}} and then we will leak the resources that the client has. We should change the {{JobClient}} so that it acquires the required {{ClusterClient}} for each method call and closes it again. This means that the users no longer have the burden of managing the JobClient lifecycle, i.e. they can freely ignore the result of executeAsync(). -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Created] (FLINK-15121) Add public constructor/getter for execution environments that takes Configuration
Aljoscha Krettek created FLINK-15121: Summary: Add public constructor/getter for execution environments that takes Configuration Key: FLINK-15121 URL: https://issues.apache.org/jira/browse/FLINK-15121 Project: Flink Issue Type: Sub-task Components: API / DataSet, API / DataStream Reporter: Aljoscha Krettek Assignee: Aljoscha Krettek Fix For: 1.10.0 Currently, you cannot create an {{ExecutionEnvironment}} from a {{Configuration}}, which you would need to configure an arbitrary Executor, such as one of the YARN executors. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Created] (FLINK-15129) Return JobClient instead of JobClient Future from executeAsync()
Aljoscha Krettek created FLINK-15129: Summary: Return JobClient instead of JobClient Future from executeAsync() Key: FLINK-15129 URL: https://issues.apache.org/jira/browse/FLINK-15129 Project: Flink Issue Type: Sub-task Components: API / DataSet, API / DataStream Reporter: Aljoscha Krettek Currently, users have to write this when they want to use the {{JobClient}}: {code} CompletableFuture jobClientFuture = env.executeAsync(); JobClient jobClient = jobClientFuture.get(); // or use thenApply/thenCompose etc. {code} instead we could always return a {{JobClient}} right away and therefore remove one step for the user. I don't know if it's always the right choice, but currently we always return an already completed future that contains the {{JobClient}}. In the future we might want to return a future that actually completes at some later point, we would not be able to do this if we directly return a {{JobClient}} and would have to block in {{executeAsync()}}. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Created] (FLINK-15157) Make ScalaShell ensureYarnConfig() and fetchConnectionInfo() public
Aljoscha Krettek created FLINK-15157: Summary: Make ScalaShell ensureYarnConfig() and fetchConnectionInfo() public Key: FLINK-15157 URL: https://issues.apache.org/jira/browse/FLINK-15157 Project: Flink Issue Type: Bug Components: Scala Shell Reporter: Aljoscha Krettek This allows users of the Scala Shell, such as Zeppelin to work better with the shell. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Created] (FLINK-12015) TaskManagerRunnerTest is unstable
Aljoscha Krettek created FLINK-12015: Summary: TaskManagerRunnerTest is unstable Key: FLINK-12015 URL: https://issues.apache.org/jira/browse/FLINK-12015 Project: Flink Issue Type: Bug Components: Runtime / Coordination Reporter: Aljoscha Krettek I saw this failure: {code:java} 17:34:16.833 [INFO] Running org.apache.flink.runtime.taskexecutor.TaskManagerRunnerTest 17:34:19.872 [ERROR] Tests run: 2, Failures: 1, Errors: 0, Skipped: 0, Time elapsed: 3.036 s <<< FAILURE! - in org.apache.flink.runtime.taskexecutor.TaskManagerRunnerTest 17:34:19.880 [ERROR] testShouldShutdownOnFatalError(org.apache.flink.runtime.taskexecutor.TaskManagerRunnerTest) Time elapsed: 0.353 s <<< FAILURE! java.lang.AssertionError: Expected: is <1> but: was <0> at org.apache.flink.runtime.taskexecutor.TaskManagerRunnerTest.testShouldShutdownOnFatalError(TaskManagerRunnerTest.java:59) {code} Travis log: https://travis-ci.org/apache/flink/jobs/511042156 -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Created] (FLINK-12019) ZooKeeperHaServicesTest.testCloseAndCleanupAllDataWithUncle is unstable
Aljoscha Krettek created FLINK-12019: Summary: ZooKeeperHaServicesTest.testCloseAndCleanupAllDataWithUncle is unstable Key: FLINK-12019 URL: https://issues.apache.org/jira/browse/FLINK-12019 Project: Flink Issue Type: Bug Components: Runtime / Coordination Affects Versions: 1.8.0 Reporter: Aljoscha Krettek This is the relevant part of the log: {code} 08:26:51.737 [INFO] Running org.apache.flink.runtime.highavailability.zookeeper.ZooKeeperHaServicesTest 08:26:52.862 [INFO] Tests run: 23, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 8.01 s - in org.apache.flink.runtime.jobmaster.JobMasterTest 08:26:53.612 [INFO] Running org.apache.flink.runtime.highavailability.zookeeper.ZooKeeperRegistryTest 08:26:54.722 [ERROR] Tests run: 3, Failures: 1, Errors: 0, Skipped: 0, Time elapsed: 2.983 s <<< FAILURE! - in org.apache.flink.runtime.highavailability.zookeeper.ZooKeeperHaServicesTest 08:26:54.722 [ERROR] testCloseAndCleanupAllDataWithUncle(org.apache.flink.runtime.highavailability.zookeeper.ZooKeeperHaServicesTest) Time elapsed: 0.067 s <<< FAILURE! java.lang.AssertionError: Expected: is null but: was <75,75,1553588814702,1553588814702,0,0,0,0,0,0,75 > at org.apache.flink.runtime.highavailability.zookeeper.ZooKeeperHaServicesTest.testCloseAndCleanupAllDataWithUncle(ZooKeeperHaServicesTest.java:160) {code} Travis run: https://travis-ci.org/apache/flink/jobs/511335500 -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Created] (FLINK-12030) KafkaITCase.testMultipleSourcesOnePartition is unstable
Aljoscha Krettek created FLINK-12030: Summary: KafkaITCase.testMultipleSourcesOnePartition is unstable Key: FLINK-12030 URL: https://issues.apache.org/jira/browse/FLINK-12030 Project: Flink Issue Type: Bug Components: Connectors / Kafka, Tests Reporter: Aljoscha Krettek This is a relevant part from the log: ``` 14:11:45,305 INFO org.apache.flink.streaming.connectors.kafka.KafkaITCase - Test testMetricsAndEndOfStream(org.apache.flink.streaming.connectors.kafka.KafkaITCase) is running. 14:11:45,310 INFO org.apache.flink.streaming.connectors.kafka.KafkaTestBase - === == Writing sequence of 300 into testEndOfStream with p=1 === 14:11:45,311 INFO org.apache.flink.streaming.connectors.kafka.KafkaTestBase - Writing attempt #1 14:11:45,316 INFO org.apache.flink.streaming.connectors.kafka.KafkaTestEnvironmentImpl - Creating topic testEndOfStream-1 14:11:45,863 WARN org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer - Property [transaction.timeout.ms] not specified. Setting it to 360 ms 14:11:45,910 WARN org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer - Using AT_LEAST_ONCE semantic, but checkpointing is not enabled. Switching to NONE semantic. 14:11:45,921 INFO org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer - Starting FlinkKafkaInternalProducer (1/1) to produce into default topic testEndOfStream-1 14:11:46,006 ERROR org.apache.flink.streaming.connectors.kafka.KafkaTestBase - Write attempt failed, trying again org.apache.flink.runtime.client.JobExecutionException: Job execution failed. at org.apache.flink.runtime.jobmaster.JobResult.toJobExecutionResult(JobResult.java:146) at org.apache.flink.runtime.minicluster.MiniCluster.executeJobBlocking(MiniCluster.java:638) at org.apache.flink.streaming.util.TestStreamEnvironment.execute(TestStreamEnvironment.java:79) at org.apache.flink.streaming.connectors.kafka.KafkaConsumerTestBase.writeSequence(KafkaConsumerTestBase.java:1918) at org.apache.flink.streaming.connectors.kafka.KafkaConsumerTestBase.runEndOfStreamTest(KafkaConsumerTestBase.java:1537) at org.apache.flink.streaming.connectors.kafka.KafkaITCase.testMetricsAndEndOfStream(KafkaITCase.java:136) 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.internal.runners.statements.FailOnTimeout$CallableStatement.call(FailOnTimeout.java:298) at org.junit.internal.runners.statements.FailOnTimeout$CallableStatement.call(FailOnTimeout.java:292) at java.util.concurrent.FutureTask.run(FutureTask.java:266) at java.lang.Thread.run(Thread.java:748) Caused by: org.apache.flink.streaming.connectors.kafka.FlinkKafkaException: Failed to send data to Kafka: This server does not host this topic-partition. at org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer.checkErroneous(FlinkKafkaProducer.java:1002) at org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer.flush(FlinkKafkaProducer.java:787) at org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer.close(FlinkKafkaProducer.java:658) at org.apache.flink.api.common.functions.util.FunctionUtils.closeFunction(FunctionUtils.java:43) at org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator.close(AbstractUdfStreamOperator.java:109) at org.apache.flink.streaming.runtime.tasks.StreamTask.closeAllOperators(StreamTask.java:443) at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:318) at org.apache.flink.runtime.taskmanager.Task.run(Task.java:711) ... 1 more ``` Travis run: https://travis-ci.org/apache/flink/jobs/510953235 -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Created] (FLINK-12164) JobMasterTest.testJobFailureWhenTaskExecutorHeartbeatTimeout is unstable
Aljoscha Krettek created FLINK-12164: Summary: JobMasterTest.testJobFailureWhenTaskExecutorHeartbeatTimeout is unstable Key: FLINK-12164 URL: https://issues.apache.org/jira/browse/FLINK-12164 Project: Flink Issue Type: Bug Components: Runtime / Coordination Reporter: Aljoscha Krettek -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Created] (FLINK-12501) AvroTypeSerializer does not work with types generated by avrohugger
Aljoscha Krettek created FLINK-12501: Summary: AvroTypeSerializer does not work with types generated by avrohugger Key: FLINK-12501 URL: https://issues.apache.org/jira/browse/FLINK-12501 Project: Flink Issue Type: Bug Components: Formats (JSON, Avro, Parquet, ORC, SequenceFile) Reporter: Aljoscha Krettek The main problem is that the code in {{SpecificData.createSchema()}} tries to reflectively read the {{SCHEMA$}} field, that is normally there in Avro generated classes. However, avrohugger generates this field in a companion object, which the reflective Java code will therefore not find. This is also described in these ML threads: * [https://lists.apache.org/thread.html/5db58c7d15e4e9aaa515f935be3b342fe036e97d32e1fb0f0d1797ee@%3Cuser.flink.apache.org%3E] * [https://lists.apache.org/thread.html/cf1c5b8fa7f095739438807de9f2497e04ffe55237c5dea83355112d@%3Cuser.flink.apache.org%3E] -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Created] (FLINK-12587) Add pure-API Watermark class
Aljoscha Krettek created FLINK-12587: Summary: Add pure-API Watermark class Key: FLINK-12587 URL: https://issues.apache.org/jira/browse/FLINK-12587 Project: Flink Issue Type: Improvement Components: API / DataStream Reporter: Aljoscha Krettek Assignee: Aljoscha Krettek Currently, {{Watermark}} extends {{StreamElement}}, which comes with a lot of transitive dependencies to classes and modules that are runtime code and shouldn't be in an API package. In the future, we want to more clearly separate API packages from runtime packages. The first instance of this is trying to put {{TableSources}} into {{flink-table-common}}, which will require our source-like interfaces to be clean of runtime dependencies. This means that we have to put {{SourceFunction}} into {{flink-core}}, which has {{Watermark}} in its interface, which we therefore have to untangle from runtime classes. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Created] (FLINK-12606) Move SourceFunction to flink-core
Aljoscha Krettek created FLINK-12606: Summary: Move SourceFunction to flink-core Key: FLINK-12606 URL: https://issues.apache.org/jira/browse/FLINK-12606 Project: Flink Issue Type: Improvement Components: API / DataStream Reporter: Aljoscha Krettek Assignee: Aljoscha Krettek As part of the efforts to merge the new Blink-based Table API runner (FLINK-11439) we need to make {{TableSource}}/{{TableSink}} interfaces available in {{flink-table-common}}. This requires, among other things, that the "physical" source interfaces are available in {{flink-core}}, independent of DataSet/DataStream dependencies. For {{InputFormat}} this is already the case, we need to do the same for {{SourceFunction}}. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Created] (FLINK-12613) Move SinkFunction to flink-core
Aljoscha Krettek created FLINK-12613: Summary: Move SinkFunction to flink-core Key: FLINK-12613 URL: https://issues.apache.org/jira/browse/FLINK-12613 Project: Flink Issue Type: Improvement Components: API / DataStream Reporter: Aljoscha Krettek Assignee: Aljoscha Krettek As part of the efforts to merge the new Blink-based Table API runner (FLINK-11439) we need to make TableSource/TableSink interfaces available in {{flink-table-common}}. This requires, among other things, that the "physical" source interfaces are available in flink-core, independent of DataSet/DataStream dependencies. For {{InputFormat}} this is already the case, we need to do the same for {{SinkFunction}}. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Created] (FLINK-12832) Enhance StreamGraphGenerator/StreamGraph for Blink Table Runner
Aljoscha Krettek created FLINK-12832: Summary: Enhance StreamGraphGenerator/StreamGraph for Blink Table Runner Key: FLINK-12832 URL: https://issues.apache.org/jira/browse/FLINK-12832 Project: Flink Issue Type: Improvement Components: API / DataStream Reporter: Biao Liu Assignee: Biao Liu Fix For: 1.9.0 This is a part of merging Blink batch runner. I would like to improvement the StreamGraph to support both streaming job and batch job. There is a google doc to describe this proposal, here is the [link|https://docs.google.com/document/d/17X6csUWKUdVn55c47YbOTEipHDWFt_1upyTPrBD6Fjc/edit?usp=sharing]. This proposal is under discussion, any feedback is welcome! The new Table API runner directly wants to translate to {{StreamGraphs}} and therefore doesn't have a {{StreamExecutionEnvironment}}, we therefore need to disentangle them. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Created] (FLINK-13041) Make ScheduleMode configurable on ExecutionConfig
Aljoscha Krettek created FLINK-13041: Summary: Make ScheduleMode configurable on ExecutionConfig Key: FLINK-13041 URL: https://issues.apache.org/jira/browse/FLINK-13041 Project: Flink Issue Type: Improvement Components: API / DataStream Reporter: Aljoscha Krettek Assignee: Aljoscha Krettek The new Blink-based Table API Runner currently needs a way of globally setting the {{ScheduleMode}} (either {{LAZY_FROM_SOURCES}} or {{EAGER}}) to support both batch-style and streaming-style execution. This is a stop-gap solution. Ideally, {{Transformations}} would know whether they are bounded or not and the scheduler could decide the scheduling mode based on that information but this requires more work on both the Table API Runner and the DAG API. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Created] (FLINK-13062) Set ScheduleMode based on boundedness of streaming Pipeline
Aljoscha Krettek created FLINK-13062: Summary: Set ScheduleMode based on boundedness of streaming Pipeline Key: FLINK-13062 URL: https://issues.apache.org/jira/browse/FLINK-13062 Project: Flink Issue Type: Improvement Components: API / DataStream Reporter: Aljoscha Krettek Assignee: Aljoscha Krettek The new Blink-based Table Runner needs "streaming pipelines" to be executed with {{ScheduleMode.LAZY_FROM_SOURCES}} if all sources are bounded. The current Blink code base uses a global flag for this and configures the {{StreamGraphGenerator}} accordingly. We propose to add an {{isBounded()}} property to {{Transformation}} (formerly known as {{StreamTransformation}}). The property would only be explicitly settable on sources, other transformations inherit the property from their inputs. The {{StreamGraphGenerator}} must use {{ScheduleMode.LAZY_FROM_SOURCES}} if all sources are bounded, otherwise, it should use {{ScheduleMode.EAGER}}, as is the currently existing behaviour. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Created] (FLINK-13396) Deprecate BucketingSink
Aljoscha Krettek created FLINK-13396: Summary: Deprecate BucketingSink Key: FLINK-13396 URL: https://issues.apache.org/jira/browse/FLINK-13396 Project: Flink Issue Type: Improvement Components: Connectors / FileSystem Reporter: Aljoscha Krettek Assignee: Kostas Kloudas Fix For: 1.9.0 This has been superseded by the new {{StreamingFileSink}} and should be removed at some point in the future. For this we need to deprecate it first. -- This message was sent by Atlassian JIRA (v7.6.14#76016)
[jira] [Created] (FLINK-13397) Remove RollingSink
Aljoscha Krettek created FLINK-13397: Summary: Remove RollingSink Key: FLINK-13397 URL: https://issues.apache.org/jira/browse/FLINK-13397 Project: Flink Issue Type: Improvement Components: Connectors / FileSystem Reporter: Aljoscha Krettek Assignee: Aljoscha Krettek Fix For: 1.9.0 This has been superseded by the {{BucketingSink}} and has been deprecated for some releases now. -- This message was sent by Atlassian JIRA (v7.6.14#76016)
[jira] [Created] (FLINK-13601) RegionFailoverITCase is unstable
Aljoscha Krettek created FLINK-13601: Summary: RegionFailoverITCase is unstable Key: FLINK-13601 URL: https://issues.apache.org/jira/browse/FLINK-13601 Project: Flink Issue Type: Improvement Components: Runtime / Checkpointing, Runtime / Coordination Affects Versions: 1.9.0, 1.10.0 Reporter: Aljoscha Krettek Fix For: 1.9.0 Excerpt from https://travis-ci.com/flink-ci/flink/jobs/222711830: {code} 10:44:31.222 [INFO] Running org.apache.flink.test.checkpointing.RegionFailoverITCase org.apache.flink.client.program.ProgramInvocationException: Job failed (JobID: 9e0fbeaa580123e05cfce5554f443d23) at org.apache.flink.client.program.MiniClusterClient.submitJob(MiniClusterClient.java:92) at org.apache.flink.test.checkpointing.RegionFailoverITCase.testMultiRegionFailover(RegionFailoverITCase.java:132) 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.internal.runners.statements.FailOnTimeout$CallableStatement.call(FailOnTimeout.java:298) at org.junit.internal.runners.statements.FailOnTimeout$CallableStatement.call(FailOnTimeout.java:292) at java.util.concurrent.FutureTask.run(FutureTask.java:266) at java.lang.Thread.run(Thread.java:748) Caused by: org.apache.flink.runtime.client.JobExecutionException: Job execution failed. at org.apache.flink.runtime.jobmaster.JobResult.toJobExecutionResult(JobResult.java:146) at org.apache.flink.client.program.MiniClusterClient.submitJob(MiniClusterClient.java:90) ... 13 more Caused by: java.lang.RuntimeException: Test failed due to unexpected recovered index: 2000, while last completed checkpoint record index: 1837 at org.apache.flink.test.checkpointing.RegionFailoverITCase$StringGeneratingSourceFunction.initializeState(RegionFailoverITCase.java:300) at org.apache.flink.streaming.util.functions.StreamingFunctionUtils.tryRestoreFunction(StreamingFunctionUtils.java:178) at org.apache.flink.streaming.util.functions.StreamingFunctionUtils.restoreFunctionState(StreamingFunctionUtils.java:160) at org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator.initializeState(AbstractUdfStreamOperator.java:96) at org.apache.flink.streaming.api.operators.AbstractStreamOperator.initializeState(AbstractStreamOperator.java:281) at org.apache.flink.streaming.runtime.tasks.StreamTask.initializeState(StreamTask.java:862) at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:367) at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:688) at org.apache.flink.runtime.taskmanager.Task.run(Task.java:518) ... 1 more 10:44:39.210 [ERROR] Tests run: 1, Failures: 1, Errors: 0, Skipped: 0, Time elapsed: 7.983 s <<< FAILURE! - in org.apache.flink.test.checkpointing.RegionFailoverITCase {code} -- This message was sent by Atlassian JIRA (v7.6.14#76016)
[jira] [Created] (FLINK-10563) Expose shaded Presto S3 filesystem under "s3p" scheme
Aljoscha Krettek created FLINK-10563: Summary: Expose shaded Presto S3 filesystem under "s3p" scheme Key: FLINK-10563 URL: https://issues.apache.org/jira/browse/FLINK-10563 Project: Flink Issue Type: Improvement Reporter: Aljoscha Krettek Assignee: Aljoscha Krettek Fix For: 1.7.0 Currently, you can't use the shaded hadoop S3 filesystem and the presto S3 filesystem at the same time. If we exposed the presto filesystem under an additional scheme we enable using both at the same time. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Created] (FLINK-10740) Refactor Source Interface
Aljoscha Krettek created FLINK-10740: Summary: Refactor Source Interface Key: FLINK-10740 URL: https://issues.apache.org/jira/browse/FLINK-10740 Project: Flink Issue Type: Improvement Components: DataSet API, DataStream API Reporter: Aljoscha Krettek Please see the FLIP for any details: https://cwiki.apache.org/confluence/display/FLINK/FLIP-27%3A+Refactor+Source+Interface -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Created] (FLINK-11539) Add TypeSerializerSnapshot for TraversableSerializer
Aljoscha Krettek created FLINK-11539: Summary: Add TypeSerializerSnapshot for TraversableSerializer Key: FLINK-11539 URL: https://issues.apache.org/jira/browse/FLINK-11539 Project: Flink Issue Type: Improvement Components: Type Serialization System Reporter: Aljoscha Krettek Assignee: Aljoscha Krettek This will replace the deprecated {{TypeSerializerConfigSnapshot}}. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Created] (FLINK-11693) Add KafkaSerializationSchema that directly uses ProducerRecord
Aljoscha Krettek created FLINK-11693: Summary: Add KafkaSerializationSchema that directly uses ProducerRecord Key: FLINK-11693 URL: https://issues.apache.org/jira/browse/FLINK-11693 Project: Flink Issue Type: New Feature Components: Kafka Connector Environment: Kafka 0.11.0.0 Flink 1.4.0 flink-connector-kafka-0.11_2.11 Reporter: Mohammad Abareghi Assignee: Aegeaner This allows using all fields that Kafka offers on {{ConsumerRecord}}, for example, timestamps and headers. h2. Old Text Kafka has introduced notion of Header for messages in version 0.11.0.0 https://issues.apache.org/jira/browse/KAFKA-4208. But flink-connector-kafka-0.11_2.11 which supports kafka 0.11.0.0 ignores headers when consuming kafka messages. It would be useful in some scenarios, such as distributed log tracing, to support message headers to FlinkKafkaConsumer011 and FlinkKafkaProducer011. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Created] (FLINK-11751) Extend release notes for Flink 1.8
Aljoscha Krettek created FLINK-11751: Summary: Extend release notes for Flink 1.8 Key: FLINK-11751 URL: https://issues.apache.org/jira/browse/FLINK-11751 Project: Flink Issue Type: Task Components: Documentation Reporter: Aljoscha Krettek Assignee: Aljoscha Krettek Fix For: 1.8.0 -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Created] (FLINK-11780) Change version scheme of Hadoop-based modules to conform to SNAPSHOT guidelines
Aljoscha Krettek created FLINK-11780: Summary: Change version scheme of Hadoop-based modules to conform to SNAPSHOT guidelines Key: FLINK-11780 URL: https://issues.apache.org/jira/browse/FLINK-11780 Project: Flink Issue Type: Task Reporter: Aljoscha Krettek Currently, modules that depend on Hadoop have the version scheme {{${project.version}-${hadoop.version}}}. With this, however, the version does not have {{SNAPSHOT}} at the end, meaning we cannot deploy snapshot releases. The change proposed by [~Zentol] is to use {{${hadoop.version}-${project.version}}}. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Created] (FLINK-11791) Describe how to build Flink with Hadoop in build guide
Aljoscha Krettek created FLINK-11791: Summary: Describe how to build Flink with Hadoop in build guide Key: FLINK-11791 URL: https://issues.apache.org/jira/browse/FLINK-11791 Project: Flink Issue Type: Task Reporter: Aljoscha Krettek Assignee: Aljoscha Krettek Fix For: 1.8.0 -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Created] (FLINK-11836) Update NOTICE-binary and licenses-binary for Flink 1.8.0
Aljoscha Krettek created FLINK-11836: Summary: Update NOTICE-binary and licenses-binary for Flink 1.8.0 Key: FLINK-11836 URL: https://issues.apache.org/jira/browse/FLINK-11836 Project: Flink Issue Type: Task Reporter: Aljoscha Krettek Assignee: Aljoscha Krettek Fix For: 1.8.0 -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Created] (FLINK-11865) Code generation in TraversableSerializer is prohibitively slow
Aljoscha Krettek created FLINK-11865: Summary: Code generation in TraversableSerializer is prohibitively slow Key: FLINK-11865 URL: https://issues.apache.org/jira/browse/FLINK-11865 Project: Flink Issue Type: Bug Reporter: Aljoscha Krettek Fix For: 1.8.0 As discussed in FLINK-11539, the new code generation makes job submissions/translation prohibitively slow. The solution should be to introduce a Cache for the generated code. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Created] (FLINK-11889) Remove "stop" signal along with Stoppable interfaces
Aljoscha Krettek created FLINK-11889: Summary: Remove "stop" signal along with Stoppable interfaces Key: FLINK-11889 URL: https://issues.apache.org/jira/browse/FLINK-11889 Project: Flink Issue Type: Improvement Components: API / DataStream Reporter: Aljoscha Krettek Fix For: 1.9.0 During the discussion of [FLIP-34|https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=103090212] we realised that it would be beneficial for this new feature to replace the existing "stop" functionality. The current "stop" functionality cannot be used because no real-world sources support the functionality. Therefore, I think it is save to remove because it should not break existing workflows. The issue proposes completely removing the old stop feature, introduced via FLINK-2111, as preparation for [FLIP-34|https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=103090212]. We have to be careful when doing this because it touches quite a few things. Basically, we have to do a manual revert of this commit: https://github.com/apache/flink/commit/bdd4024e20fdfb0accb6121a68780ce3a0c218c0 -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Created] (FLINK-12006) ZooKeeperHaServicesTest is unstable
Aljoscha Krettek created FLINK-12006: Summary: ZooKeeperHaServicesTest is unstable Key: FLINK-12006 URL: https://issues.apache.org/jira/browse/FLINK-12006 Project: Flink Issue Type: Bug Components: Runtime / Coordination Reporter: Aljoscha Krettek This is the log from a failed Travis run: https://travis-ci.org/apache/flink/jobs/510263371 The relevant section is: {code} 11:15:11.201 [INFO] Running org.apache.flink.runtime.highavailability.zookeeper.ZooKeeperHaServicesTest 11:15:12.908 [INFO] Tests run: 23, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 9.255 s - in org.apache.flink.runtime.jobmaster.JobMasterTest 11:15:13.646 [INFO] Running org.apache.flink.runtime.highavailability.zookeeper.ZooKeeperRegistryTest 11:15:14.874 [ERROR] Tests run: 3, Failures: 1, Errors: 0, Skipped: 0, Time elapsed: 3.67 s <<< FAILURE! - in org.apache.flink.runtime.highavailability.zookeeper.ZooKeeperHaServicesTest 11:15:14.874 [ERROR] testSimpleCloseAndCleanupAllData(org.apache.flink.runtime.highavailability.zookeeper.ZooKeeperHaServicesTest) Time elapsed: 0.132 s <<< FAILURE! java.lang.AssertionError: Expected: is <[zookeeper]> but: was <[zookeeper, foo]> at org.apache.flink.runtime.highavailability.zookeeper.ZooKeeperHaServicesTest.testSimpleCloseAndCleanupAllData(ZooKeeperHaServicesTest.java:136) {code} -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Created] (FLINK-15251) Fabric8FlinkKubeClient doesn't work if ingress has hostname but no IP
Aljoscha Krettek created FLINK-15251: Summary: Fabric8FlinkKubeClient doesn't work if ingress has hostname but no IP Key: FLINK-15251 URL: https://issues.apache.org/jira/browse/FLINK-15251 Project: Flink Issue Type: Bug Components: Deployment / Kubernetes Affects Versions: 1.10.0 Environment: Kubernetes for Docker on MacOS Reporter: Aljoscha Krettek Fix For: 1.10.0 In my setup the ingress has a hostname but no IP here: https://github.com/apache/flink/blob/f49e632bb290ded45b320f5d00ceaa1543a6bb1c/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/Fabric8FlinkKubeClient.java#L199 This means that when I try to use the Kubernetes Executor I will get {code} Exception in thread "main" java.lang.NullPointerException: Address should not be null. at org.apache.flink.util.Preconditions.checkNotNull(Preconditions.java:75) at org.apache.flink.kubernetes.kubeclient.Endpoint.(Endpoint.java:33) at org.apache.flink.kubernetes.kubeclient.Fabric8FlinkKubeClient.getRestEndpoint(Fabric8FlinkKubeClient.java:209) at org.apache.flink.kubernetes.KubernetesClusterDescriptor.lambda$createClusterClientProvider$0(KubernetesClusterDescriptor.java:82) at org.apache.flink.kubernetes.KubernetesClusterDescriptor.retrieve(KubernetesClusterDescriptor.java:106) at org.apache.flink.kubernetes.KubernetesClusterDescriptor.retrieve(KubernetesClusterDescriptor.java:53) at org.apache.flink.client.deployment.AbstractSessionClusterExecutor.execute(AbstractSessionClusterExecutor.java:60) at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.executeAsync(StreamExecutionEnvironment.java:1741) at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.executeAsync(StreamExecutionEnvironment.java:1712) at org.apache.flink.streaming.examples.statemachine.StateMachineExample.main(StateMachineExample.java:169) {code} I think we can just check if a hostname is set and use that if there is no IP. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Created] (FLINK-15265) Remove "-executor" suffix from executor names
Aljoscha Krettek created FLINK-15265: Summary: Remove "-executor" suffix from executor names Key: FLINK-15265 URL: https://issues.apache.org/jira/browse/FLINK-15265 Project: Flink Issue Type: Bug Components: API / Core Reporter: Aljoscha Krettek Assignee: Aljoscha Krettek The executor names always have "-executor" as a suffix, this is reduntant. Currently, the executor name is also used to retrieve a {{ClusterClient}}, where it is unfortunate that the name has executor as a suffix. In the future we might provide something like a {{FlinkClient}} that offers a programmatic API for the functionality of {{bin/flink}}, here we would also use the same names. In reality, the "executor names" are not names of executors but deployment targets. That's why the current naming seems a bit unnatural. This is a simple search-and-replace job, no new functionality. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Created] (FLINK-15474) In TypeSerializerUpgradeTestBase, create serializer snapshots "on demand"
Aljoscha Krettek created FLINK-15474: Summary: In TypeSerializerUpgradeTestBase, create serializer snapshots "on demand" Key: FLINK-15474 URL: https://issues.apache.org/jira/browse/FLINK-15474 Project: Flink Issue Type: Bug Components: API / Type Serialization System, Tests Affects Versions: 1.9.0, 1.10.0 Reporter: Aljoscha Krettek Assignee: Aljoscha Krettek Fix For: 1.10.0 Currently, we store binary snapshots in the repository for all the different serializer upgrade test configurations (see linked POC for which snapshots are there for just the POJO serializer). This is hard to maintain, because someone has to go back and generate snapshtos from previous Flink versions and add them to the repo when updating the tests for a new Flink version. It's also problematic from a repository perspective because we keep piling up binary snapshots. Instead, we can create a snapshot "on demand" from a previous Flink version by using a classloader that has the previous Flink jar. I created a POC which demonstrated the approach: [https://github.com/aljoscha/flink/tree/jit-serializer-test-base]. The advantage is that we don't need binary snapshots in the repo anymore, updating the tests to a newer Flink version should be as easy as adding a new migration version and Flink download url. The downside is that the test now downloads Flink releases (in the PoC this is done using the caching infra introduced for e2e tests), which is costly and also re-generates the snapshots for every test, which is also costly. The test time (minus downloading) goes up from about 300 ms to roughly 6 seconds. That's not something I would call a unit test. We could call these "integration tests" (or even e2e tests) and only run them nightly. Side note, we don't have test coverage for serializer upgrades from 1.8 and 1.9 currently, so even doing it nightly would be an improvement. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Created] (FLINK-15518) Don't hide web frontend side pane automatically
Aljoscha Krettek created FLINK-15518: Summary: Don't hide web frontend side pane automatically Key: FLINK-15518 URL: https://issues.apache.org/jira/browse/FLINK-15518 Project: Flink Issue Type: Bug Components: Runtime / Web Frontend Affects Versions: 1.9.0, 1.10.0 Reporter: Aljoscha Krettek Fix For: 1.10.0 As mentioned in FLINK-13386 the side pane hides automatically in some cases but not all cases. When I was debugging or trying the web frontend I found this behaviour a bit disconcerting. Could we disable the hiding by default? The user can still manually hide if they want to. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Created] (FLINK-15735) Too many warnings when running bin/start-cluster.sh
Aljoscha Krettek created FLINK-15735: Summary: Too many warnings when running bin/start-cluster.sh Key: FLINK-15735 URL: https://issues.apache.org/jira/browse/FLINK-15735 Project: Flink Issue Type: Bug Components: Command Line Client Reporter: Aljoscha Krettek Fix For: 1.10.0 Running {{bin/start-cluster.sh}} now prints: {code} $ bin/start-cluster.sh Starting cluster. Starting standalonesession daemon on host Aljoschas-MacBook-Pro-Work.local. log4j:WARN No appenders could be found for logger (org.apache.flink.configuration.GlobalConfiguration). log4j:WARN Please initialize the log4j system properly. log4j:WARN See http://logging.apache.org/log4j/1.2/faq.html#noconfig for more info. log4j:WARN No appenders could be found for logger (org.apache.flink.configuration.GlobalConfiguration). log4j:WARN Please initialize the log4j system properly. log4j:WARN See http://logging.apache.org/log4j/1.2/faq.html#noconfig for more info. Starting taskexecutor daemon on host Aljoschas-MacBook-Pro-Work.local. {code} I think we were not printing that before and we probably shouldn't do so now. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Created] (FLINK-15904) Make Kafka Consumer work with activated "disableGenericTypes()"
Aljoscha Krettek created FLINK-15904: Summary: Make Kafka Consumer work with activated "disableGenericTypes()" Key: FLINK-15904 URL: https://issues.apache.org/jira/browse/FLINK-15904 Project: Flink Issue Type: Bug Components: Connectors / Kafka Reporter: Aljoscha Krettek A user reported a problem that the Kafka Consumer doesn't work in that case: https://lists.apache.org/thread.html/r462a854e8a0ab3512e2906b40411624f3164ea3af7cba61ee94cd760%40%3Cuser.flink.apache.org%3E. We should use a different constructor for {{ListStateDescriptor}} that takes {{TypeSerializer}} here: https://github.com/apache/flink/blob/68cc21e4af71505efa142110e35a1f8b1c25fe6e/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumerBase.java#L860. This will circumvent the check. My full analysis from the email thread: {quote} Unfortunately, the fact that the Kafka Sources use Kryo for state serialization is a very early design misstep that we cannot get rid of for now. We will get rid of that when the new source interface lands ([1]) and when we have a new Kafka Source based on that. As a workaround, we should change the Kafka Consumer to go through a different constructor of ListStateDescriptor which directly takes a TypeSerializer instead of a TypeInformation here: [2]. This should sidestep the "no generic types" check. [1] https://cwiki.apache.org/confluence/display/FLINK/FLIP-27%3A+Refactor+Source+Interface [2] https://github.com/apache/flink/blob/68cc21e4af71505efa142110e35a1f8b1c25fe6e/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumerBase.java#L860 {quote} -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Created] (FLINK-15993) Add timeout to 404 documentation redirect, add explanation
Aljoscha Krettek created FLINK-15993: Summary: Add timeout to 404 documentation redirect, add explanation Key: FLINK-15993 URL: https://issues.apache.org/jira/browse/FLINK-15993 Project: Flink Issue Type: Sub-task Components: Documentation Reporter: Aljoscha Krettek Assignee: Aljoscha Krettek -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Created] (FLINK-15997) Make documentation 404 page look like a documentation page
Aljoscha Krettek created FLINK-15997: Summary: Make documentation 404 page look like a documentation page Key: FLINK-15997 URL: https://issues.apache.org/jira/browse/FLINK-15997 Project: Flink Issue Type: Sub-task Reporter: Aljoscha Krettek Assignee: Aljoscha Krettek -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Created] (FLINK-15998) Revert rename of "Job Cluster" to "Application Cluster" in documentation
Aljoscha Krettek created FLINK-15998: Summary: Revert rename of "Job Cluster" to "Application Cluster" in documentation Key: FLINK-15998 URL: https://issues.apache.org/jira/browse/FLINK-15998 Project: Flink Issue Type: Sub-task Components: Documentation Affects Versions: 1.10.0, 1.11.0 Reporter: Aljoscha Krettek Assignee: Aljoscha Krettek Fix For: 1.10.1, 1.11.0 [~kkl0u] and I are working on an upcoming FLIP that will propose a "real" application mode. The gist of it will be that an application cluster is roughly responsible for all the jobs that are executed in a user {{main()}} method. Each individual invocation of {{execute()}} in said {{main()}} function would launch a job on the aforementioned application cluster. The {{main()}} method would run inside the cluster (or not, that's up for discussion). FLINK-12625 introduced a glossary that describes what was previously known as a "Job Cluster" as an "Application Cluster". We think that in the future users will have both: per-job clusters and application clusters. We therefore think that we should describe our current per-job clusters as Per-job clusters (or job clusters) in the glossary and documentation and reserve the name application cluster for the "real" forthcoming application clusters. This would also render FLINK-12885 outdated. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Created] (FLINK-15999) Extract “Concepts” material from API/Library sections and start proper concepts section
Aljoscha Krettek created FLINK-15999: Summary: Extract “Concepts” material from API/Library sections and start proper concepts section Key: FLINK-15999 URL: https://issues.apache.org/jira/browse/FLINK-15999 Project: Flink Issue Type: Sub-task Components: Documentation Reporter: Aljoscha Krettek Assignee: Aljoscha Krettek -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Created] (FLINK-16000) Move "Project Build Setup" to "Getting Started" in documentation
Aljoscha Krettek created FLINK-16000: Summary: Move "Project Build Setup" to "Getting Started" in documentation Key: FLINK-16000 URL: https://issues.apache.org/jira/browse/FLINK-16000 Project: Flink Issue Type: Sub-task Components: Documentation Reporter: Aljoscha Krettek Assignee: Aljoscha Krettek -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Created] (FLINK-16041) Expand "popular" documentation sections by default
Aljoscha Krettek created FLINK-16041: Summary: Expand "popular" documentation sections by default Key: FLINK-16041 URL: https://issues.apache.org/jira/browse/FLINK-16041 Project: Flink Issue Type: Sub-task Components: Documentation Reporter: Aljoscha Krettek Assignee: Aljoscha Krettek Attachments: current-state.png, proposed-state.png Currently, when the documentation page is loaded all sections are collapsed, this means that some prominent subsections are not easily discoverable. I think we should expand the "Getting Started", "Concepts", and "API" sections by default. (See also https://cwiki.apache.org/confluence/display/FLINK/FLIP-42%3A+Rework+Flink+Documentation, because "API" doesn't exist yet in the current documentation. I attached screenshots to show what I mean. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Created] (FLINK-16045) Extract connectors documentation to a top-level section
Aljoscha Krettek created FLINK-16045: Summary: Extract connectors documentation to a top-level section Key: FLINK-16045 URL: https://issues.apache.org/jira/browse/FLINK-16045 Project: Flink Issue Type: Sub-task Components: Documentation Reporter: Aljoscha Krettek Assignee: Aljoscha Krettek -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Created] (FLINK-16044) Extract libraries documentation to a top-level section
Aljoscha Krettek created FLINK-16044: Summary: Extract libraries documentation to a top-level section Key: FLINK-16044 URL: https://issues.apache.org/jira/browse/FLINK-16044 Project: Flink Issue Type: Sub-task Components: Documentation Reporter: Aljoscha Krettek Assignee: Aljoscha Krettek -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Created] (FLINK-16049) Remove outdated "Best Pracatices" section from Application Development Section
Aljoscha Krettek created FLINK-16049: Summary: Remove outdated "Best Pracatices" section from Application Development Section Key: FLINK-16049 URL: https://issues.apache.org/jira/browse/FLINK-16049 Project: Flink Issue Type: Bug Components: Documentation Reporter: Aljoscha Krettek Assignee: Aljoscha Krettek -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Created] (FLINK-16144) Add client.timeout setting and use that for CLI operations
Aljoscha Krettek created FLINK-16144: Summary: Add client.timeout setting and use that for CLI operations Key: FLINK-16144 URL: https://issues.apache.org/jira/browse/FLINK-16144 Project: Flink Issue Type: Improvement Components: Command Line Client Reporter: Aljoscha Krettek Fix For: 1.11.0 Currently, the Cli uses the {{akka.client.timeout}} setting. This has historical reasons but can be very confusing for users. We should introduce a new setting {{client.timeout}} that is used for the client, with a fallback to the previous {{akka.client.timeout}}. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Created] (FLINK-16207) In stream processing concepts section, rework distribution patterns description
Aljoscha Krettek created FLINK-16207: Summary: In stream processing concepts section, rework distribution patterns description Key: FLINK-16207 URL: https://issues.apache.org/jira/browse/FLINK-16207 Project: Flink Issue Type: Sub-task Components: Documentation Reporter: Aljoscha Krettek Currently, we only distinguish between one-to-one and redistribution. We should instead describe it as - Forward - Broadcast - Random - Keyed -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Created] (FLINK-16208) Add introduction to timely stream processing concepts documentation
Aljoscha Krettek created FLINK-16208: Summary: Add introduction to timely stream processing concepts documentation Key: FLINK-16208 URL: https://issues.apache.org/jira/browse/FLINK-16208 Project: Flink Issue Type: Sub-task Components: Documentation Reporter: Aljoscha Krettek -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Created] (FLINK-16209) Add Latency and Completeness section in timely stream processing concepts
Aljoscha Krettek created FLINK-16209: Summary: Add Latency and Completeness section in timely stream processing concepts Key: FLINK-16209 URL: https://issues.apache.org/jira/browse/FLINK-16209 Project: Flink Issue Type: Sub-task Components: Documentation Reporter: Aljoscha Krettek -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Created] (FLINK-16210) Add section about applications and clusters/session in concepts documentation
Aljoscha Krettek created FLINK-16210: Summary: Add section about applications and clusters/session in concepts documentation Key: FLINK-16210 URL: https://issues.apache.org/jira/browse/FLINK-16210 Project: Flink Issue Type: Sub-task Reporter: Aljoscha Krettek This can either go into the existing _Flink Architecture_ ({{flink-architecture.md}}) documentation or be a new section. We can possibly remove the old _Flink Architecture_ section then. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Created] (FLINK-16211) Add introduction to stream processing concepts documentation
Aljoscha Krettek created FLINK-16211: Summary: Add introduction to stream processing concepts documentation Key: FLINK-16211 URL: https://issues.apache.org/jira/browse/FLINK-16211 Project: Flink Issue Type: Sub-task Components: Documentation Reporter: Aljoscha Krettek -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Created] (FLINK-16212) Describe how Flink is a unified batch/stream processing system in concepts documentation
Aljoscha Krettek created FLINK-16212: Summary: Describe how Flink is a unified batch/stream processing system in concepts documentation Key: FLINK-16212 URL: https://issues.apache.org/jira/browse/FLINK-16212 Project: Flink Issue Type: Sub-task Components: Documentation Reporter: Aljoscha Krettek -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Created] (FLINK-16213) Add "What Is State" section in concepts documentation
Aljoscha Krettek created FLINK-16213: Summary: Add "What Is State" section in concepts documentation Key: FLINK-16213 URL: https://issues.apache.org/jira/browse/FLINK-16213 Project: Flink Issue Type: Sub-task Reporter: Aljoscha Krettek This should go into {{stateful-stream-processing.md}}. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Created] (FLINK-16214) Describe how state is different for stream/batch programs in concepts documentation
Aljoscha Krettek created FLINK-16214: Summary: Describe how state is different for stream/batch programs in concepts documentation Key: FLINK-16214 URL: https://issues.apache.org/jira/browse/FLINK-16214 Project: Flink Issue Type: Sub-task Components: Documentation Reporter: Aljoscha Krettek This should go into {{stateful-stream-processing.md}}. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Created] (FLINK-16216) Describe end-to-end exactly once programs in stateful stream processing concepts documentation
Aljoscha Krettek created FLINK-16216: Summary: Describe end-to-end exactly once programs in stateful stream processing concepts documentation Key: FLINK-16216 URL: https://issues.apache.org/jira/browse/FLINK-16216 Project: Flink Issue Type: Sub-task Components: Documentation Reporter: Aljoscha Krettek This should go into {{stateful-stream-processing.md}}. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Created] (FLINK-16572) CheckPubSubEmulatorTest is flaky on Azure
Aljoscha Krettek created FLINK-16572: Summary: CheckPubSubEmulatorTest is flaky on Azure Key: FLINK-16572 URL: https://issues.apache.org/jira/browse/FLINK-16572 Project: Flink Issue Type: Bug Components: Build System / Azure Pipelines Reporter: Aljoscha Krettek Log: https://dev.azure.com/aljoschakrettek/Flink/_build/results?buildId=56&view=logs&j=1f3ed471-1849-5d3c-a34c-19792af4ad16&t=ce095137-3e3b-5f73-4b79-c42d3d5f8283&l=7842 -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Created] (FLINK-2891) Key for Keyed State is not set upon Window Evaluation
Aljoscha Krettek created FLINK-2891: --- Summary: Key for Keyed State is not set upon Window Evaluation Key: FLINK-2891 URL: https://issues.apache.org/jira/browse/FLINK-2891 Project: Flink Issue Type: Sub-task Components: Streaming Affects Versions: 0.10 Reporter: Aljoscha Krettek Assignee: Aljoscha Krettek Priority: Blocker In both the aligned and the general-purpose windows the key for the keyed operator state is not set when evaluating the windows. This silently leads to incorrect results. -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Created] (FLINK-2922) Add Queryable Window Operator
Aljoscha Krettek created FLINK-2922: --- Summary: Add Queryable Window Operator Key: FLINK-2922 URL: https://issues.apache.org/jira/browse/FLINK-2922 Project: Flink Issue Type: Improvement Components: Streaming Reporter: Aljoscha Krettek Assignee: Aljoscha Krettek The idea is to provide a window operator that allows to query the current window result at any time without discarding the current result. For example, a user might have an aggregation window operation with tumbling windows of 1 hour. Now, at any time they might be interested in the current aggregated value for the currently in-flight hour window. The idea is to make the operator a two input operator where normal elements arrive on input one while queries arrive on input two. The query stream must be keyed by the same key as the input stream. If an input arrives for a key the current value for that key is emitted along with the query element so that the user can map the result to the query. -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Created] (FLINK-2991) Extend Window Operators to Allow Efficient Fold Operation
Aljoscha Krettek created FLINK-2991: --- Summary: Extend Window Operators to Allow Efficient Fold Operation Key: FLINK-2991 URL: https://issues.apache.org/jira/browse/FLINK-2991 Project: Flink Issue Type: Improvement Components: Streaming Reporter: Aljoscha Krettek Right now, a window fold is implemented as a WindowFunction that gets all the elements as input. No pre-aggregation is performed. The window operator should be extended to also allow the fold to also be pre-aggregated. This requires changing the signature of the {{WindowBuffer}} so that it can emit a type other than the input type. -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Created] (FLINK-3024) TimestampExtractor Does not Work When returning Long.MIN_VALUE
Aljoscha Krettek created FLINK-3024: --- Summary: TimestampExtractor Does not Work When returning Long.MIN_VALUE Key: FLINK-3024 URL: https://issues.apache.org/jira/browse/FLINK-3024 Project: Flink Issue Type: Bug Components: Streaming Affects Versions: 0.10.0 Reporter: Aljoscha Krettek Assignee: Aljoscha Krettek The problem is that {{ExtractTimestampsOperator}} wrongly updates the {{currentWatermark}} in the {{trigger}} method. This will interfere with values returned from {{extractWatermark}}. -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Created] (FLINK-3174) Add merging WindowAssigner
Aljoscha Krettek created FLINK-3174: --- Summary: Add merging WindowAssigner Key: FLINK-3174 URL: https://issues.apache.org/jira/browse/FLINK-3174 Project: Flink Issue Type: Improvement Components: Streaming Reporter: Aljoscha Krettek Assignee: Aljoscha Krettek Fix For: 1.0.0 We should add the possibility for WindowAssigners to merge windows. This will enable Session windowing support, similar to how Google Cloud Dataflow supports. For session windows, each element would initially be assigned to its own window. When triggering we check the windows and see if any can be merged. This way, elements with overlapping session windows merge into one session. -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Created] (FLINK-3177) Add Stream Operator that Discards Late Elements
Aljoscha Krettek created FLINK-3177: --- Summary: Add Stream Operator that Discards Late Elements Key: FLINK-3177 URL: https://issues.apache.org/jira/browse/FLINK-3177 Project: Flink Issue Type: Improvement Components: Streaming Reporter: Aljoscha Krettek We should introduce an operator (and API call on DataStream) that allows users to discard late elements (with respect to their timestamps and the current watermark). The operator could have two modes, one for silently discarding elements that are late and one for failing the job if late elements arrive (as a sort of sanity check). -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Created] (FLINK-3178) Make Closing Behavior of Window Operators Configurable
Aljoscha Krettek created FLINK-3178: --- Summary: Make Closing Behavior of Window Operators Configurable Key: FLINK-3178 URL: https://issues.apache.org/jira/browse/FLINK-3178 Project: Flink Issue Type: Improvement Components: Streaming Reporter: Aljoscha Krettek Right now, all window operators emit (incomplete) in-flight windows when they are closing down. This happens for two reasons: 1. sources emit a final Long.MAX_VALUE watermark to flush out windows when they close. 2. window operators emit in-flight windows in their close() method For some users this might not be the desired behavior so we should either change it or make it configurable. -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Created] (FLINK-3200) Use Partitioned State Abstraction in WindowOperator
Aljoscha Krettek created FLINK-3200: --- Summary: Use Partitioned State Abstraction in WindowOperator Key: FLINK-3200 URL: https://issues.apache.org/jira/browse/FLINK-3200 Project: Flink Issue Type: Improvement Components: Streaming Affects Versions: 1.0.0 Reporter: Aljoscha Krettek Assignee: Aljoscha Krettek Right now, the WindowOperator uses snapshotState and restoreState to do custom serialization/deserialization of the windowing state. Moving this to partitioned state would allow us to re-scale in the future. Also, once the partitioned state has support to run on ManagedMemory/Out-of-core state the WindowOperator would also automatically benefit from this, allowing very large windows and large window state. The necessary steps are these: - Enhance state interface with new primitive states: ReducableState, ListState, etc.. Also make state scoped to a namespace. Namespaces are necessary because we need state that is scoped by key and by window (window would be the namespace in this case) - Enhance timer interface to also make timers scoped to a key/namespace - Change WindowOperator to use these new interfaces -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Created] (FLINK-3201) Enhance Partitioned State Interface with State Types
Aljoscha Krettek created FLINK-3201: --- Summary: Enhance Partitioned State Interface with State Types Key: FLINK-3201 URL: https://issues.apache.org/jira/browse/FLINK-3201 Project: Flink Issue Type: Sub-task Components: Streaming Affects Versions: 1.0.0 Reporter: Aljoscha Krettek Assignee: Aljoscha Krettek We should enhance the partitioned state with different state types, so that the system knows about the semantics of the State. I propose for now: - ValueState, this behaves like the current OperatorState: state is one value that can be set and retrieved - ListState, state is a list that can be appended to and iterated over - ReducingState, state is one value that other values can be added to ListState and ReducingState would share a common superclass to allow them to be used in the same places. For example, the WindowOperator would use ReducingState and ListState interchangeably, depending on whether we have a ReduceFunction or not. These additions allow the system to be clever about how state is checkpointed in the future. Think ManageMemory/Out-of-core state and incremental checkpoints. Also, state should be scoped to both a key and a namespace. This will allow the WindowOperator to use the interface. Right now, WindowOperator has a custom state implementation that uses a two-level Map (by key and by window). In this case the window would be the namespace. -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Created] (FLINK-3202) Make Timer/Trigger Service Scoped to Key/Namespace
Aljoscha Krettek created FLINK-3202: --- Summary: Make Timer/Trigger Service Scoped to Key/Namespace Key: FLINK-3202 URL: https://issues.apache.org/jira/browse/FLINK-3202 Project: Flink Issue Type: Sub-task Components: Streaming Affects Versions: 1.0.0 Reporter: Aljoscha Krettek Assignee: Aljoscha Krettek For the WindowOperator to be able to use the partitioned state abstraction timers/triggers also need to be scoped to the same key/namespace so that firing triggers can be matched to the correct window. -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Created] (FLINK-3242) User-specified StateBackend is not Respected if Checkpointing is Disabled
Aljoscha Krettek created FLINK-3242: --- Summary: User-specified StateBackend is not Respected if Checkpointing is Disabled Key: FLINK-3242 URL: https://issues.apache.org/jira/browse/FLINK-3242 Project: Flink Issue Type: Bug Components: Streaming Affects Versions: 1.0.0 Reporter: Aljoscha Krettek Priority: Blocker In {{StreamingJobGraphGenerator.java}} (line 281) the StateBackend is only set on the {{StreamConfig}} if checkpointing is enabled. This is easy to fix but we also need to add test to prevent this from happening in the future. -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Created] (FLINK-3243) Fix Interplay of TimeCharacteristic and Time Windows
Aljoscha Krettek created FLINK-3243: --- Summary: Fix Interplay of TimeCharacteristic and Time Windows Key: FLINK-3243 URL: https://issues.apache.org/jira/browse/FLINK-3243 Project: Flink Issue Type: Bug Components: Streaming Affects Versions: 1.0.0 Reporter: Aljoscha Krettek Assignee: Aljoscha Krettek Priority: Blocker As per the discussion on the Dev ML: http://apache-flink-mailing-list-archive.1008284.n3.nabble.com/DISCUSS-Time-Behavior-in-Streaming-Jobs-Event-time-processing-time-td9616.html. The discussion seems to have converged on option 2): - Add dedicated WindowAssigners for processing time and event time - {{timeWindow()}} and {{timeWindowAll()}} respect the set {{TimeCharacteristic}}. This will make the easy stuff easy, i.e. using time windows and quickly switching the time characteristic. Users will then have the flexibility to mix different kinds of window assigners in their job. -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Created] (FLINK-3278) Add Partitioned State Backend Based on RocksDB
Aljoscha Krettek created FLINK-3278: --- Summary: Add Partitioned State Backend Based on RocksDB Key: FLINK-3278 URL: https://issues.apache.org/jira/browse/FLINK-3278 Project: Flink Issue Type: Improvement Components: Streaming Reporter: Aljoscha Krettek Assignee: Aljoscha Krettek This would allow users to have state that can grow larger than the available memory. Once WindowOperator is based on the partitioned state abstraction (FLINK-3200) this will also allow windows to grow larger than the available memory. -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Created] (FLINK-3315) Fix Slot Sharing in Streaming API
Aljoscha Krettek created FLINK-3315: --- Summary: Fix Slot Sharing in Streaming API Key: FLINK-3315 URL: https://issues.apache.org/jira/browse/FLINK-3315 Project: Flink Issue Type: Improvement Components: Streaming Affects Versions: 1.0.0 Reporter: Aljoscha Krettek Assignee: Aljoscha Krettek Right now, the slot sharing/resource group logic is a bit "nebulous". The slot sharing group that operators are put in depends on the order in which operations are created. For example, in this case: {code} Source a = env.source() Source b = env.source() a.map().startNewResourceGroup().sink() b.map().sink() {code} We end up with two resource groups: - group 1: source a - group 2: map(), sink(), source b, map(), sink() The reason is that the slot sharing id is incremented when transforming the {{startNewResouceGroup()}} call and all operators that are transformed afterwards in graph traversal get that new slot sharing id. (There is also {{isolateResources()}} which can be used to isolate an operator.) What I propose is to remove {{startNewResourceGroup()}} and {{isolateResouces()}} and replace it with {{slotSharingGroup(String)}}. By default, operations would be in slot sharing group "default". This allows very fine grained control over what operators end up in which slot sharing group. For example, I could have this topology: {code} Source a = env.source().slotSharingGroup("sources") Source b = env.source().slotSharingGroup("sources") a.map().slotSharingGroup("heavy a").sink().slotSharingGroup("sinks") b.map().slotSharingGroup("heavy b").sink().slotSharingGroup("sinks") {code} Which would isolate the lightweight sources and sinks in a group and put heavy operations inside their own slot groups. This is a bit more low level than the previous API and requires more calls than a simple {{startNewResourceGroup()}} but I think not many people would use this feature and this design makes it very clear what operations end up in the same group. -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Created] (FLINK-3336) Add Semi-Rebalance Data Shipping for DataStream
Aljoscha Krettek created FLINK-3336: --- Summary: Add Semi-Rebalance Data Shipping for DataStream Key: FLINK-3336 URL: https://issues.apache.org/jira/browse/FLINK-3336 Project: Flink Issue Type: Improvement Components: Streaming Reporter: Aljoscha Krettek Assignee: Aljoscha Krettek Fix For: 1.0.0 This feature has recently been requested on the ML: http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/Distribution-of-sinks-among-the-nodes-td4640.html The new data shipping pattern would allow to rebalance data only to a subset of downstream operations. The subset of downstream operations to which the upstream operation would send elements depends on the degree of parallelism of both the upstream and downstream operation. For example, if the upstream operation has parallelism 2 and the downstream operation has parallelism 4, then one upstream operation would distribute elements to two downstream operations while the other upstream operation would distribute to the other two downstream operations. If, on the other hand, the downstream operation had parallelism 2 while the upstream operation has parallelism 4 then two upstream operations would distribute to one downstream operation while the other two upstream operations would distribute to the other downstream operations. In cases where the different parallelisms are not multiples of each other one or several downstream operations would have a differing number of inputs from upstream operations. -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Created] (FLINK-3400) RocksDB Backend does not work when not in Flink lib folder
Aljoscha Krettek created FLINK-3400: --- Summary: RocksDB Backend does not work when not in Flink lib folder Key: FLINK-3400 URL: https://issues.apache.org/jira/browse/FLINK-3400 Project: Flink Issue Type: Bug Components: Streaming Affects Versions: 1.0.0 Reporter: Aljoscha Krettek Assignee: Aljoscha Krettek Priority: Blocker Fix For: 1.0.0 The problem is that the external process runner cannot load the {{HDFSCopyFromLocal}} and {{HDFSCopyToLocal}} classes when the rocksdb backend jar is not in the Flink lib folder. If rocksdb is only in the user-code jar the code cannot be loaded when trying to start the external copy runners. -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Created] (FLINK-3402) Refactor Common Parts of Stream/Batch Documentation
Aljoscha Krettek created FLINK-3402: --- Summary: Refactor Common Parts of Stream/Batch Documentation Key: FLINK-3402 URL: https://issues.apache.org/jira/browse/FLINK-3402 Project: Flink Issue Type: Improvement Components: Documentation Reporter: Aljoscha Krettek Assignee: Aljoscha Krettek I want to build on the work of [~uce] in refactoring the streaming guide. With the release on the horizon I think it is important to have good structure in the documentation. I propose to move the following sections from the Streaming Doc to a new Section "Basic Concepts" (name up for discussion): - Linking With Flink - DataStream abstraction (remove, this can be covered by an extended "Lazy Evaluation") - Lazy Evaluation - Specifying Keys - Passing Functions to Flink - Data Types - Debugging - Program Packaging and Distributed Execution - Parallel Execution - Execution Plans -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Created] (FLINK-3403) Create Section "Working with Time" in Streaming Guide
Aljoscha Krettek created FLINK-3403: --- Summary: Create Section "Working with Time" in Streaming Guide Key: FLINK-3403 URL: https://issues.apache.org/jira/browse/FLINK-3403 Project: Flink Issue Type: Improvement Components: Documentation Reporter: Aljoscha Krettek Assignee: Aljoscha Krettek We should add a proper page for this. Where we explain the notions of time and how to setup programs. Also, we would explain how to work with watermarks and the different timestamp extractors. -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Created] (FLINK-3417) Add RocksDB StateBackendFactory and integrate with Flink Config
Aljoscha Krettek created FLINK-3417: --- Summary: Add RocksDB StateBackendFactory and integrate with Flink Config Key: FLINK-3417 URL: https://issues.apache.org/jira/browse/FLINK-3417 Project: Flink Issue Type: Improvement Components: Streaming Affects Versions: 1.0.0 Reporter: Aljoscha Krettek Priority: Blocker Currently, only memory and file backends can be configured as default backends in {{flink-conf.yaml}}. To change that we need to add a StateBackendFactory. Also, there cannot be a short name for this such as how you can now specify "jobmanager" or "filesystem" in the config because RocksDB is not in the standard dist. -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Created] (FLINK-3419) Drop partitionByHash from DataStream
Aljoscha Krettek created FLINK-3419: --- Summary: Drop partitionByHash from DataStream Key: FLINK-3419 URL: https://issues.apache.org/jira/browse/FLINK-3419 Project: Flink Issue Type: Improvement Components: Streaming Affects Versions: 1.0.0 Reporter: Aljoscha Krettek Priority: Blocker The behavior is no different from {{keyBy}}, except that you cannot use keyed state and windows if you use {{partitionByHash}} so I suggest to drop it. We might also want to think about dropping {{shuffle}} and {{global}}. -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Created] (FLINK-3430) Remove "no POJO" warning in TypeAnalyzer
Aljoscha Krettek created FLINK-3430: --- Summary: Remove "no POJO" warning in TypeAnalyzer Key: FLINK-3430 URL: https://issues.apache.org/jira/browse/FLINK-3430 Project: Flink Issue Type: Improvement Components: Scala API Reporter: Aljoscha Krettek Assignee: Aljoscha Krettek Right now, the TypeAnalyzer creates compilation warnings when a type cannot be analyzed as a POJO. This could confuse some users so we should remove it. -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Created] (FLINK-3436) Remove ComplexIntegrationITCase
Aljoscha Krettek created FLINK-3436: --- Summary: Remove ComplexIntegrationITCase Key: FLINK-3436 URL: https://issues.apache.org/jira/browse/FLINK-3436 Project: Flink Issue Type: Improvement Components: Streaming Reporter: Aljoscha Krettek Assignee: Aljoscha Krettek Most of the tests are disabled and for the rest I am not exactly sure what is being tested. Most stuff should be covered by other, specific, tests. -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Created] (FLINK-3439) Remove final Long.MAX_VALUE Watermark in StreamSource
Aljoscha Krettek created FLINK-3439: --- Summary: Remove final Long.MAX_VALUE Watermark in StreamSource Key: FLINK-3439 URL: https://issues.apache.org/jira/browse/FLINK-3439 Project: Flink Issue Type: Bug Components: Streaming Reporter: Aljoscha Krettek Assignee: Aljoscha Krettek I think this was an oversight. At one point we decided to not emit in-flight windows when closing the topology or sources. -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Created] (FLINK-3450) RocksDB Backed Window Fails with KryoSerializer
Aljoscha Krettek created FLINK-3450: --- Summary: RocksDB Backed Window Fails with KryoSerializer Key: FLINK-3450 URL: https://issues.apache.org/jira/browse/FLINK-3450 Project: Flink Issue Type: Bug Components: Streaming Affects Versions: 1.0.0 Reporter: Aljoscha Krettek Assignee: Aljoscha Krettek Priority: Blocker Fix For: 1.0.0 It fails due to concurrent use of the KryoSerializer when asynchronously serializing the {{StateDescriptor}} during checkpointing. -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Created] (FLINK-3512) Savepoint backend should not revert to "jobmanager"
Aljoscha Krettek created FLINK-3512: --- Summary: Savepoint backend should not revert to "jobmanager" Key: FLINK-3512 URL: https://issues.apache.org/jira/browse/FLINK-3512 Project: Flink Issue Type: Bug Reporter: Aljoscha Krettek Assignee: Ufuk Celebi Priority: Blocker When not setting a state backend in the config the savepoint coordinator defaults to use "jobmanager" as savepoint backend, ignoring the setting for the savepoint backend in the config. This means that when setting a StateBackend only per job the system always falls back to "jobmanager" even though it could use "filesystem" to make savepoints persistent. See SavepointStoreFactory.java:71 -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Created] (FLINK-3513) Fix interplay of automatic Operator UID and Changing name of WindowOperator
Aljoscha Krettek created FLINK-3513: --- Summary: Fix interplay of automatic Operator UID and Changing name of WindowOperator Key: FLINK-3513 URL: https://issues.apache.org/jira/browse/FLINK-3513 Project: Flink Issue Type: Bug Components: Streaming Affects Versions: 1.0.0 Reporter: Aljoscha Krettek Priority: Blocker WindowOperator can have a changing name because it has the TypeSerializer .toString() output in it's name. For some type serializers that don't implement toString() this means that the name changes. This means that savepoint restore does not work for the automatically generated UID. -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Created] (FLINK-3521) Make Iterable part of method signature for WindowFunction
Aljoscha Krettek created FLINK-3521: --- Summary: Make Iterable part of method signature for WindowFunction Key: FLINK-3521 URL: https://issues.apache.org/jira/browse/FLINK-3521 Project: Flink Issue Type: Improvement Components: Streaming Reporter: Aljoscha Krettek Assignee: Aljoscha Krettek Priority: Blocker After discussion with [~StephanEwen],[~rmetzger], [~fhueske] it was decided that the signature of WindowFunction should be changed. Right now, the signature is {{WindowFunction}}. With the decision of whether the input is an {{Iterable}} nor not residing in the signature of the method on {{WindowedStream}}, for example: - {{WindowedStream.apply(WindowFunction>, OUT, KEY, WINDOW>}} - {{WindowedStream.apply(ReduceFunction, WindowFunction}} The {{Iterable<>}} should be moved into the signature of the {{apply}} method, so that it is: {{void apply(KEY key, W window, Iterable input, Collector out)}} -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Created] (FLINK-3527) Scala DataStream has no transform method
Aljoscha Krettek created FLINK-3527: --- Summary: Scala DataStream has no transform method Key: FLINK-3527 URL: https://issues.apache.org/jira/browse/FLINK-3527 Project: Flink Issue Type: Improvement Components: Streaming Reporter: Aljoscha Krettek Assignee: Aljoscha Krettek Priority: Blocker Some people might want to write custom operators and use them in the Scala API. Not having the transform in there directly makes it cumbersome for them to use custom operators. -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Created] (FLINK-3528) Add Incremental Fold for Non-Keyed Window Operator
Aljoscha Krettek created FLINK-3528: --- Summary: Add Incremental Fold for Non-Keyed Window Operator Key: FLINK-3528 URL: https://issues.apache.org/jira/browse/FLINK-3528 Project: Flink Issue Type: Improvement Components: Streaming Reporter: Aljoscha Krettek Assignee: Aljoscha Krettek Priority: Blocker Right now, non-keyed fold does not have constant space requirements. -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Created] (FLINK-3548) Remove unnecessary generic parameter from SingleOutputStreamOperator
Aljoscha Krettek created FLINK-3548: --- Summary: Remove unnecessary generic parameter from SingleOutputStreamOperator Key: FLINK-3548 URL: https://issues.apache.org/jira/browse/FLINK-3548 Project: Flink Issue Type: Improvement Components: Streaming Reporter: Aljoscha Krettek Assignee: Aljoscha Krettek -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Created] (FLINK-3565) FlinkKafkaConsumer does not work with Scala 2.11
Aljoscha Krettek created FLINK-3565: --- Summary: FlinkKafkaConsumer does not work with Scala 2.11 Key: FLINK-3565 URL: https://issues.apache.org/jira/browse/FLINK-3565 Project: Flink Issue Type: Bug Components: Streaming Affects Versions: 1.0.0 Reporter: Aljoscha Krettek Priority: Blocker Running a program built against Flink_2.11 dependencies fails on a Flink_2.11 cluster with this exception: java.lang.NoClassDefFoundError: scala/collection/GenTraversableOnce$class at kafka.utils.Pool.(Pool.scala:28) at kafka.consumer.FetchRequestAndResponseStatsRegistry$.(FetchRequestAndResponseStats.scala:60) at kafka.consumer.FetchRequestAndResponseStatsRegistry$.(FetchRequestAndResponseStats.scala) at kafka.consumer.SimpleConsumer.(SimpleConsumer.scala:39) at kafka.javaapi.consumer.SimpleConsumer.(SimpleConsumer.scala:34) at org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer08.getPartitionsForTopic(FlinkKafkaConsumer08.java:518) at org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer08.(FlinkKafkaConsumer08.java:218) at org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer08.(FlinkKafkaConsumer08.java:193) at org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer08.(FlinkKafkaConsumer08.java:160) at com.dataartisans.querywindow.WindowJob.main(WindowJob.java:93) at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57) at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) at java.lang.reflect.Method.invoke(Method.java:606) at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:505) at org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:403) at org.apache.flink.client.program.Client.runBlocking(Client.java:248) at org.apache.flink.client.CliFrontend.executeProgramBlocking(CliFrontend.java:866) at org.apache.flink.client.CliFrontend.run(CliFrontend.java:333) at org.apache.flink.client.CliFrontend.parseParameters(CliFrontend.java:1189) at org.apache.flink.client.CliFrontend.main(CliFrontend.java:1239) Caused by: java.lang.ClassNotFoundException: scala.collection.GenTraversableOnce$class at java.net.URLClassLoader$1.run(URLClassLoader.java:366) at java.net.URLClassLoader$1.run(URLClassLoader.java:355) at java.security.AccessController.doPrivileged(Native Method) at java.net.URLClassLoader.findClass(URLClassLoader.java:354) at java.lang.ClassLoader.loadClass(ClassLoader.java:425) at java.lang.ClassLoader.loadClass(ClassLoader.java:358) ... 21 more -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Created] (FLINK-3575) Update Working With State Section in Doc
Aljoscha Krettek created FLINK-3575: --- Summary: Update Working With State Section in Doc Key: FLINK-3575 URL: https://issues.apache.org/jira/browse/FLINK-3575 Project: Flink Issue Type: Improvement Components: Documentation Reporter: Aljoscha Krettek Assignee: Aljoscha Krettek We need to update it to match the new Key/Value state interface. -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Created] (FLINK-3581) Add Non-Keyed Window Trigger
Aljoscha Krettek created FLINK-3581: --- Summary: Add Non-Keyed Window Trigger Key: FLINK-3581 URL: https://issues.apache.org/jira/browse/FLINK-3581 Project: Flink Issue Type: Improvement Components: Streaming Reporter: Aljoscha Krettek Assignee: Aljoscha Krettek The current Window Trigger is per key. Meaning every window has a (logical) Trigger for every key in the window, i.e. there will be state and time triggers per key per window. For some types of windows, i.e. based on time it is possible to use a single Trigger to fire for all keys at the same time. In that case we would save a lot of space on state and timers. Which makes state snapshots a lot smaller. -- This message was sent by Atlassian JIRA (v6.3.4#6332)