[GitHub] [flink] wsry commented on pull request #13595: [FLINK-19582][network] Introduce sort-merge based blocking shuffle to Flink
wsry commented on pull request #13595: URL: https://github.com/apache/flink/pull/13595#issuecomment-716321736 @StephanEwen @zhijiangW @gaoyunhaii Thanks for the review and comments, I have updated the PR according to the review comments. This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[jira] [Updated] (FLINK-17372) SlotManager should expose total required resources
[
https://issues.apache.org/jira/browse/FLINK-17372?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Xintong Song updated FLINK-17372:
-
Fix Version/s: (was: 1.12.0)
1.13.0
> SlotManager should expose total required resources
> --
>
> Key: FLINK-17372
> URL: https://issues.apache.org/jira/browse/FLINK-17372
> Project: Flink
> Issue Type: Improvement
> Components: Runtime / Coordination
>Affects Versions: 1.11.0
>Reporter: Till Rohrmann
>Priority: Major
> Fix For: 1.13.0
>
>
> Currently, the {{SlotManager}} exposes the set of required resources which
> have not been fulfilled via {{SlotManager.getRequiredResources}}. The idea of
> this function is to allow the {{ResourceManager}} to decide whether new
> pods/containers need to be started or not.
> The problem is that once a resource has been registered at the
> {{SlotManager}} it will decrease the set of required resources. If now a
> pod/container fails, then the {{ResourceManager}} won't know whether it needs
> to restart the container or not.
> In order to simplify the interaction, I propose to let the {{SlotManager}}
> announce all of its required resources (pending + registered resources). That
> way the {{ResourceManager}} only needs to compare the set of required
> resources with the set of pending and allocated containers/pods.
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[jira] [Updated] (FLINK-13554) ResourceManager should have a timeout on starting new TaskExecutors.
[ https://issues.apache.org/jira/browse/FLINK-13554?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Xintong Song updated FLINK-13554: - Fix Version/s: (was: 1.12.0) 1.13.0 > ResourceManager should have a timeout on starting new TaskExecutors. > > > Key: FLINK-13554 > URL: https://issues.apache.org/jira/browse/FLINK-13554 > Project: Flink > Issue Type: Improvement > Components: Runtime / Coordination >Affects Versions: 1.9.0 >Reporter: Xintong Song >Priority: Critical > Fix For: 1.13.0 > > > Recently, we encountered a case that one TaskExecutor get stuck during > launching on Yarn (without fail), causing that job cannot recover from > continuous failovers. > The reason the TaskExecutor gets stuck is due to our environment problem. The > TaskExecutor gets stuck somewhere after the ResourceManager starts the > TaskExecutor and waiting for the TaskExecutor to be brought up and register. > Later when the slot request timeouts, the job fails over and requests slots > from ResourceManager again, the ResourceManager still see a TaskExecutor (the > stuck one) is being started and will not request new container from Yarn. > Therefore, the job can not recover from failure. > I think to avoid such unrecoverable status, the ResourceManager need to have > a timeout on starting new TaskExecutor. If the starting of TaskExecutor takes > too long, it should just fail the TaskExecutor and starts a new one. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Updated] (FLINK-18229) Pending worker requests should be properly cleared
[
https://issues.apache.org/jira/browse/FLINK-18229?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Xintong Song updated FLINK-18229:
-
Fix Version/s: (was: 1.12.0)
1.13.0
> Pending worker requests should be properly cleared
> --
>
> Key: FLINK-18229
> URL: https://issues.apache.org/jira/browse/FLINK-18229
> Project: Flink
> Issue Type: Improvement
> Components: Deployment / Kubernetes, Deployment / YARN, Runtime /
> Coordination
>Affects Versions: 1.9.3, 1.10.1, 1.11.0
>Reporter: Xintong Song
>Priority: Major
> Fix For: 1.13.0
>
>
> Currently, if Kubernetes/Yarn does not have enough resources to fulfill
> Flink's resource requirement, there will be pending pod/container requests on
> Kubernetes/Yarn. These pending resource requirements are never cleared until
> either fulfilled or the Flink cluster is shutdown.
> However, sometimes Flink no longer needs the pending resources. E.g., the
> slot request is then fulfilled by another slots that become available, or the
> job failed due to slot request timeout (in a session cluster). In such cases,
> Flink does not remove the resource request until the resource is allocated,
> then it discovers that it no longer needs the allocated resource and release
> them. This would affect the underlying Kubernetes/Yarn cluster, especially
> when the cluster is under heavy workload.
> It would be good for Flink to cancel pod/container requests as earlier as
> possible if it can discover that some of the pending workers are no longer
> needed.
> There are several approaches potentially achieve this.
> # We can always check whether there's a pending worker that can be canceled
> when a \{{PendingTaskManagerSlot}} is unassigned.
> # We can have a separate timeout for requesting new worker. If the resource
> cannot be allocated within the given time since requested, we should cancel
> that resource request and claim a resource allocation failure.
> # We can share the same timeout for starting new worker (proposed in
> FLINK-13554). This is similar to 2), but it requires the worker to be
> registered, rather than allocated, before timeout.
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[GitHub] [flink] flinkbot edited a comment on pull request #13767: [FLINK-19787][table-runtime] Migrate Filesystem connector to new table source sink interface
flinkbot edited a comment on pull request #13767: URL: https://github.com/apache/flink/pull/13767#issuecomment-715256450 ## CI report: * 11330173d2fcaeead16e067e1f7d31c50081c2e1 UNKNOWN * c532d886b68b7dd265c9056945a3cfc94e8352b7 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8218) * 0b644cc4e0ffb153ecdd67e8d4ba214341dc1a06 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8260) * 2979dcd1fa92a3a5653914203311aba30f62ad09 UNKNOWN Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build - `@flinkbot run azure` re-run the last Azure build This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13771: [FLINK-19789][hive] Migrate Hive connector to new table source sink interface
flinkbot edited a comment on pull request #13771: URL: https://github.com/apache/flink/pull/13771#issuecomment-715323930 ## CI report: * e5442954ffac19a92ba1f9a4620584e61acb601a Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8219) * 7cb29bdf576ab721181dcf526e0ec1575971ee33 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8264) * 625bb55bba2c6febed70a805ffddef883b736dbc UNKNOWN Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build - `@flinkbot run azure` re-run the last Azure build This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13677: Single task scheduler
flinkbot edited a comment on pull request #13677: URL: https://github.com/apache/flink/pull/13677#issuecomment-711161777 ## CI report: * b74b5ddf6a01dcb080cf5cf2e6773d919a6ecdec Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8111) * dd609bc6af031d863304573763df73555ee6f817 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8266) Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build - `@flinkbot run azure` re-run the last Azure build This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13664: [FLINK-19673] Translate "Standalone Cluster" page into Chinese
flinkbot edited a comment on pull request #13664: URL: https://github.com/apache/flink/pull/13664#issuecomment-710017891 ## CI report: * 9be74ce0ce0d57bbda859ca9f5ec7d226df277b1 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7913) * 87c6e10a3e55796d1848b0ccfe3f5da1d48deae5 UNKNOWN Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build - `@flinkbot run azure` re-run the last Azure build This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [flink] Shawn-Hx commented on pull request #13664: [FLINK-19673] Translate "Standalone Cluster" page into Chinese
Shawn-Hx commented on pull request #13664: URL: https://github.com/apache/flink/pull/13664#issuecomment-716314480 Hi, @klion26 I have made some changes according to your advice. Also leave some comments about the last advice. Please take a look. Thanks. This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [flink] zhuzhurk commented on a change in pull request #13730: [FLINK-19552][coordination] Consider only available input location preferences for slot profile in pipelined region scheduling
zhuzhurk commented on a change in pull request #13730:
URL: https://github.com/apache/flink/pull/13730#discussion_r511720650
##
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/SlotSharingExecutionSlotAllocator.java
##
@@ -212,20 +214,24 @@ private ResourceProfile
getPhysicalSlotResourceProfile(ExecutionSlotSharingGroup
.reduce(ResourceProfile.ZERO, (r, e) ->
r.merge(resourceProfileRetriever.apply(e)), ResourceProfile::merge);
}
- private SharingPhysicalSlotRequestBulk
createBulk(Map> executions) {
- Map pendingRequests
= executions
- .keySet()
- .stream()
- .collect(Collectors.toMap(
- group -> group,
- group ->
sharedSlots.get(group).getPhysicalSlotResourceProfile()
- ));
+ private Optional createBulk(
+ Map>
executions) {
+ Map pendingRequests
= new HashMap<>();
+ for (ExecutionSlotSharingGroup group : executions.keySet()) {
+ SharedSlot sharedSlot = sharedSlots.get(group);
+ if (sharedSlot == null ||
sharedSlot.getSlotContextFuture().isCompletedExceptionally()) {
Review comment:
I think this case is not possible to happen.
If so it can happen only when there is a bug, can we just throw exception to
expose the bug?
Correct me if I'm wrong.
##
File path:
flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/DefaultSyncPreferredLocationsRetrieverTest.java
##
@@ -0,0 +1,77 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.scheduler;
+
+import org.apache.flink.runtime.scheduler.strategy.ExecutionVertexID;
+import org.apache.flink.runtime.taskmanager.TaskManagerLocation;
+
+import org.junit.Test;
+
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Optional;
+
+import static
org.apache.flink.runtime.executiongraph.ExecutionGraphTestUtils.createRandomExecutionVertexId;
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.Matchers.hasSize;
+import static org.junit.Assert.assertThat;
+
+/**
+ * Tests for {@link DefaultSyncPreferredLocationsRetriever}.
+ */
+public class DefaultSyncPreferredLocationsRetrieverTest {
Review comment:
should extend TestLogger
##
File path:
flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/DefaultSyncPreferredLocationsRetrieverTest.java
##
@@ -0,0 +1,77 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.scheduler;
+
+import org.apache.flink.runtime.scheduler.strategy.ExecutionVertexID;
+import org.apache.flink.runtime.taskmanager.TaskManagerLocation;
+
+import org.junit.Test;
+
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Optional;
+
+import static
org.apache.flink.runtime.executiongraph.ExecutionGraphTestUtils.createRandomExecutionVertexId;
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.Matchers.hasSize;
+import static org.junit.Assert.assertThat;
+
+/**
+ * Tests for {@link DefaultSyncPreferredLocationsRetriever}.
+ */
+public class DefaultSyncPreferredLocationsRetrieverTest {
+ private static final ExecutionVertexID EV1 =
[GitHub] [flink] flinkbot edited a comment on pull request #13789: [FLINK-19727][table-runtime] Implement ParallelismProvider for sink i…
flinkbot edited a comment on pull request #13789: URL: https://github.com/apache/flink/pull/13789#issuecomment-716299701 ## CI report: * 6329930d7acc5d2d4d84e777561e79bd25ac9367 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8265) Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build - `@flinkbot run azure` re-run the last Azure build This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13771: [FLINK-19789][hive] Migrate Hive connector to new table source sink interface
flinkbot edited a comment on pull request #13771: URL: https://github.com/apache/flink/pull/13771#issuecomment-715323930 ## CI report: * e5442954ffac19a92ba1f9a4620584e61acb601a Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8219) * 7cb29bdf576ab721181dcf526e0ec1575971ee33 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8264) Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build - `@flinkbot run azure` re-run the last Azure build This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13766: [FLINK-19703][runtime] Untrack a result partition if its producer task failed in TaskManager
flinkbot edited a comment on pull request #13766: URL: https://github.com/apache/flink/pull/13766#issuecomment-715255564 ## CI report: * 42f066384d04bcfe67ccbb5766f09ab5dde9e19c Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8175) * b46f97dd6b1fa67b4dd169867c1967877ac8534c Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8263) Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build - `@flinkbot run azure` re-run the last Azure build This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13744: [FLINK-19766][table-runtime] Introduce File streaming compaction operators
flinkbot edited a comment on pull request #13744: URL: https://github.com/apache/flink/pull/13744#issuecomment-714369975 ## CI report: * 4d7e601658c0dacc100c572f4fe8878c058e035f Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8156) * 074ab07749ff798563e785d2c2fafeee1302ac74 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8262) Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build - `@flinkbot run azure` re-run the last Azure build This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13677: Single task scheduler
flinkbot edited a comment on pull request #13677: URL: https://github.com/apache/flink/pull/13677#issuecomment-711161777 ## CI report: * b74b5ddf6a01dcb080cf5cf2e6773d919a6ecdec Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8111) * dd609bc6af031d863304573763df73555ee6f817 UNKNOWN Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build - `@flinkbot run azure` re-run the last Azure build This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [flink] Shawn-Hx commented on a change in pull request #13664: [FLINK-19673] Translate "Standalone Cluster" page into Chinese
Shawn-Hx commented on a change in pull request #13664:
URL: https://github.com/apache/flink/pull/13664#discussion_r511723364
##
File path: docs/ops/deployment/cluster_setup.zh.md
##
@@ -100,52 +105,60 @@ configuration files (which need to be accessible at the
same path on all machine
-The Flink directory must be available on every worker under the same path. You
can use a shared NFS directory, or copy the entire Flink directory to every
worker node.
+Flink 目录必须放在所有 worker 节点的相同目录下。你可以使用共享的 NFS 目录，或将 Flink 目录复制到每个 worker 节点上。
-Please see the [configuration page](../config.html) for details and additional
configuration options.
+请参考 [配置参数页面]({% link ops/config.zh.md %}) 获取更多细节以及额外的配置项。
-In particular,
+特别地，
- * the amount of available memory per JobManager
(`jobmanager.memory.process.size`),
- * the amount of available memory per TaskManager
(`taskmanager.memory.process.size` and check [memory setup
guide](../memory/mem_tuning.html#configure-memory-for-standalone-deployment)),
- * the number of available CPUs per machine (`taskmanager.numberOfTaskSlots`),
- * the total number of CPUs in the cluster (`parallelism.default`) and
- * the temporary directories (`io.tmp.dirs`)
+* 每个 JobManager 的可用内存值（`jobmanager.memory.process.size`），
+* 每个 TaskManager 的可用内存值 （`taskmanager.memory.process.size`，并检查 [内存调优指南]({%
link ops/memory/mem_tuning.zh.md
%}#configure-memory-for-standalone-deployment)），
+* 每台机器的可用 CPU 数（`taskmanager.numberOfTaskSlots`），
+* 集群中所有 CPU 数（`parallelism.default`）和
+* 临时目录（`io.tmp.dirs`）
-are very important configuration values.
+的值都是非常重要的配置项。
{% top %}
-### Starting Flink
+
-The following script starts a JobManager on the local node and connects via
SSH to all worker nodes listed in the *workers* file to start the TaskManager
on each node. Now your Flink system is up and running. The JobManager running
on the local node will now accept jobs at the configured RPC port.
+### 启动 Flink
-Assuming that you are on the master node and inside the Flink directory:
+下面的脚本在本地节点启动了一个 JobManager 并通过 SSH 连接到 *workers* 文件中所有的 worker 节点，在每个节点上启动
TaskManager。现在你的 Flink 系统已经启动并运行着。在本地节点上运行的 JobManager 会在配置的 RPC 端口上接收作业。
Review comment:
这里的原文为 “The JobManager running on the local node will now accept jobs at
the configured RPC port.”
是否可以翻译成：“在本地节点上运行的 JobManager 会通过配置的 RPC 端口接收提交的作业。”
或者意译为：“可以通过配置的 RPC 端口向本地节点上的 JobManager 提交作业。”
这两种哪个更好一点?
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [flink] klion26 commented on a change in pull request #13664: [FLINK-19673] Translate "Standalone Cluster" page into Chinese
klion26 commented on a change in pull request #13664:
URL: https://github.com/apache/flink/pull/13664#discussion_r511322491
##
File path: docs/ops/deployment/cluster_setup.zh.md
##
@@ -22,65 +22,70 @@ specific language governing permissions and limitations
under the License.
-->
-This page provides instructions on how to run Flink in a *fully distributed
fashion* on a *static* (but possibly heterogeneous) cluster.
+本页面提供了关于如何在*静态*（但可能异构）集群上以*完全分布式方式*运行 Flink 的说明。
* This will be replaced by the TOC
{:toc}
-## Requirements
+
-### Software Requirements
+## 需求
-Flink runs on all *UNIX-like environments*, e.g. **Linux**, **Mac OS X**, and
**Cygwin** (for Windows) and expects the cluster to consist of **one master
node** and **one or more worker nodes**. Before you start to setup the system,
make sure you have the following software installed **on each node**:
+
-- **Java 1.8.x** or higher,
-- **ssh** (sshd must be running to use the Flink scripts that manage
- remote components)
+### 软件需求
-If your cluster does not fulfill these software requirements you will need to
install/upgrade it.
+Flink 运行在所有*类 UNIX 环境*下，例如 **Linux**，**Mac OS X** 和 **Cygwin**
（Windows），并且认为集群由**一个 master 节点**以及**一个或多个 worker
节点**构成。在配置系统之前，请确保**在每个节点上**安装有以下软件：
-Having __passwordless SSH__ and
-__the same directory structure__ on all your cluster nodes will allow you to
use our scripts to control
-everything.
+- **Java 1.8.x** 或更高版本，
+- **ssh** （必须运行 sshd 以使用 Flink 脚本管理远程组件）
Review comment:
这里翻译成 `必须运行 sshd 以执行用于管理管理 Flink 各组件的脚本` 会好一些吗？
##
File path: docs/ops/deployment/cluster_setup.md
##
@@ -80,7 +80,7 @@ configuration files (which need to be accessible at the same
path on all machine
-
+
Review comment:
我建议 cluster_setup.md 的单独放到一个 hotfix pr 或者至少放到一个单独的 commit 中，因为这个和翻译是无关的。
##
File path: docs/ops/deployment/cluster_setup.zh.md
##
@@ -22,65 +22,70 @@ specific language governing permissions and limitations
under the License.
-->
-This page provides instructions on how to run Flink in a *fully distributed
fashion* on a *static* (but possibly heterogeneous) cluster.
+本页面提供了关于如何在*静态*（但可能异构）集群上以*完全分布式方式*运行 Flink 的说明。
* This will be replaced by the TOC
{:toc}
-## Requirements
+
-### Software Requirements
+## 需求
-Flink runs on all *UNIX-like environments*, e.g. **Linux**, **Mac OS X**, and
**Cygwin** (for Windows) and expects the cluster to consist of **one master
node** and **one or more worker nodes**. Before you start to setup the system,
make sure you have the following software installed **on each node**:
+
-- **Java 1.8.x** or higher,
-- **ssh** (sshd must be running to use the Flink scripts that manage
- remote components)
+### 软件需求
-If your cluster does not fulfill these software requirements you will need to
install/upgrade it.
+Flink 运行在所有*类 UNIX 环境*下，例如 **Linux**，**Mac OS X** 和 **Cygwin**
（Windows），并且认为集群由**一个 master 节点**以及**一个或多个 worker
节点**构成。在配置系统之前，请确保**在每个节点上**安装有以下软件：
Review comment:
`并且认为` 这里能否去掉呢？直接翻译成 `集群xx` 或者能否优化一下呢？
##
File path: docs/ops/deployment/cluster_setup.zh.md
##
@@ -22,65 +22,70 @@ specific language governing permissions and limitations
under the License.
-->
-This page provides instructions on how to run Flink in a *fully distributed
fashion* on a *static* (but possibly heterogeneous) cluster.
+本页面提供了关于如何在*静态*（但可能异构）集群上以*完全分布式方式*运行 Flink 的说明。
* This will be replaced by the TOC
{:toc}
-## Requirements
+
-### Software Requirements
+## 需求
-Flink runs on all *UNIX-like environments*, e.g. **Linux**, **Mac OS X**, and
**Cygwin** (for Windows) and expects the cluster to consist of **one master
node** and **one or more worker nodes**. Before you start to setup the system,
make sure you have the following software installed **on each node**:
+
-- **Java 1.8.x** or higher,
-- **ssh** (sshd must be running to use the Flink scripts that manage
- remote components)
+### 软件需求
-If your cluster does not fulfill these software requirements you will need to
install/upgrade it.
+Flink 运行在所有*类 UNIX 环境*下，例如 **Linux**，**Mac OS X** 和 **Cygwin**
（Windows），并且认为集群由**一个 master 节点**以及**一个或多个 worker
节点**构成。在配置系统之前，请确保**在每个节点上**安装有以下软件：
-Having __passwordless SSH__ and
-__the same directory structure__ on all your cluster nodes will allow you to
use our scripts to control
-everything.
+- **Java 1.8.x** 或更高版本，
+- **ssh** （必须运行 sshd 以使用 Flink 脚本管理远程组件）
+
+如果集群不满足软件要求，那么你需要安装/更新这些软件。
+
+使集群中所有节点使用**免密码 SSH** 以及拥有**相同的目录结构**可以让你使用脚本来控制一切。
{% top %}
-### `JAVA_HOME` Configuration
+
+
+### `JAVA_HOME` 配置
-Flink requires the `JAVA_HOME` environment variable to be set on the master
and all worker nodes and point to the directory of your Java installation.
+Flink 需要 master 和所有 worker 节点设置 `JAVA_HOME` 环境变量，并指向你的 Java 安装目录。
-You can set this variable in `conf/flink-conf.yaml` via the `env.java.home`
key.
+你可以在 `conf/flink-conf.yaml` 文件中通过
[GitHub] [flink] flinkbot commented on pull request #13789: [FLINK-19727][table-runtime] Implement ParallelismProvider for sink i…
flinkbot commented on pull request #13789: URL: https://github.com/apache/flink/pull/13789#issuecomment-716299701 ## CI report: * 6329930d7acc5d2d4d84e777561e79bd25ac9367 UNKNOWN Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build - `@flinkbot run azure` re-run the last Azure build This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13788: [FLINK-19252] Create the working dir for Jaas conf proactively.
flinkbot edited a comment on pull request #13788: URL: https://github.com/apache/flink/pull/13788#issuecomment-716292584 ## CI report: * 0f484654476d9ab49f6ce4a82475fe665e88fe5e Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8261) Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build - `@flinkbot run azure` re-run the last Azure build This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13771: [FLINK-19789][hive] Migrate Hive connector to new table source sink interface
flinkbot edited a comment on pull request #13771: URL: https://github.com/apache/flink/pull/13771#issuecomment-715323930 ## CI report: * e5442954ffac19a92ba1f9a4620584e61acb601a Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8219) * 7cb29bdf576ab721181dcf526e0ec1575971ee33 UNKNOWN Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build - `@flinkbot run azure` re-run the last Azure build This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13767: [FLINK-19787][table-runtime] Migrate Filesystem connector to new table source sink interface
flinkbot edited a comment on pull request #13767: URL: https://github.com/apache/flink/pull/13767#issuecomment-715256450 ## CI report: * 11330173d2fcaeead16e067e1f7d31c50081c2e1 UNKNOWN * c532d886b68b7dd265c9056945a3cfc94e8352b7 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8218) * 0b644cc4e0ffb153ecdd67e8d4ba214341dc1a06 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8260) Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build - `@flinkbot run azure` re-run the last Azure build This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13766: [FLINK-19703][runtime] Untrack a result partition if its producer task failed in TaskManager
flinkbot edited a comment on pull request #13766: URL: https://github.com/apache/flink/pull/13766#issuecomment-715255564 ## CI report: * 42f066384d04bcfe67ccbb5766f09ab5dde9e19c Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8175) * b46f97dd6b1fa67b4dd169867c1967877ac8534c UNKNOWN Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build - `@flinkbot run azure` re-run the last Azure build This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13744: [FLINK-19766][table-runtime] Introduce File streaming compaction operators
flinkbot edited a comment on pull request #13744: URL: https://github.com/apache/flink/pull/13744#issuecomment-714369975 ## CI report: * 4d7e601658c0dacc100c572f4fe8878c058e035f Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8156) * 074ab07749ff798563e785d2c2fafeee1302ac74 UNKNOWN Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build - `@flinkbot run azure` re-run the last Azure build This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13759: [FLINK-19412][python] Re-layer Python Operation Make it Possible to Provide only Python implementation
flinkbot edited a comment on pull request #13759: URL: https://github.com/apache/flink/pull/13759#issuecomment-715027223 ## CI report: * 2e62d962b92ff2c18567a80dcae46fdcfe61d10d Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8259) Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8164) Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build - `@flinkbot run azure` re-run the last Azure build This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [flink] shouweikun commented on a change in pull request #13789: [FLINK-19727][table-runtime] Implement ParallelismProvider for sink i…
shouweikun commented on a change in pull request #13789:
URL: https://github.com/apache/flink/pull/13789#discussion_r511716192
##
File path:
flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/nodes/common/CommonPhysicalSink.scala
##
@@ -99,11 +105,33 @@ class CommonPhysicalSink (
val operator = new SinkOperator(env.clean(sinkFunction),
rowtimeFieldIndex, enforcer)
+val inputParallelism = inputTransformation.getParallelism
+val taskParallelism = env.getParallelism
+val parallelism = if
(runtimeProvider.isInstanceOf[ParallelismProvider])
runtimeProvider.asInstanceOf[ParallelismProvider].getParallelism.orElse(inputParallelism).intValue()
+else inputParallelism
+
+if (implicitly[Ordering[Int]].lteq(parallelism, 0)) throw new
RuntimeException(s"the configured sink parallelism: $parallelism should not be
less than zero or equal to zero")
+if (implicitly[Ordering[Int]].gt(parallelism, taskParallelism)) throw
new RuntimeException(s"the configured sink parallelism: $parallelism is larger
than the task max parallelism: $taskParallelism")
+
+val primaryKeys =
TableSchemaUtils.getPrimaryKeyIndices(catalogTable.getSchema)
+val containedRowKinds = changelogMode.getContainedKinds.toSet
+val theFinalInputTransformation = if(inputParallelism == parallelism)
inputTransformation //if the parallelism is not changed, do nothing
+else (containedRowKinds, primaryKeys.toList) match {
+// fixme : if rowKinds only contains delete, is there somethinng to
do with? Currently do nothing.
+case (_, _) if(containedRowKinds == Set(RowKind.DELETE)) =>
inputTransformation
+case (_, _) if(containedRowKinds == Set(RowKind.INSERT)) =>
inputTransformation
+// fixme: for retract mode (insert and delete contains only), is there
somethinng to do with? Currently do nothing.
+case (_, _) if(containedRowKinds ==
Set(RowKind.INSERT,RowKind.DELETE)) => inputTransformation
+case (_, Nil) if(containedRowKinds.contains(RowKind.UPDATE_AFTER)) =>
throw new RuntimeException(s"ChangelogMode contains ${RowKind.UPDATE_AFTER},
but no primaryKeys were found")
+case (_, _) if(containedRowKinds.contains(RowKind.UPDATE_AFTER)) =>
new
DataStream[RowData](env,inputTransformation).keyBy(primaryKeys:_*).getTransformation
+case _ => throw new RuntimeException(s"the changelogMode is:
${containedRowKinds.mkString(",")}, which is not supported")
+ }
+
Review comment:
I enumerated all the changelog mode conditions by using Scala match
pattern.
So that we can esaily modify every condition in its own scope.
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13631: [FLINK-19639][table sql/planner]Support SupportsNestedProjectionPushD…
flinkbot edited a comment on pull request #13631: URL: https://github.com/apache/flink/pull/13631#issuecomment-708335709 ## CI report: * 11b0dfc5a9a5e6aa335f070abfc6781569deb96d UNKNOWN * a7fe7dec5d7c869ccbc644bba45df241dd05e953 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8252) * 2fdb44da037e75537996e45b164f2a7968c78c73 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8255) Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build - `@flinkbot run azure` re-run the last Azure build This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [flink] shouweikun commented on a change in pull request #13789: [FLINK-19727][table-runtime] Implement ParallelismProvider for sink i…
shouweikun commented on a change in pull request #13789:
URL: https://github.com/apache/flink/pull/13789#discussion_r511713970
##
File path:
flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/nodes/common/CommonPhysicalSink.scala
##
@@ -99,11 +105,33 @@ class CommonPhysicalSink (
val operator = new SinkOperator(env.clean(sinkFunction),
rowtimeFieldIndex, enforcer)
+val inputParallelism = inputTransformation.getParallelism
+val taskParallelism = env.getParallelism
+val parallelism = if
(runtimeProvider.isInstanceOf[ParallelismProvider])
runtimeProvider.asInstanceOf[ParallelismProvider].getParallelism.orElse(inputParallelism).intValue()
+else inputParallelism
+
+if (implicitly[Ordering[Int]].lteq(parallelism, 0)) throw new
RuntimeException(s"the configured sink parallelism: $parallelism should not be
less than zero or equal to zero")
+if (implicitly[Ordering[Int]].gt(parallelism, taskParallelism)) throw
new RuntimeException(s"the configured sink parallelism: $parallelism is larger
than the task max parallelism: $taskParallelism")
+
+val primaryKeys =
TableSchemaUtils.getPrimaryKeyIndices(catalogTable.getSchema)
+val containedRowKinds = changelogMode.getContainedKinds.toSet
+val theFinalInputTransformation = if(inputParallelism == parallelism)
inputTransformation //if the parallelism is not changed, do nothing
+else (containedRowKinds, primaryKeys.toList) match {
+// fixme : if rowKinds only contains delete, is there somethinng to
do with? Currently do nothing.
Review comment:
Actually, this part should be under well discussed.
For every changelog mode, from my perspective, should be treated
differently. Now I only figured out what should be done in `UPSERT MODE` or
`INSERT_ONLY` mode.
What shall we do for other changelog mode?
WDYT?
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [flink] shouweikun commented on a change in pull request #13789: [FLINK-19727][table-runtime] Implement ParallelismProvider for sink i…
shouweikun commented on a change in pull request #13789:
URL: https://github.com/apache/flink/pull/13789#discussion_r511713970
##
File path:
flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/nodes/common/CommonPhysicalSink.scala
##
@@ -99,11 +105,33 @@ class CommonPhysicalSink (
val operator = new SinkOperator(env.clean(sinkFunction),
rowtimeFieldIndex, enforcer)
+val inputParallelism = inputTransformation.getParallelism
+val taskParallelism = env.getParallelism
+val parallelism = if
(runtimeProvider.isInstanceOf[ParallelismProvider])
runtimeProvider.asInstanceOf[ParallelismProvider].getParallelism.orElse(inputParallelism).intValue()
+else inputParallelism
+
+if (implicitly[Ordering[Int]].lteq(parallelism, 0)) throw new
RuntimeException(s"the configured sink parallelism: $parallelism should not be
less than zero or equal to zero")
+if (implicitly[Ordering[Int]].gt(parallelism, taskParallelism)) throw
new RuntimeException(s"the configured sink parallelism: $parallelism is larger
than the task max parallelism: $taskParallelism")
+
+val primaryKeys =
TableSchemaUtils.getPrimaryKeyIndices(catalogTable.getSchema)
+val containedRowKinds = changelogMode.getContainedKinds.toSet
+val theFinalInputTransformation = if(inputParallelism == parallelism)
inputTransformation //if the parallelism is not changed, do nothing
+else (containedRowKinds, primaryKeys.toList) match {
+// fixme : if rowKinds only contains delete, is there somethinng to
do with? Currently do nothing.
Review comment:
Actually, this part should be under well discussed.
For every changelog mode, from my perspective, should be treated
differently. Now I only figured out what should be done in `UPSERT MODE` or
`INSERT_ONLY` mode.
What shall we do for other changelog modes?
WDYT?
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [flink] shouweikun commented on pull request #13789: [FLINK-19727][table-runtime] Implement ParallelismProvider for sink i…
shouweikun commented on pull request #13789: URL: https://github.com/apache/flink/pull/13789#issuecomment-716296334 @JingsongLi I push the pr, still some questions should be well discussed. I have left some comments. This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [flink] shouweikun commented on a change in pull request #13789: [FLINK-19727][table-runtime] Implement ParallelismProvider for sink i…
shouweikun commented on a change in pull request #13789:
URL: https://github.com/apache/flink/pull/13789#discussion_r511713970
##
File path:
flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/nodes/common/CommonPhysicalSink.scala
##
@@ -99,11 +105,33 @@ class CommonPhysicalSink (
val operator = new SinkOperator(env.clean(sinkFunction),
rowtimeFieldIndex, enforcer)
+val inputParallelism = inputTransformation.getParallelism
+val taskParallelism = env.getParallelism
+val parallelism = if
(runtimeProvider.isInstanceOf[ParallelismProvider])
runtimeProvider.asInstanceOf[ParallelismProvider].getParallelism.orElse(inputParallelism).intValue()
+else inputParallelism
+
+if (implicitly[Ordering[Int]].lteq(parallelism, 0)) throw new
RuntimeException(s"the configured sink parallelism: $parallelism should not be
less than zero or equal to zero")
+if (implicitly[Ordering[Int]].gt(parallelism, taskParallelism)) throw
new RuntimeException(s"the configured sink parallelism: $parallelism is larger
than the task max parallelism: $taskParallelism")
+
+val primaryKeys =
TableSchemaUtils.getPrimaryKeyIndices(catalogTable.getSchema)
+val containedRowKinds = changelogMode.getContainedKinds.toSet
+val theFinalInputTransformation = if(inputParallelism == parallelism)
inputTransformation //if the parallelism is not changed, do nothing
+else (containedRowKinds, primaryKeys.toList) match {
+// fixme : if rowKinds only contains delete, is there somethinng to
do with? Currently do nothing.
Review comment:
Actually, this part should be under well discussed.
For every changelog mode, from my perspective, should be treated
differently. Now I only figured out what should be done in "UPSERT MODE" or
`INSERT_ONLY` mode.
What shall we do for other changelog?
WDYT?
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [flink] flinkbot commented on pull request #13789: [FLINK-19727][table-runtime] Implement ParallelismProvider for sink i…
flinkbot commented on pull request #13789: URL: https://github.com/apache/flink/pull/13789#issuecomment-716295445 Thanks a lot for your contribution to the Apache Flink project. I'm the @flinkbot. I help the community to review your pull request. We will use this comment to track the progress of the review. ## Automated Checks Last check on commit 6329930d7acc5d2d4d84e777561e79bd25ac9367 (Mon Oct 26 04:25:14 UTC 2020) **Warnings:** * No documentation files were touched! Remember to keep the Flink docs up to date! Mention the bot in a comment to re-run the automated checks. ## Review Progress * ❓ 1. The [description] looks good. * ❓ 2. There is [consensus] that the contribution should go into to Flink. * ❓ 3. Needs [attention] from. * ❓ 4. The change fits into the overall [architecture]. * ❓ 5. Overall code [quality] is good. Please see the [Pull Request Review Guide](https://flink.apache.org/contributing/reviewing-prs.html) for a full explanation of the review process. The Bot is tracking the review progress through labels. Labels are applied according to the order of the review items. For consensus, approval by a Flink committer of PMC member is required Bot commands The @flinkbot bot supports the following commands: - `@flinkbot approve description` to approve one or more aspects (aspects: `description`, `consensus`, `architecture` and `quality`) - `@flinkbot approve all` to approve all aspects - `@flinkbot approve-until architecture` to approve everything until `architecture` - `@flinkbot attention @username1 [@username2 ..]` to require somebody's attention - `@flinkbot disapprove architecture` to remove an approval you gave earlier This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[jira] [Updated] (FLINK-19727) Implement ParallelismProvider for sink in blink planner
[ https://issues.apache.org/jira/browse/FLINK-19727?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] ASF GitHub Bot updated FLINK-19727: --- Labels: pull-request-available (was: ) > Implement ParallelismProvider for sink in blink planner > --- > > Key: FLINK-19727 > URL: https://issues.apache.org/jira/browse/FLINK-19727 > Project: Flink > Issue Type: Sub-task > Components: Table SQL / Planner >Reporter: Jingsong Lee >Assignee: Lsw_aka_laplace >Priority: Major > Labels: pull-request-available > -- This message was sent by Atlassian Jira (v8.3.4#803005)
[GitHub] [flink] shouweikun opened a new pull request #13789: [FLINK-19727][table-runtime] Implement ParallelismProvider for sink i…
shouweikun opened a new pull request #13789: URL: https://github.com/apache/flink/pull/13789 …n blink planner ## What is the purpose of the change - Implement `ParallelismProvider` for sink in blink planner - `SinkFunctionProvider` and `OutputFormatProvider` work with `ParallelismProvider` by implementing `ParallelismProvider` - Prepare for the future work that make all the classes that implements `SinkRuntimeProvider` currently using in all kind of existing connectors implement `ParallelismProvider` in order to configure the sink parallelism ## Brief change log - `CommonPhysicalSink` : get parallelism from `ParallelismProvider` if possible and configure the sink parallelism into sink transformation after validation - `SinkFunctionProvider`: implements `ParallelismProvider`, add factory method to work with parameter called **parallelism** to set the specified sink parallelism - `OutputFormatProvider`: implements `ParallelismProvider`, add factory method to work with parameter called **parallelism** to set the specified sink parallelism - Add necessary tests and helper class for testing the new code mentioned above ## Verifying this change This change added tests and can be verified as follows: - Add helper class `InternalDataStreamSinkProviderWithParallelism` for the convenience of testing `DataStreamSinkProvider` working with `ParallelismProvider` - Add new field: `@Nullable` parallelism in `TestValuesTableSink` for the test related to `ParallelismProvider` in sink - Exception is expected if `DataStreamSinkProvider` work with `ParallelismProvider` - Test `SinkFunctionProvider` working with `ParallelismProvider` - Test `OutputFormatProvider` working with `ParallelismProvider` ## Does this pull request potentially affect one of the following parts: - Dependencies (does it add or upgrade a dependency): (yes / *no*) - The public API, i.e., is any changed class annotated with `@Public(Evolving)`: (*yes* / no) - The serializers: (yes / *no* / don't know) - The runtime per-record code paths (performance sensitive): (yes /*no* / don't know) - Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn/Mesos, ZooKeeper: (yes / *no* / don't know) - The S3 file system connector: (yes / *no* / don't know) ## Documentation - Does this pull request introduce a new feature? (no) - If yes, how is the feature documented? (not applicable) This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [flink] flinkbot commented on pull request #13788: [FLINK-19252] Create the working dir for Jaas conf proactively.
flinkbot commented on pull request #13788: URL: https://github.com/apache/flink/pull/13788#issuecomment-716292584 ## CI report: * 0f484654476d9ab49f6ce4a82475fe665e88fe5e UNKNOWN Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build - `@flinkbot run azure` re-run the last Azure build This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13787: [FLINK-19522][jdbc] Fix NPE when accessing array using postgres
flinkbot edited a comment on pull request #13787: URL: https://github.com/apache/flink/pull/13787#issuecomment-716285196 ## CI report: * 93ee445aac08ecae0dd55add35411b67a0ec0b01 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8254) Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build - `@flinkbot run azure` re-run the last Azure build This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13759: [FLINK-19412][python] Re-layer Python Operation Make it Possible to Provide only Python implementation
flinkbot edited a comment on pull request #13759: URL: https://github.com/apache/flink/pull/13759#issuecomment-715027223 ## CI report: * 2e62d962b92ff2c18567a80dcae46fdcfe61d10d Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8164) Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8259) Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build - `@flinkbot run azure` re-run the last Azure build This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [flink] HuangXingBo commented on pull request #13759: [FLINK-19412][python] Re-layer Python Operation Make it Possible to Provide only Python implementation
HuangXingBo commented on pull request #13759: URL: https://github.com/apache/flink/pull/13759#issuecomment-716291287 @flinkbot run azure This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[jira] [Closed] (FLINK-19068) Filter verbose pod events for KubernetesResourceManagerDriver
[
https://issues.apache.org/jira/browse/FLINK-19068?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Xintong Song closed FLINK-19068.
Resolution: Fixed
Fixed via
* master: 4b6d0e62cba16e4f43866e68f72bcc3a54337d85
> Filter verbose pod events for KubernetesResourceManagerDriver
> -
>
> Key: FLINK-19068
> URL: https://issues.apache.org/jira/browse/FLINK-19068
> Project: Flink
> Issue Type: Improvement
> Components: Deployment / Kubernetes
>Reporter: Xintong Song
>Assignee: Xintong Song
>Priority: Major
> Labels: pull-request-available
>
> A status of a Kubernetes pod consists of many detailed fields. Currently,
> Flink receives pod {{MODIFIED}} events from the {{KubernetesPodsWatcher}} on
> every single change to these fields, many of which Flink does not care.
> The verbose events will not affect the functionality of Flink, but will
> pollute the logs with repeated messages, because Flink only looks into the
> fields it interested in and those fields are identical.
> E.g., when a task manager is stopped due to idle timeout, Flink receives 3
> events:
> * MODIFIED: container terminated
> * MODIFIED: {{deletionGracePeriodSeconds}} changes from 30 to 0, which is a
> Kubernetes internal status change after containers are gracefully terminated
> * DELETED: Flink removes metadata of the terminated pod
> Among the 3 messages, Flink is only interested in the 1st MODIFIED message,
> but will try to process all of them because the container status is
> terminated.
> I propose to Filter the verbose events in
> {{KubernetesResourceManagerDriver.PodCallbackHandlerImpl}}, to only process
> the status changes interested by Flink. This probably requires recording the
> status of all living pods, to compare with the incoming events for detecting
> status changes.
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[GitHub] [flink] xintongsong closed pull request #13765: [FLINK-19068][k8s] Improve log readability against duplicated pod termination events.
xintongsong closed pull request #13765: URL: https://github.com/apache/flink/pull/13765 This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[jira] [Closed] (FLINK-19750) Deserializer is not opened in Kafka consumer when restoring from state
[
https://issues.apache.org/jira/browse/FLINK-19750?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Jiangjie Qin closed FLINK-19750.
Release Note:
Patch merged.
master: 200be714ff7476f66b7f56cd9b266670b752afe0
release-1.11: 78cd4628285c9dfdc0ada3f59cdefc4e8a7eb0bc
Resolution: Fixed
> Deserializer is not opened in Kafka consumer when restoring from state
> --
>
> Key: FLINK-19750
> URL: https://issues.apache.org/jira/browse/FLINK-19750
> Project: Flink
> Issue Type: Bug
> Components: Connectors / Kafka
>Affects Versions: 1.11.0, 1.11.1, 1.11.2
>Reporter: Qingsheng Ren
>Assignee: Qingsheng Ren
>Priority: Critical
> Labels: pull-request-available
> Fix For: 1.12.0, 1.11.3
>
>
> When a job using Kafka consumer is recovered from a checkpoint or savepoint,
> the {{open}} method of the record deserializer is not called. This is
> possibly because {{this.deserializer.open}} is put into the else clause by
> mistake, which will only be called if the job has a clean start.
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[GitHub] [flink] wuchong commented on a change in pull request #13721: [FLINK-19694][table] Support Upsert ChangelogMode for ScanTableSource
wuchong commented on a change in pull request #13721: URL: https://github.com/apache/flink/pull/13721#discussion_r511709209 ## File path: flink-table/flink-table-planner-blink/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/join/SemiAntiJoinTest.xml ## @@ -307,7 +307,7 @@ Join(joinType=[LeftSemiJoin], where=[$f0], select=[a, b, c], leftInputSpec=[NoUn : +- LegacyTableSourceScan(table=[[default_catalog, default_database, l, source: [TestTableSource(a, b, c)]]], fields=[a, b, c]) +- Exchange(distribution=[single]) +- Calc(select=[IS NOT NULL(m) AS $f0]) - +- GroupAggregate(select=[MIN(i) AS m]) + +- GroupAggregate(select=[MIN_RETRACT(i) AS m]) Review comment: This is a bug in previous `FlinkRelMdModifiedMonotonicity`, the need retraction inference was not correct. This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [flink] becketqin merged pull request #13785: [FLINK-19750][Kafka] Fix bug of not opening DeserializationSchema when FlinkKafkaConsumerBase recovering from state
becketqin merged pull request #13785: URL: https://github.com/apache/flink/pull/13785 This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [flink] wuchong commented on a change in pull request #13721: [FLINK-19694][table] Support Upsert ChangelogMode for ScanTableSource
wuchong commented on a change in pull request #13721:
URL: https://github.com/apache/flink/pull/13721#discussion_r511707227
##
File path:
flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/rules/physical/stream/StreamExecTableSourceScanRule.scala
##
@@ -56,12 +66,57 @@ class StreamExecTableSourceScanRule
def convert(rel: RelNode): RelNode = {
val scan = rel.asInstanceOf[FlinkLogicalTableSourceScan]
val traitSet: RelTraitSet =
rel.getTraitSet.replace(FlinkConventions.STREAM_PHYSICAL)
-
-new StreamExecTableSourceScan(
+val newScan = new StreamExecTableSourceScan(
rel.getCluster,
traitSet,
- scan.getTable.asInstanceOf[TableSourceTable]
-)
+ scan.getTable.asInstanceOf[TableSourceTable])
+
+val table = scan.getTable.asInstanceOf[TableSourceTable]
+val tableSource = table.tableSource.asInstanceOf[ScanTableSource]
+val changelogMode = tableSource.getChangelogMode
+if (changelogMode.contains(RowKind.UPDATE_AFTER) &&
+!changelogMode.contains(RowKind.UPDATE_BEFORE)) {
+ // generate upsert materialize node for upsert source
+ val primaryKey = table.catalogTable.getSchema.getPrimaryKey
+ if (!primaryKey.isPresent) {
+throw new TableException(s"Table
'${table.tableIdentifier.asSummaryString()}' produces" +
+ " a changelog stream contains UPDATE_AFTER but no UPDATE_BEFORE," +
+ " this requires to define primary key on the table.")
+ }
+ val keyFields = primaryKey.get().getColumns
+ val inputFieldNames = newScan.getRowType.getFieldNames
+ val primaryKeyIndices = getPrimaryKeyIndices(inputFieldNames, keyFields)
+ val requiredDistribution = FlinkRelDistribution.hash(primaryKeyIndices,
requireStrict = true)
+ val requiredTraitSet = rel.getCluster.getPlanner.emptyTraitSet()
+.replace(requiredDistribution)
+.replace(FlinkConventions.STREAM_PHYSICAL)
+ val newInput: RelNode = RelOptRule.convert(newScan, requiredTraitSet)
+
+ new StreamExecUpsertMaterialize(
Review comment:
Why we have to add a Calc here? The transformed tree has the same output
row type with the original Scan node. If the primary key fields are never used
in the following nodes, there should already be a Calc after the Scan node. So
I think we shouldn't add a Calc.
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [flink] wuchong commented on a change in pull request #13721: [FLINK-19694][table] Support Upsert ChangelogMode for ScanTableSource
wuchong commented on a change in pull request #13721:
URL: https://github.com/apache/flink/pull/13721#discussion_r511707353
##
File path:
flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdColumnUniqueness.scala
##
@@ -304,6 +304,14 @@ class FlinkRelMdColumnUniqueness private extends
MetadataHandler[BuiltInMetadata
columns != null && util.Arrays.equals(columns.toArray, rel.getUniqueKeys)
}
+ def areColumnsUnique(
+ rel: StreamExecUpsertMaterialize,
+ mq: RelMetadataQuery,
+ columns: ImmutableBitSet,
+ ignoreNulls: Boolean): JBoolean = {
+columns != null && util.Arrays.equals(columns.toArray, rel.uniqueKeys)
Review comment:
Good point!
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [flink] wuchong commented on a change in pull request #13721: [FLINK-19694][table] Support Upsert ChangelogMode for ScanTableSource
wuchong commented on a change in pull request #13721: URL: https://github.com/apache/flink/pull/13721#discussion_r511706661 ## File path: flink-table/flink-table-planner-blink/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/TableScanTest.xml ## @@ -87,6 +129,39 @@ LogicalProject(a=[$0], b=[$1], c=[+($0, 1)], d=[TO_TIMESTAMP($1)], e=[my_udf($0) + + + + + + + + + + +
[jira] [Assigned] (FLINK-19795) Flink SQL throws exception when changelog source contains duplicate change events
[
https://issues.apache.org/jira/browse/FLINK-19795?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Jark Wu reassigned FLINK-19795:
---
Assignee: Jark Wu
> Flink SQL throws exception when changelog source contains duplicate change
> events
> -
>
> Key: FLINK-19795
> URL: https://issues.apache.org/jira/browse/FLINK-19795
> Project: Flink
> Issue Type: Sub-task
> Components: Table SQL / Runtime
>Affects Versions: 1.11.2
>Reporter: jinxin
>Assignee: Jark Wu
>Priority: Major
> Fix For: 1.12.0
>
>
> We are using Canal to synchornize MySQL data into Kafka, the synchornization
> delivery is not exactly-once, so there might be dupcliate
> INSERT/UPDATE/DELETE messages for the same primary key. We are using
> {{'connecotr' = 'kafka', 'format' = 'canal-json'}} to consume such topic.
> However, when appling TopN query on this created source table, the TopN
> operator will thrown exception: {{Caused by: java.lang.RuntimeException: Can
> not retract a non-existent record. This should never happen.}}
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[jira] [Updated] (FLINK-19795) Flink SQL throws exception when changelog source contains duplicate change events
[
https://issues.apache.org/jira/browse/FLINK-19795?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Jark Wu updated FLINK-19795:
Parent: FLINK-18822
Issue Type: Sub-task (was: Bug)
> Flink SQL throws exception when changelog source contains duplicate change
> events
> -
>
> Key: FLINK-19795
> URL: https://issues.apache.org/jira/browse/FLINK-19795
> Project: Flink
> Issue Type: Sub-task
> Components: Table SQL / Runtime
>Affects Versions: 1.11.2
>Reporter: jinxin
>Priority: Major
> Fix For: 1.12.0
>
>
> We are using Canal to synchornize MySQL data into Kafka, the synchornization
> delivery is not exactly-once, so there might be dupcliate
> INSERT/UPDATE/DELETE messages for the same primary key. We are using
> {{'connecotr' = 'kafka', 'format' = 'canal-json'}} to consume such topic.
> However, when appling TopN query on this created source table, the TopN
> operator will thrown exception: {{Caused by: java.lang.RuntimeException: Can
> not retract a non-existent record. This should never happen.}}
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[GitHub] [flink] flinkbot commented on pull request #13788: [FLINK-19252] Create the working dir for Jaas conf proactively.
flinkbot commented on pull request #13788: URL: https://github.com/apache/flink/pull/13788#issuecomment-716285137 Thanks a lot for your contribution to the Apache Flink project. I'm the @flinkbot. I help the community to review your pull request. We will use this comment to track the progress of the review. ## Automated Checks Last check on commit 0f484654476d9ab49f6ce4a82475fe665e88fe5e (Mon Oct 26 03:41:24 UTC 2020) **Warnings:** * No documentation files were touched! Remember to keep the Flink docs up to date! * **This pull request references an unassigned [Jira ticket](https://issues.apache.org/jira/browse/FLINK-19252).** According to the [code contribution guide](https://flink.apache.org/contributing/contribute-code.html), tickets need to be assigned before starting with the implementation work. Mention the bot in a comment to re-run the automated checks. ## Review Progress * ❓ 1. The [description] looks good. * ❓ 2. There is [consensus] that the contribution should go into to Flink. * ❓ 3. Needs [attention] from. * ❓ 4. The change fits into the overall [architecture]. * ❓ 5. Overall code [quality] is good. Please see the [Pull Request Review Guide](https://flink.apache.org/contributing/reviewing-prs.html) for a full explanation of the review process. The Bot is tracking the review progress through labels. Labels are applied according to the order of the review items. For consensus, approval by a Flink committer of PMC member is required Bot commands The @flinkbot bot supports the following commands: - `@flinkbot approve description` to approve one or more aspects (aspects: `description`, `consensus`, `architecture` and `quality`) - `@flinkbot approve all` to approve all aspects - `@flinkbot approve-until architecture` to approve everything until `architecture` - `@flinkbot attention @username1 [@username2 ..]` to require somebody's attention - `@flinkbot disapprove architecture` to remove an approval you gave earlier This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [flink] flinkbot commented on pull request #13787: [FLINK-19522][jdbc] Fix NPE when accessing array using postgres
flinkbot commented on pull request #13787: URL: https://github.com/apache/flink/pull/13787#issuecomment-716285196 ## CI report: * 93ee445aac08ecae0dd55add35411b67a0ec0b01 UNKNOWN Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build - `@flinkbot run azure` re-run the last Azure build This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[jira] [Updated] (FLINK-19252) Jaas file created under io.tmp.dirs - folder not created if not exists
[ https://issues.apache.org/jira/browse/FLINK-19252?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] ASF GitHub Bot updated FLINK-19252: --- Labels: pull-request-available (was: ) > Jaas file created under io.tmp.dirs - folder not created if not exists > -- > > Key: FLINK-19252 > URL: https://issues.apache.org/jira/browse/FLINK-19252 > Project: Flink > Issue Type: Improvement > Components: Runtime / Coordination >Affects Versions: 1.10.0, 1.10.1, 1.10.2 >Reporter: Koala Lam >Priority: Major > Labels: pull-request-available > > https://issues.apache.org/jira/browse/FLINK-14433 > > The security module installation happens before the tmp directory check: > [https://github.com/apache/flink/blob/master/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerServices.java#L390] -- This message was sent by Atlassian Jira (v8.3.4#803005)
[GitHub] [flink] flinkbot edited a comment on pull request #13595: [FLINK-19582][network] Introduce sort-merge based blocking shuffle to Flink
flinkbot edited a comment on pull request #13595: URL: https://github.com/apache/flink/pull/13595#issuecomment-707168234 ## CI report: * 690ec79d32dbd8e2a9c62556c0548a56c6a52521 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8248) * 9b97362104410740de393cbbfb1b27ca564dadc0 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8253) Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build - `@flinkbot run azure` re-run the last Azure build This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13631: [FLINK-19639][table sql/planner]Support SupportsNestedProjectionPushD…
flinkbot edited a comment on pull request #13631: URL: https://github.com/apache/flink/pull/13631#issuecomment-708335709 ## CI report: * 11b0dfc5a9a5e6aa335f070abfc6781569deb96d UNKNOWN * a7fe7dec5d7c869ccbc644bba45df241dd05e953 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8252) * 2fdb44da037e75537996e45b164f2a7968c78c73 UNKNOWN Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build - `@flinkbot run azure` re-run the last Azure build This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [flink] KarmaGYZ opened a new pull request #13788: [FLINK-19252] Create the working dir for Jaas conf proactively.
KarmaGYZ opened a new pull request #13788: URL: https://github.com/apache/flink/pull/13788 ## What is the purpose of the change Create the working dir for Jaas conf proactively. cc @zentol This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[jira] [Updated] (FLINK-19795) Flink SQL throws exception when changelog source contains duplicate change events
[
https://issues.apache.org/jira/browse/FLINK-19795?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Jark Wu updated FLINK-19795:
Fix Version/s: 1.12.0
> Flink SQL throws exception when changelog source contains duplicate change
> events
> -
>
> Key: FLINK-19795
> URL: https://issues.apache.org/jira/browse/FLINK-19795
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Runtime
>Affects Versions: 1.11.2
>Reporter: jinxin
>Priority: Major
> Fix For: 1.12.0
>
>
> We are using Canal to synchornize MySQL data into Kafka, the synchornization
> delivery is not exactly-once, so there might be dupcliate
> INSERT/UPDATE/DELETE messages for the same primary key. We are using
> {{'connecotr' = 'kafka', 'format' = 'canal-json'}} to consume such topic.
> However, when appling TopN query on this created source table, the TopN
> operator will thrown exception: {{Caused by: java.lang.RuntimeException: Can
> not retract a non-existent record. This should never happen.}}
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[jira] [Commented] (FLINK-19795) Flink SQL throws exception when changelog source contains duplicate change events
[
https://issues.apache.org/jira/browse/FLINK-19795?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17220462#comment-17220462
]
Jark Wu commented on FLINK-19795:
-
Regarding to this issue, I think this is very common, because usually the
message delivery between CDC tool and message queue is at-least-once semantic.
Therefore, the change event may duplicate. We have to drop such duplicate
events before processing by the query operators (aggregate, join, topn,
etc...).
> Flink SQL throws exception when changelog source contains duplicate change
> events
> -
>
> Key: FLINK-19795
> URL: https://issues.apache.org/jira/browse/FLINK-19795
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Runtime
>Affects Versions: 1.11.2
>Reporter: jinxin
>Priority: Major
>
> We are using Canal to synchornize MySQL data into Kafka, the synchornization
> delivery is not exactly-once, so there might be dupcliate
> INSERT/UPDATE/DELETE messages for the same primary key. We are using
> {{'connecotr' = 'kafka', 'format' = 'canal-json'}} to consume such topic.
> However, when appling TopN query on this created source table, the TopN
> operator will thrown exception: {{Caused by: java.lang.RuntimeException: Can
> not retract a non-existent record. This should never happen.}}
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[jira] [Commented] (FLINK-19795) Flink SQL throws exception when changelog source contains duplicate change events
[
https://issues.apache.org/jira/browse/FLINK-19795?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17220461#comment-17220461
]
Jark Wu commented on FLINK-19795:
-
After offline discussion with [~shen_dijie], I helped to update the JIRA title
and description.
> Flink SQL throws exception when changelog source contains duplicate change
> events
> -
>
> Key: FLINK-19795
> URL: https://issues.apache.org/jira/browse/FLINK-19795
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Runtime
>Affects Versions: 1.11.2
>Reporter: jinxin
>Priority: Major
>
> We are using Canal to synchornize MySQL data into Kafka, the synchornization
> delivery is not exactly-once, so there might be dupcliate
> INSERT/UPDATE/DELETE messages for the same primary key. We are using
> {{'connecotr' = 'kafka', 'format' = 'canal-json'}} to consume such topic.
> However, when appling TopN query on this created source table, the TopN
> operator will thrown exception: {{Caused by: java.lang.RuntimeException: Can
> not retract a non-existent record. This should never happen. }}
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[jira] [Updated] (FLINK-19795) Flink SQL throws exception when changelog source contains duplicate change events
[
https://issues.apache.org/jira/browse/FLINK-19795?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Jark Wu updated FLINK-19795:
Description: We are using Canal to synchornize MySQL data into Kafka, the
synchornization delivery is not exactly-once, so there might be dupcliate
INSERT/UPDATE/DELETE messages for the same primary key. We are using
{{'connecotr' = 'kafka', 'format' = 'canal-json'}} to consume such topic.
However, when appling TopN query on this created source table, the TopN
operator will thrown exception: {{Caused by: java.lang.RuntimeException: Can
not retract a non-existent record. This should never happen.}} (was: We are
using Canal to synchornize MySQL data into Kafka, the synchornization delivery
is not exactly-once, so there might be dupcliate INSERT/UPDATE/DELETE messages
for the same primary key. We are using {{'connecotr' = 'kafka', 'format' =
'canal-json'}} to consume such topic. However, when appling TopN query on this
created source table, the TopN operator will thrown exception: {{Caused by:
java.lang.RuntimeException: Can not retract a non-existent record. This should
never happen. }})
> Flink SQL throws exception when changelog source contains duplicate change
> events
> -
>
> Key: FLINK-19795
> URL: https://issues.apache.org/jira/browse/FLINK-19795
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Runtime
>Affects Versions: 1.11.2
>Reporter: jinxin
>Priority: Major
>
> We are using Canal to synchornize MySQL data into Kafka, the synchornization
> delivery is not exactly-once, so there might be dupcliate
> INSERT/UPDATE/DELETE messages for the same primary key. We are using
> {{'connecotr' = 'kafka', 'format' = 'canal-json'}} to consume such topic.
> However, when appling TopN query on this created source table, the TopN
> operator will thrown exception: {{Caused by: java.lang.RuntimeException: Can
> not retract a non-existent record. This should never happen.}}
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[jira] [Updated] (FLINK-19795) Flink SQL throws exception when changelog source contains duplicate change events
[
https://issues.apache.org/jira/browse/FLINK-19795?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Jark Wu updated FLINK-19795:
Summary: Flink SQL throws exception when changelog source contains
duplicate change events (was: When TOPN handles duplicate data, it will throw
an exception)
> Flink SQL throws exception when changelog source contains duplicate change
> events
> -
>
> Key: FLINK-19795
> URL: https://issues.apache.org/jira/browse/FLINK-19795
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Runtime
>Affects Versions: 1.11.2
>Reporter: jinxin
>Priority: Major
>
> We are using Canal to synchornize MySQL data into Kafka, the synchornization
> delivery is not exactly-once, so there might be dupcliate
> INSERT/UPDATE/DELETE messages for the same primary key. We are using
> {{'connecotr' = 'kafka', 'format' = 'canal-json'}} to consume such topic.
> However, when appling TopN query on this created source table, the TopN
> operator will thrown exception: {{Caused by: java.lang.RuntimeException: Can
> not retract a non-existent record. This should never happen. }}
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[jira] [Updated] (FLINK-19795) When TOPN handles duplicate data, it will throw an exception
[
https://issues.apache.org/jira/browse/FLINK-19795?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Jark Wu updated FLINK-19795:
Description: We are using Canal to synchornize MySQL data into Kafka, the
synchornization delivery is not exactly-once, so there might be dupcliate
INSERT/UPDATE/DELETE messages for the same primary key. We are using
{{'connecotr' = 'kafka', 'format' = 'canal-json'}} to consume such topic.
However, when appling TopN query on this created source table, the TopN
operator will thrown exception: {{Caused by: java.lang.RuntimeException: Can
not retract a non-existent record. This should never happen. }} (was: When
TOPN handles duplicate data, it will throw an exception :
"Can not retract a non-existent record. This should never happen.")
> When TOPN handles duplicate data, it will throw an exception
> -
>
> Key: FLINK-19795
> URL: https://issues.apache.org/jira/browse/FLINK-19795
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Runtime
>Affects Versions: 1.11.2
>Reporter: jinxin
>Priority: Major
>
> We are using Canal to synchornize MySQL data into Kafka, the synchornization
> delivery is not exactly-once, so there might be dupcliate
> INSERT/UPDATE/DELETE messages for the same primary key. We are using
> {{'connecotr' = 'kafka', 'format' = 'canal-json'}} to consume such topic.
> However, when appling TopN query on this created source table, the TopN
> operator will thrown exception: {{Caused by: java.lang.RuntimeException: Can
> not retract a non-existent record. This should never happen. }}
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[GitHub] [flink] flinkbot edited a comment on pull request #13767: [FLINK-19787][table-runtime] Migrate Filesystem connector to new table source sink interface
flinkbot edited a comment on pull request #13767: URL: https://github.com/apache/flink/pull/13767#issuecomment-715256450 ## CI report: * 11330173d2fcaeead16e067e1f7d31c50081c2e1 UNKNOWN * c532d886b68b7dd265c9056945a3cfc94e8352b7 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8218) * 0b644cc4e0ffb153ecdd67e8d4ba214341dc1a06 UNKNOWN Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build - `@flinkbot run azure` re-run the last Azure build This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [flink] flinkbot commented on pull request #13787: [FLINK-19522][jdbc] Fix NPE when accessing array using postgres
flinkbot commented on pull request #13787: URL: https://github.com/apache/flink/pull/13787#issuecomment-716280232 Thanks a lot for your contribution to the Apache Flink project. I'm the @flinkbot. I help the community to review your pull request. We will use this comment to track the progress of the review. ## Automated Checks Last check on commit 93ee445aac08ecae0dd55add35411b67a0ec0b01 (Mon Oct 26 03:20:21 UTC 2020) **Warnings:** * No documentation files were touched! Remember to keep the Flink docs up to date! Mention the bot in a comment to re-run the automated checks. ## Review Progress * ❓ 1. The [description] looks good. * ❓ 2. There is [consensus] that the contribution should go into to Flink. * ❓ 3. Needs [attention] from. * ❓ 4. The change fits into the overall [architecture]. * ❓ 5. Overall code [quality] is good. Please see the [Pull Request Review Guide](https://flink.apache.org/contributing/reviewing-prs.html) for a full explanation of the review process. The Bot is tracking the review progress through labels. Labels are applied according to the order of the review items. For consensus, approval by a Flink committer of PMC member is required Bot commands The @flinkbot bot supports the following commands: - `@flinkbot approve description` to approve one or more aspects (aspects: `description`, `consensus`, `architecture` and `quality`) - `@flinkbot approve all` to approve all aspects - `@flinkbot approve-until architecture` to approve everything until `architecture` - `@flinkbot attention @username1 [@username2 ..]` to require somebody's attention - `@flinkbot disapprove architecture` to remove an approval you gave earlier This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13631: [FLINK-19639][table sql/planner]Support SupportsNestedProjectionPushD…
flinkbot edited a comment on pull request #13631: URL: https://github.com/apache/flink/pull/13631#issuecomment-708335709 ## CI report: * 11b0dfc5a9a5e6aa335f070abfc6781569deb96d UNKNOWN * a7fe7dec5d7c869ccbc644bba45df241dd05e953 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8252) Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build - `@flinkbot run azure` re-run the last Azure build This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [flink] dforciea commented on pull request #13787: [FLINK-19522][jdbc] Fix NPE when accessing array using postgres
dforciea commented on pull request #13787: URL: https://github.com/apache/flink/pull/13787#issuecomment-716280150 The second commit gets rid of the PostgresRowConverterTest that I had originally added. Let me know if you think this is useful, and I can bring it back. Otherwise, I'll just rebase and eliminate that commit. This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13595: [FLINK-19582][network] Introduce sort-merge based blocking shuffle to Flink
flinkbot edited a comment on pull request #13595: URL: https://github.com/apache/flink/pull/13595#issuecomment-707168234 ## CI report: * 690ec79d32dbd8e2a9c62556c0548a56c6a52521 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8248) * 9b97362104410740de393cbbfb1b27ca564dadc0 UNKNOWN Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build - `@flinkbot run azure` re-run the last Azure build This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[jira] [Commented] (FLINK-19797) NPE when we use decimal array as input of UDAF
[
https://issues.apache.org/jira/browse/FLINK-19797?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17220457#comment-17220457
]
Jark Wu commented on FLINK-19797:
-
Which version are you using?
> NPE when we use decimal array as input of UDAF
> --
>
> Key: FLINK-19797
> URL: https://issues.apache.org/jira/browse/FLINK-19797
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Runtime
>Reporter: Zou
>Priority: Major
>
> Flink throws NPE when we use decimal array as input of UDAF.
> Here is a simple example:
> {code:java}
> select percentile(num, array[0.10, 0.50, 0.99]) from source
> {code}
> And the exception is:
> {code:java}
> Caused by: java.lang.NullPointerException: null
> at
> org.apache.flink.table.data.util.DataFormatConverters$BigDecimalConverter.toExternalImpl(DataFormatConverters.java:680)
> ~[classes/:na]
> at
> org.apache.flink.table.data.util.DataFormatConverters$BigDecimalConverter.toExternalImpl(DataFormatConverters.java:661)
> ~[classes/:na]
> at
> org.apache.flink.table.data.util.DataFormatConverters.arrayDataToJavaArray(DataFormatConverters.java:1195)
> ~[classes/:na]
> at
> org.apache.flink.table.data.util.DataFormatConverters.access$200(DataFormatConverters.java:104)
> ~[classes/:na]
> at
> org.apache.flink.table.data.util.DataFormatConverters$ObjectArrayConverter.toExternalImpl(DataFormatConverters.java:1148)
> ~[classes/:na]
> at
> org.apache.flink.table.data.util.DataFormatConverters$ObjectArrayConverter.toExternalImpl(DataFormatConverters.java:1095)
> ~[classes/:na]
> at
> org.apache.flink.table.data.util.DataFormatConverters$DataFormatConverter.toExternal(DataFormatConverters.java:383)
> ~[classes/:na]
> at NoGroupingAggregateWithoutKeys$25.processElement(Unknown Source)
> ~[na:na]
> at
> org.apache.flink.streaming.runtime.tasks.OneInputStreamTask$StreamTaskNetworkOutput.emitRecord(OneInputStreamTask.java:161)
> ~[classes/:na]
> at
> org.apache.flink.streaming.runtime.io.StreamTaskNetworkInput.processElement(StreamTaskNetworkInput.java:183)
> ~[classes/:na]
> at
> org.apache.flink.streaming.runtime.io.StreamTaskNetworkInput.emitNext(StreamTaskNetworkInput.java:158)
> ~[classes/:na]
> at
> org.apache.flink.streaming.runtime.io.StreamOneInputProcessor.processInput(StreamOneInputProcessor.java:67)
> ~[classes/:na]
> at
> org.apache.flink.streaming.runtime.tasks.StreamTask.processInput(StreamTask.java:351)
> ~[classes/:na]
> at
> org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxStep(MailboxProcessor.java:191)
> ~[classes/:na]
> at
> org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:181)
> ~[classes/:na]
> at
> org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:565)
> ~[classes/:na]
> at
> org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:536)
> ~[classes/:na]
> at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:721)
> ~[classes/:na]
> at org.apache.flink.runtime.taskmanager.Task.run(Task.java:546)
> ~[classes/:na]
> at java.lang.Thread.run(Thread.java:748) ~[na:1.8.0_251]
> {code}
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[GitHub] [flink] dforciea opened a new pull request #13787: Flink 19771
dforciea opened a new pull request #13787: URL: https://github.com/apache/flink/pull/13787 ## What is the purpose of the change Fixes an issue where null arrays in postgres cause a NullPointerException when retrieving data ## Brief change log - Update PostgresRowConverter to override createInternalConverter rather than createNullableInternalConverter - Update tests to include cases for null postgres arrays, for both paths of the PostgresRowConverter (both bytea[] and standard arrays) ## Verifying this change This change added tests and can be verified as follows: - Amended getArrayTable in PostgresCatalogTestBase to include fields for null bytea[] and standard arrays - Update PostgresCatalogITCase to expect nulls for these columns - Validated that this test failed before the change, and succeeds after the change ## Does this pull request potentially affect one of the following parts: - Dependencies (does it add or upgrade a dependency): no - The public API, i.e., is any changed class annotated with `@Public(Evolving)`: no - The serializers: no - The runtime per-record code paths (performance sensitive): no - Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn/Mesos, ZooKeeper: no - The S3 file system connector: no ## Documentation - Does this pull request introduce a new feature? no - If yes, how is the feature documented? not applicable This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [flink] TsReaper commented on a change in pull request #13742: [FLINK-19626][table-planner-blink] Introduce multi-input operator construction algorithm
TsReaper commented on a change in pull request #13742:
URL: https://github.com/apache/flink/pull/13742#discussion_r511700500
##
File path:
flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/plan/reuse/MultipleInputNodeCreationProcessor.java
##
@@ -0,0 +1,481 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.planner.plan.reuse;
+
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.streaming.api.transformations.ShuffleMode;
+import org.apache.flink.streaming.api.transformations.SourceTransformation;
+import
org.apache.flink.table.planner.plan.nodes.exec.AbstractExecNodeExactlyOnceVisitor;
+import org.apache.flink.table.planner.plan.nodes.exec.ExecEdge;
+import org.apache.flink.table.planner.plan.nodes.exec.ExecNode;
+import
org.apache.flink.table.planner.plan.nodes.physical.batch.BatchExecBoundedStreamScan;
+import
org.apache.flink.table.planner.plan.nodes.physical.batch.BatchExecExchange;
+import
org.apache.flink.table.planner.plan.nodes.physical.batch.BatchExecMultipleInputNode;
+import org.apache.flink.table.planner.plan.nodes.physical.batch.BatchExecUnion;
+import
org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecDataStreamScan;
+import
org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecExchange;
+import
org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecMultipleInputNode;
+import
org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecUnion;
+import org.apache.flink.table.planner.plan.nodes.process.DAGProcessContext;
+import org.apache.flink.table.planner.plan.nodes.process.DAGProcessor;
+import org.apache.flink.util.Preconditions;
+
+import org.apache.calcite.rel.RelDistribution;
+import org.apache.calcite.rel.RelNode;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Queue;
+import java.util.Set;
+
+/**
+ * A {@link DAGProcessor} which organize {@link ExecNode}s into multiple input
nodes.
+ *
+ * For a detailed explanation of the algorithm, see appendix of the
+ * https://docs.google.com/document/d/1qKVohV12qn-bM51cBZ8Hcgp31ntwClxjoiNBUOqVHsI;>design
doc.
+ */
+public class MultipleInputNodeCreationProcessor implements DAGProcessor {
+
+ private final boolean isStreaming;
+
+ public MultipleInputNodeCreationProcessor(boolean isStreaming) {
+ this.isStreaming = isStreaming;
+ }
+
+ @Override
+ public List> process(List> sinkNodes,
DAGProcessContext context) {
+ if (!isStreaming) {
+ // As multiple input nodes use function call to deliver
records between sub-operators,
+ // we cannot rely on network buffers to buffer records
not yet ready to be read,
+ // so only BLOCKING dam behavior is safe here.
+ // If conflict is detected under this stricter
constraint,
+ // we add a PIPELINED exchange to mark that its input
and output node cannot be merged
+ // into the same multiple input node
+ InputPriorityConflictResolver resolver = new
InputPriorityConflictResolver(
+ sinkNodes,
+ Collections.emptySet(),
+ ExecEdge.DamBehavior.BLOCKING,
+ ShuffleMode.PIPELINED);
+ resolver.detectAndResolve();
+ }
+
+ List sinkWrappers = wrapExecNodes(sinkNodes);
+ // sort all nodes in topological order, sinks come first and
sources come last
+ List orderedWrappers =
topologicalSort(sinkWrappers);
+ // group nodes into multiple input groups
+ createMultipleInputGroups(orderedWrappers);
+ // apply optimizations to remove unnecessary nodes out of
multiple input groups
+ optimizeMultipleInputGroups(orderedWrappers);
+
+ // create the real multiple input
[GitHub] [flink] TsReaper commented on a change in pull request #13742: [FLINK-19626][table-planner-blink] Introduce multi-input operator construction algorithm
TsReaper commented on a change in pull request #13742:
URL: https://github.com/apache/flink/pull/13742#discussion_r511700192
##
File path:
flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/plan/reuse/MultipleInputNodeCreationProcessor.java
##
@@ -0,0 +1,481 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.planner.plan.reuse;
+
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.streaming.api.transformations.ShuffleMode;
+import org.apache.flink.streaming.api.transformations.SourceTransformation;
+import
org.apache.flink.table.planner.plan.nodes.exec.AbstractExecNodeExactlyOnceVisitor;
+import org.apache.flink.table.planner.plan.nodes.exec.ExecEdge;
+import org.apache.flink.table.planner.plan.nodes.exec.ExecNode;
+import
org.apache.flink.table.planner.plan.nodes.physical.batch.BatchExecBoundedStreamScan;
+import
org.apache.flink.table.planner.plan.nodes.physical.batch.BatchExecExchange;
+import
org.apache.flink.table.planner.plan.nodes.physical.batch.BatchExecMultipleInputNode;
+import org.apache.flink.table.planner.plan.nodes.physical.batch.BatchExecUnion;
+import
org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecDataStreamScan;
+import
org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecExchange;
+import
org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecMultipleInputNode;
+import
org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecUnion;
+import org.apache.flink.table.planner.plan.nodes.process.DAGProcessContext;
+import org.apache.flink.table.planner.plan.nodes.process.DAGProcessor;
+import org.apache.flink.util.Preconditions;
+
+import org.apache.calcite.rel.RelDistribution;
+import org.apache.calcite.rel.RelNode;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Queue;
+import java.util.Set;
+
+/**
+ * A {@link DAGProcessor} which organize {@link ExecNode}s into multiple input
nodes.
+ *
+ * For a detailed explanation of the algorithm, see appendix of the
+ * https://docs.google.com/document/d/1qKVohV12qn-bM51cBZ8Hcgp31ntwClxjoiNBUOqVHsI;>design
doc.
+ */
+public class MultipleInputNodeCreationProcessor implements DAGProcessor {
+
+ private final boolean isStreaming;
+
+ public MultipleInputNodeCreationProcessor(boolean isStreaming) {
+ this.isStreaming = isStreaming;
+ }
+
+ @Override
+ public List> process(List> sinkNodes,
DAGProcessContext context) {
+ if (!isStreaming) {
+ // As multiple input nodes use function call to deliver
records between sub-operators,
+ // we cannot rely on network buffers to buffer records
not yet ready to be read,
+ // so only BLOCKING dam behavior is safe here.
+ // If conflict is detected under this stricter
constraint,
+ // we add a PIPELINED exchange to mark that its input
and output node cannot be merged
+ // into the same multiple input node
+ InputPriorityConflictResolver resolver = new
InputPriorityConflictResolver(
+ sinkNodes,
+ Collections.emptySet(),
+ ExecEdge.DamBehavior.BLOCKING,
+ ShuffleMode.PIPELINED);
+ resolver.detectAndResolve();
+ }
+
+ List sinkWrappers = wrapExecNodes(sinkNodes);
+ // sort all nodes in topological order, sinks come first and
sources come last
+ List orderedWrappers =
topologicalSort(sinkWrappers);
+ // group nodes into multiple input groups
+ createMultipleInputGroups(orderedWrappers);
+ // apply optimizations to remove unnecessary nodes out of
multiple input groups
+ optimizeMultipleInputGroups(orderedWrappers);
+
+ // create the real multiple input
[jira] [Commented] (FLINK-11220) Can not Select row time field in JOIN query
[
https://issues.apache.org/jira/browse/FLINK-11220?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17220455#comment-17220455
]
Jark Wu commented on FLINK-11220:
-
Hi [~twalthr], I think they are different issue.
IIUC, FLINK-10211 is an issue that the time indicators of input table of
regular join node is not materialized.
However, this issue is that the output of SQL query contains multiple time
indicators (planner don't know to choose which rowtime field). But this problem
should only happen when convert SQL query to DataStream, but this exception is
hard thrown even for pure SQL queries.
> Can not Select row time field in JOIN query
> ---
>
> Key: FLINK-11220
> URL: https://issues.apache.org/jira/browse/FLINK-11220
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / API
>Affects Versions: 1.8.0
>Reporter: sunjincheng
>Priority: Major
>
> SQL:
> {code:java}
> Orders...toTable(tEnv, 'orderId, 'orderTime.rowtime)
> Payment...toTable(tEnv, 'orderId, 'payTime.rowtime)
> SELECT orderTime, o.orderId, payTime
> FROM Orders AS o JOIN Payment AS p
> ON o.orderId = p.orderId AND
> p.payTime BETWEEN orderTime AND orderTime + INTERVAL '1' HOUR
> {code}
> Execption:
> {code:java}
> org.apache.flink.table.api.TableException: Found more than one rowtime field:
> [orderTime, payTime] in the table that should be converted to a DataStream.
> Please select the rowtime field that should be used as event-time timestamp
> for the DataStream by casting all other fields to TIMESTAMP.
> at
> org.apache.flink.table.api.StreamTableEnvironment.translate(StreamTableEnvironment.scala:906)
> {code}
> The reason for the error is that we have 2 time fields `orderTime` and
> `payTime`. I think we do not need throw the exception, and we can remove
> the logic of `plan.process(new OutputRowtimeProcessFunction[A](conversion,
> rowtimeFields.head.getIndex))`, if we want using the timestamp after
> toDataSteram, we should using `assignTimestampsAndWatermarks()`.
> What do you think ? [~twalthr] [~fhueske]
>
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[GitHub] [flink] TsReaper commented on a change in pull request #13742: [FLINK-19626][table-planner-blink] Introduce multi-input operator construction algorithm
TsReaper commented on a change in pull request #13742:
URL: https://github.com/apache/flink/pull/13742#discussion_r511699119
##
File path:
flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/plan/reuse/MultipleInputNodeCreationProcessor.java
##
@@ -0,0 +1,481 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.planner.plan.reuse;
+
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.streaming.api.transformations.ShuffleMode;
+import org.apache.flink.streaming.api.transformations.SourceTransformation;
+import
org.apache.flink.table.planner.plan.nodes.exec.AbstractExecNodeExactlyOnceVisitor;
+import org.apache.flink.table.planner.plan.nodes.exec.ExecEdge;
+import org.apache.flink.table.planner.plan.nodes.exec.ExecNode;
+import
org.apache.flink.table.planner.plan.nodes.physical.batch.BatchExecBoundedStreamScan;
+import
org.apache.flink.table.planner.plan.nodes.physical.batch.BatchExecExchange;
+import
org.apache.flink.table.planner.plan.nodes.physical.batch.BatchExecMultipleInputNode;
+import org.apache.flink.table.planner.plan.nodes.physical.batch.BatchExecUnion;
+import
org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecDataStreamScan;
+import
org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecExchange;
+import
org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecMultipleInputNode;
+import
org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecUnion;
+import org.apache.flink.table.planner.plan.nodes.process.DAGProcessContext;
+import org.apache.flink.table.planner.plan.nodes.process.DAGProcessor;
+import org.apache.flink.util.Preconditions;
+
+import org.apache.calcite.rel.RelDistribution;
+import org.apache.calcite.rel.RelNode;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Queue;
+import java.util.Set;
+
+/**
+ * A {@link DAGProcessor} which organize {@link ExecNode}s into multiple input
nodes.
+ *
+ * For a detailed explanation of the algorithm, see appendix of the
+ * https://docs.google.com/document/d/1qKVohV12qn-bM51cBZ8Hcgp31ntwClxjoiNBUOqVHsI;>design
doc.
+ */
+public class MultipleInputNodeCreationProcessor implements DAGProcessor {
+
+ private final boolean isStreaming;
+
+ public MultipleInputNodeCreationProcessor(boolean isStreaming) {
+ this.isStreaming = isStreaming;
+ }
+
+ @Override
+ public List> process(List> sinkNodes,
DAGProcessContext context) {
+ if (!isStreaming) {
+ // As multiple input nodes use function call to deliver
records between sub-operators,
+ // we cannot rely on network buffers to buffer records
not yet ready to be read,
+ // so only BLOCKING dam behavior is safe here.
+ // If conflict is detected under this stricter
constraint,
+ // we add a PIPELINED exchange to mark that its input
and output node cannot be merged
+ // into the same multiple input node
+ InputPriorityConflictResolver resolver = new
InputPriorityConflictResolver(
+ sinkNodes,
+ Collections.emptySet(),
+ ExecEdge.DamBehavior.BLOCKING,
+ ShuffleMode.PIPELINED);
+ resolver.detectAndResolve();
+ }
+
+ List sinkWrappers = wrapExecNodes(sinkNodes);
+ // sort all nodes in topological order, sinks come first and
sources come last
+ List orderedWrappers =
topologicalSort(sinkWrappers);
+ // group nodes into multiple input groups
+ createMultipleInputGroups(orderedWrappers);
+ // apply optimizations to remove unnecessary nodes out of
multiple input groups
+ optimizeMultipleInputGroups(orderedWrappers);
+
+ // create the real multiple input
[GitHub] [flink] TsReaper commented on a change in pull request #13742: [FLINK-19626][table-planner-blink] Introduce multi-input operator construction algorithm
TsReaper commented on a change in pull request #13742:
URL: https://github.com/apache/flink/pull/13742#discussion_r511699281
##
File path:
flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/plan/reuse/MultipleInputNodeCreationProcessor.java
##
@@ -0,0 +1,481 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.planner.plan.reuse;
+
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.streaming.api.transformations.ShuffleMode;
+import org.apache.flink.streaming.api.transformations.SourceTransformation;
+import
org.apache.flink.table.planner.plan.nodes.exec.AbstractExecNodeExactlyOnceVisitor;
+import org.apache.flink.table.planner.plan.nodes.exec.ExecEdge;
+import org.apache.flink.table.planner.plan.nodes.exec.ExecNode;
+import
org.apache.flink.table.planner.plan.nodes.physical.batch.BatchExecBoundedStreamScan;
+import
org.apache.flink.table.planner.plan.nodes.physical.batch.BatchExecExchange;
+import
org.apache.flink.table.planner.plan.nodes.physical.batch.BatchExecMultipleInputNode;
+import org.apache.flink.table.planner.plan.nodes.physical.batch.BatchExecUnion;
+import
org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecDataStreamScan;
+import
org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecExchange;
+import
org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecMultipleInputNode;
+import
org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecUnion;
+import org.apache.flink.table.planner.plan.nodes.process.DAGProcessContext;
+import org.apache.flink.table.planner.plan.nodes.process.DAGProcessor;
+import org.apache.flink.util.Preconditions;
+
+import org.apache.calcite.rel.RelDistribution;
+import org.apache.calcite.rel.RelNode;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Queue;
+import java.util.Set;
+
+/**
+ * A {@link DAGProcessor} which organize {@link ExecNode}s into multiple input
nodes.
+ *
+ * For a detailed explanation of the algorithm, see appendix of the
+ * https://docs.google.com/document/d/1qKVohV12qn-bM51cBZ8Hcgp31ntwClxjoiNBUOqVHsI;>design
doc.
+ */
+public class MultipleInputNodeCreationProcessor implements DAGProcessor {
+
+ private final boolean isStreaming;
+
+ public MultipleInputNodeCreationProcessor(boolean isStreaming) {
+ this.isStreaming = isStreaming;
+ }
+
+ @Override
+ public List> process(List> sinkNodes,
DAGProcessContext context) {
+ if (!isStreaming) {
+ // As multiple input nodes use function call to deliver
records between sub-operators,
+ // we cannot rely on network buffers to buffer records
not yet ready to be read,
+ // so only BLOCKING dam behavior is safe here.
+ // If conflict is detected under this stricter
constraint,
+ // we add a PIPELINED exchange to mark that its input
and output node cannot be merged
+ // into the same multiple input node
+ InputPriorityConflictResolver resolver = new
InputPriorityConflictResolver(
+ sinkNodes,
+ Collections.emptySet(),
+ ExecEdge.DamBehavior.BLOCKING,
+ ShuffleMode.PIPELINED);
+ resolver.detectAndResolve();
+ }
+
+ List sinkWrappers = wrapExecNodes(sinkNodes);
+ // sort all nodes in topological order, sinks come first and
sources come last
+ List orderedWrappers =
topologicalSort(sinkWrappers);
+ // group nodes into multiple input groups
+ createMultipleInputGroups(orderedWrappers);
+ // apply optimizations to remove unnecessary nodes out of
multiple input groups
Review comment:
They're best-to-have, not a must.
[jira] [Commented] (FLINK-19771) NullPointerException when accessing null array from postgres in JDBC Connector
[
https://issues.apache.org/jira/browse/FLINK-19771?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17220454#comment-17220454
]
Leonard Xu commented on FLINK-19771:
[~dforciea]
Your branch makes sense to me, I think you can submit a PR to fix this, thanks.
> NullPointerException when accessing null array from postgres in JDBC Connector
> --
>
> Key: FLINK-19771
> URL: https://issues.apache.org/jira/browse/FLINK-19771
> Project: Flink
> Issue Type: Bug
> Components: Connectors / JDBC
>Affects Versions: 1.11.2
>Reporter: Dylan Forciea
>Priority: Major
>
> When trying to utilize the JDBC Connector for Postgres, I tried to read in a
> text array. When a row that was null was attempted to be read in, the
> connector threw an exception and execution stopped. It appears looking at the
> source code that if the row is null that it will still attempt to grab the
> contents out:
> [https://github.com/apache/flink/blob/release-1.11.2/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/internal/converter/PostgresRowConverter.java#L97]
> The stack trace is as follows:
> {code:java}
> [error] Caused by: java.io.IOException: Couldn't access resultSet
> [error] at
> org.apache.flink.connector.jdbc.table.JdbcRowDataInputFormat.nextRecord(JdbcRowDataInputFormat.java:266)
> [error] at
> org.apache.flink.connector.jdbc.table.JdbcRowDataInputFormat.nextRecord(JdbcRowDataInputFormat.java:57)
> [error] at
> org.apache.flink.streaming.api.functions.source.InputFormatSourceFunction.run(InputFormatSourceFunction.java:91)
> [error] at
> org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:100)
> [error] at
> org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:63)
> [error] at
> org.apache.flink.streaming.runtime.tasks.SourceStreamTask$LegacySourceFunctionThread.run(SourceStreamTask.java:213)
> [error] Caused by: java.lang.NullPointerException
> [error] at
> org.apache.flink.connector.jdbc.internal.converter.PostgresRowConverter.lambda$createPostgresArrayConverter$c06ce9f4$2(PostgresRowConverter.java:97)
> [error] at
> org.apache.flink.connector.jdbc.internal.converter.AbstractJdbcRowConverter.toInternal(AbstractJdbcRowConverter.java:79)
> [error] at
> org.apache.flink.connector.jdbc.table.JdbcRowDataInputFormat.nextRecord(JdbcRowDataInputFormat.java:259)
> [error] ... 5 more {code}
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[jira] [Assigned] (FLINK-19771) NullPointerException when accessing null array from postgres in JDBC Connector
[
https://issues.apache.org/jira/browse/FLINK-19771?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Jark Wu reassigned FLINK-19771:
---
Assignee: Dylan Forciea
> NullPointerException when accessing null array from postgres in JDBC Connector
> --
>
> Key: FLINK-19771
> URL: https://issues.apache.org/jira/browse/FLINK-19771
> Project: Flink
> Issue Type: Bug
> Components: Connectors / JDBC
>Affects Versions: 1.11.2
>Reporter: Dylan Forciea
>Assignee: Dylan Forciea
>Priority: Major
>
> When trying to utilize the JDBC Connector for Postgres, I tried to read in a
> text array. When a row that was null was attempted to be read in, the
> connector threw an exception and execution stopped. It appears looking at the
> source code that if the row is null that it will still attempt to grab the
> contents out:
> [https://github.com/apache/flink/blob/release-1.11.2/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/internal/converter/PostgresRowConverter.java#L97]
> The stack trace is as follows:
> {code:java}
> [error] Caused by: java.io.IOException: Couldn't access resultSet
> [error] at
> org.apache.flink.connector.jdbc.table.JdbcRowDataInputFormat.nextRecord(JdbcRowDataInputFormat.java:266)
> [error] at
> org.apache.flink.connector.jdbc.table.JdbcRowDataInputFormat.nextRecord(JdbcRowDataInputFormat.java:57)
> [error] at
> org.apache.flink.streaming.api.functions.source.InputFormatSourceFunction.run(InputFormatSourceFunction.java:91)
> [error] at
> org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:100)
> [error] at
> org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:63)
> [error] at
> org.apache.flink.streaming.runtime.tasks.SourceStreamTask$LegacySourceFunctionThread.run(SourceStreamTask.java:213)
> [error] Caused by: java.lang.NullPointerException
> [error] at
> org.apache.flink.connector.jdbc.internal.converter.PostgresRowConverter.lambda$createPostgresArrayConverter$c06ce9f4$2(PostgresRowConverter.java:97)
> [error] at
> org.apache.flink.connector.jdbc.internal.converter.AbstractJdbcRowConverter.toInternal(AbstractJdbcRowConverter.java:79)
> [error] at
> org.apache.flink.connector.jdbc.table.JdbcRowDataInputFormat.nextRecord(JdbcRowDataInputFormat.java:259)
> [error] ... 5 more {code}
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[GitHub] [flink] flinkbot edited a comment on pull request #13786: [FLINK-19764] Add More Metrics to TaskManager in Web UI
flinkbot edited a comment on pull request #13786: URL: https://github.com/apache/flink/pull/13786#issuecomment-716269806 ## CI report: * f6bedbd6f4cc29777a888be558030b56ba803a54 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8251) Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build - `@flinkbot run azure` re-run the last Azure build This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [flink] wsry commented on pull request #13595: [FLINK-19582][network] Introduce sort-merge based blocking shuffle to Flink
wsry commented on pull request #13595: URL: https://github.com/apache/flink/pull/13595#issuecomment-716275114 > Thanks a lot, the code looks generally good to me. > > I think the current configuration works also as a feature flag, which is nice. If I read it correctly, sort-shuffle will start from a parallelism of INT_MAX, which means it is never used unless users explicitly configure it to be used. So we do not need any additional feature flag. > > What confused my were the minimum and maximum buffer requirements. > > * There is a maximum number of buffer (`taskmanager.network.sort-merge-blocking-shuffle.max-buffers-per-partition`). Why do we need that? What would happen if there were more buffers? > * We also still have a minimum of `numSubpartitions + 1` buffers. Is this a requirement for the write-phase, or only for the read phase? As I understand it, a part of the motivation of this sort-nased shuffle is to decouple the number of needed memory buffers from the parallelism. Could we drop this minimum? @StephanEwen 1. As suggested by @zhijiangW Zhijiang in the discussion thread, I replace `taskmanager.network.sort-shuffle.min-parallelism` with a feature flag `taskmanager.network.sort-shuffle.enabled` for ease of use. 2. You are right, the motivation of `taskmanager.network.sort-shuffle.max-buffers` is used to decouple the number of needed memory buffers from the parallelism. So it is the upper bound of required memory, if `numSubpartitions + 1` is greater than this config value, this config value is used. The config option is currently used by write-phase, after we implement partition request restriction or buffer management in FLINK-16641, this option can also be reused by read phase. Then we can totally decouple the number of needed memory buffers from the parallelism for both write-phase and read-phase. This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[jira] [Commented] (FLINK-19761) Add lookup method for registered ShuffleDescriptor in ShuffleMaster
[ https://issues.apache.org/jira/browse/FLINK-19761?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17220452#comment-17220452 ] Xuannan Su commented on FLINK-19761: You are right that not returning all the information it needs to consume the result partition will weaken the functionality when sharing cluster partition across clusters. In that case, I agree to send the shuffle descriptor back to the client. However, I feel like it is still missing something without the look-up method. e.g., If the user indeed using an external shuffle service, and some partition is promoted so that it remains after the job is finished. In that case, the external shuffle service should not rely on the client to keep the information to consume the partition. > Add lookup method for registered ShuffleDescriptor in ShuffleMaster > --- > > Key: FLINK-19761 > URL: https://issues.apache.org/jira/browse/FLINK-19761 > Project: Flink > Issue Type: Improvement > Components: Runtime / Network >Reporter: Xuannan Su >Priority: Major > > Currently, the ShuffleMaster can register a partition and get the shuffle > descriptor. However, it lacks the ability to look up the registered > ShuffleDescriptors belongs to an IntermediateResult by the > IntermediateDataSetID. > Adding the lookup method to the ShuffleMaster can make reusing the cluster > partition more easily. For example, we don't have to return the > ShuffleDescriptor to the client just so that the other job can somehow encode > the ShuffleDescriptor in the JobGraph to consume the cluster partition. > Instead, we only need to return the IntermediateDatSetID and use it to lookup > the ShuffleDescriptor by another job. > By adding the lookup method in ShuffleMaster, if we have an external shuffle > service and the lifecycle of the IntermediateResult is not bounded to the > cluster, we can look up the ShuffleDescriptor and reuse the > IntermediateResult by a job running on another cluster even if the cluster > that produced the IntermediateResult is shutdown. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Commented] (FLINK-19771) NullPointerException when accessing null array from postgres in JDBC Connector
[
https://issues.apache.org/jira/browse/FLINK-19771?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17220450#comment-17220450
]
Dylan Forciea commented on FLINK-19771:
---
[~Leonard Xu] Thank you for the pointer on where there are postgres-specific
tests. I have updated my branch accordingly to accomplish what you've
described. Let me know if you want me to submit it as a PR.
> NullPointerException when accessing null array from postgres in JDBC Connector
> --
>
> Key: FLINK-19771
> URL: https://issues.apache.org/jira/browse/FLINK-19771
> Project: Flink
> Issue Type: Bug
> Components: Connectors / JDBC
>Affects Versions: 1.11.2
>Reporter: Dylan Forciea
>Priority: Major
>
> When trying to utilize the JDBC Connector for Postgres, I tried to read in a
> text array. When a row that was null was attempted to be read in, the
> connector threw an exception and execution stopped. It appears looking at the
> source code that if the row is null that it will still attempt to grab the
> contents out:
> [https://github.com/apache/flink/blob/release-1.11.2/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/internal/converter/PostgresRowConverter.java#L97]
> The stack trace is as follows:
> {code:java}
> [error] Caused by: java.io.IOException: Couldn't access resultSet
> [error] at
> org.apache.flink.connector.jdbc.table.JdbcRowDataInputFormat.nextRecord(JdbcRowDataInputFormat.java:266)
> [error] at
> org.apache.flink.connector.jdbc.table.JdbcRowDataInputFormat.nextRecord(JdbcRowDataInputFormat.java:57)
> [error] at
> org.apache.flink.streaming.api.functions.source.InputFormatSourceFunction.run(InputFormatSourceFunction.java:91)
> [error] at
> org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:100)
> [error] at
> org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:63)
> [error] at
> org.apache.flink.streaming.runtime.tasks.SourceStreamTask$LegacySourceFunctionThread.run(SourceStreamTask.java:213)
> [error] Caused by: java.lang.NullPointerException
> [error] at
> org.apache.flink.connector.jdbc.internal.converter.PostgresRowConverter.lambda$createPostgresArrayConverter$c06ce9f4$2(PostgresRowConverter.java:97)
> [error] at
> org.apache.flink.connector.jdbc.internal.converter.AbstractJdbcRowConverter.toInternal(AbstractJdbcRowConverter.java:79)
> [error] at
> org.apache.flink.connector.jdbc.table.JdbcRowDataInputFormat.nextRecord(JdbcRowDataInputFormat.java:259)
> [error] ... 5 more {code}
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[GitHub] [flink] flinkbot edited a comment on pull request #13631: [FLINK-19639][table sql/planner]Support SupportsNestedProjectionPushD…
flinkbot edited a comment on pull request #13631: URL: https://github.com/apache/flink/pull/13631#issuecomment-708335709 ## CI report: * 11b0dfc5a9a5e6aa335f070abfc6781569deb96d UNKNOWN * 46f99482ad7bea8e467130fe5dd92b61b7d3d444 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7858) * a7fe7dec5d7c869ccbc644bba45df241dd05e953 UNKNOWN Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build - `@flinkbot run azure` re-run the last Azure build This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[jira] [Updated] (FLINK-16012) Reduce the default number of exclusive buffers from 2 to 1 on receiver side
[ https://issues.apache.org/jira/browse/FLINK-16012?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Yingjie Cao updated FLINK-16012: Fix Version/s: (was: 1.12.0) 1.13.0 > Reduce the default number of exclusive buffers from 2 to 1 on receiver side > --- > > Key: FLINK-16012 > URL: https://issues.apache.org/jira/browse/FLINK-16012 > Project: Flink > Issue Type: Sub-task > Components: Runtime / Network >Reporter: Zhijiang >Assignee: Yingjie Cao >Priority: Major > Labels: pull-request-available > Fix For: 1.13.0 > > Time Spent: 10m > Remaining Estimate: 0h > > In order to reduce the inflight buffers for checkpoint in the case of back > pressure, we can reduce the number of exclusive buffers for remote input > channel from default 2 to 1 as the first step. Besides that, the total > required buffers are also reduced as a result. We can further verify the > performance effect via various of benchmarks. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Updated] (FLINK-16641) Announce sender's backlog to solve the deadlock issue without exclusive buffers
[ https://issues.apache.org/jira/browse/FLINK-16641?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Yingjie Cao updated FLINK-16641: Fix Version/s: (was: 1.12.0) 1.13.0 > Announce sender's backlog to solve the deadlock issue without exclusive > buffers > --- > > Key: FLINK-16641 > URL: https://issues.apache.org/jira/browse/FLINK-16641 > Project: Flink > Issue Type: Sub-task > Components: Runtime / Network >Reporter: Zhijiang >Assignee: Yingjie Cao >Priority: Major > Labels: pull-request-available > Fix For: 1.13.0 > > > This is the second ingredient besides FLINK-16404 to solve the deadlock > problem without exclusive buffers. > The scenario is as follows: > * The data in subpartition with positive backlog can be sent without doubt > because the exclusive credits would be feedback finally. > * Without exclusive buffers, the receiver would not request floating buffers > for 0 backlog. But when the new backlog is added into such subpartition, it > has no way to notify the receiver side without positive credits ATM. > * So it would result in waiting for each other between receiver and sender > sides to cause deadlock. The sender waits for credit to notify backlog and > the receiver waits for backlog to request floating credits. > To solve the above problem, the sender needs a separate message to announce > backlog sometimes besides existing `BufferResponse`. Then the receiver can > get this info to request floating buffers to feedback. > The side effect brought is to increase network transport delay and throughput > regression. We can measure how much it effects in existing micro-benchmark. > It might probably bear this effect to get a benefit of fast checkpoint > without exclusive buffers. We can give the proper explanations in respective > configuration options to let users make the final decision in practice. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Updated] (FLINK-18762) Make network buffers per incoming/outgoing channel can be configured separately
[ https://issues.apache.org/jira/browse/FLINK-18762?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Yingjie Cao updated FLINK-18762: Fix Version/s: (was: 1.12.0) 1.13.0 > Make network buffers per incoming/outgoing channel can be configured > separately > --- > > Key: FLINK-18762 > URL: https://issues.apache.org/jira/browse/FLINK-18762 > Project: Flink > Issue Type: Sub-task > Components: Runtime / Network >Reporter: Yingjie Cao >Priority: Minor > Labels: pull-request-available > Fix For: 1.13.0 > > > In FLINK-16012, we want to decrease the default number of exclusive buffers > at receiver side from 2 to 1 to accelerate checkpoint in cases of > backpressure. However, number of buffers per outgoing and incoming channels > are configured by a single configuration key. It is better to make network > buffers per incoming/outgoing channel can be configured separately which is > more flexible. At the same time, we can keep the default behavior compatible. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Updated] (FLINK-17572) Remove checkpoint alignment buffered metric from webui
[ https://issues.apache.org/jira/browse/FLINK-17572?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Yingjie Cao updated FLINK-17572: Fix Version/s: (was: 1.12.0) 1.13.0 > Remove checkpoint alignment buffered metric from webui > -- > > Key: FLINK-17572 > URL: https://issues.apache.org/jira/browse/FLINK-17572 > Project: Flink > Issue Type: Sub-task > Components: Runtime / Web Frontend >Affects Versions: 1.11.0 >Reporter: Yingjie Cao >Assignee: Nicholas Jiang >Priority: Minor > Labels: pull-request-available > Fix For: 1.13.0 > > > After FLINK-16404, we never cache buffers while checkpoint barrier alignment, > so the checkpoint alignment buffered metric will be always 0, we should > remove it directly. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Updated] (FLINK-16428) Fine-grained network buffer management for backpressure
[ https://issues.apache.org/jira/browse/FLINK-16428?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Yingjie Cao updated FLINK-16428: Fix Version/s: (was: 1.12.0) 1.13.0 > Fine-grained network buffer management for backpressure > --- > > Key: FLINK-16428 > URL: https://issues.apache.org/jira/browse/FLINK-16428 > Project: Flink > Issue Type: Improvement > Components: Runtime / Network >Reporter: Zhijiang >Priority: Critical > Fix For: 1.13.0 > > > It is an umbrella ticket for tracing the progress of this improvement. > This is the second ingredient to solve the “checkpoints under backpressure” > problem (together with unaligned checkpoints). It consists of two steps: > * See if we can use less network memory in general for streaming jobs (with > potentially different distribution of floating buffers in the input side) > * Under backpressure, reduce network memory to have less in-flight data > (needs specification of algorithm and experiments) -- This message was sent by Atlassian Jira (v8.3.4#803005)
[GitHub] [flink] flinkbot commented on pull request #13786: [FLINK-19764] Add More Metrics to TaskManager in Web UI
flinkbot commented on pull request #13786: URL: https://github.com/apache/flink/pull/13786#issuecomment-716269806 ## CI report: * f6bedbd6f4cc29777a888be558030b56ba803a54 UNKNOWN Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build - `@flinkbot run azure` re-run the last Azure build This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[jira] [Assigned] (FLINK-19727) Implement ParallelismProvider for sink in blink planner
[ https://issues.apache.org/jira/browse/FLINK-19727?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Jingsong Lee reassigned FLINK-19727: Assignee: Lsw_aka_laplace > Implement ParallelismProvider for sink in blink planner > --- > > Key: FLINK-19727 > URL: https://issues.apache.org/jira/browse/FLINK-19727 > Project: Flink > Issue Type: Sub-task > Components: Table SQL / Planner >Reporter: Jingsong Lee >Assignee: Lsw_aka_laplace >Priority: Major > -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Commented] (FLINK-19727) Implement ParallelismProvider for sink in blink planner
[ https://issues.apache.org/jira/browse/FLINK-19727?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17220447#comment-17220447 ] Jingsong Lee commented on FLINK-19727: -- Assigned to you. [~neighborhood] It should be noted that when processing the sink that receives the upsert data, when the parallelism is different, we need to keyBy according to the key. > Implement ParallelismProvider for sink in blink planner > --- > > Key: FLINK-19727 > URL: https://issues.apache.org/jira/browse/FLINK-19727 > Project: Flink > Issue Type: Sub-task > Components: Table SQL / Planner >Reporter: Jingsong Lee >Priority: Major > -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Commented] (FLINK-19777) Fix NullPointException for WindowOperator.close()
[
https://issues.apache.org/jira/browse/FLINK-19777?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17220443#comment-17220443
]
hailong wang commented on FLINK-19777:
--
You are right. More correctly, the dispose method is called before the open
method when job is failed before the operator initialized. For the close method
is invoked when job finished or is canceled, and when job is failed, the
dispose method is invoked.
> Fix NullPointException for WindowOperator.close()
> -
>
> Key: FLINK-19777
> URL: https://issues.apache.org/jira/browse/FLINK-19777
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Runtime
>Affects Versions: 1.11.2
> Environment: jdk 1.8.0_262
> flink 1.11.1
>Reporter: frank wang
>Assignee: Jark Wu
>Priority: Major
> Labels: pull-request-available
> Fix For: 1.12.0, 1.11.3
>
>
> i use flink sql run a job,the sql and metadata is :
> meta :
> 1>soure: kafka
> create table metric_source_window_table(
> `metricName` String,
> `namespace` String,
> `timestamp` BIGINT,
> `doubleValue` DOUBLE,
> `longValue` BIGINT,
> `metricsValue` String,
> `tags` MAP,
> `meta` Map,
> t as TO_TIMESTAMP(FROM_UNIXTIME(`timestamp`/1000,'-MM-dd HH:mm:ss')),
> WATERMARK FOR t AS t) WITH (
> 'connector' = 'kafka',
> 'topic' = 'ai-platform',
> 'properties.bootstrap.servers' = 'xxx',
> 'properties.group.id' = 'metricgroup',
> 'scan.startup.mode'='earliest-offset',
> 'format' = 'json',
> 'json.fail-on-missing-field' = 'false',
> 'json.ignore-parse-errors' = 'true')
> 2>sink to clickhouse(the clickhouse-connector was developed by ourself)
> create table flink_metric_window_table(
> `timestamp` BIGINT,
> `longValue` BIGINT,
> `metricName` String,
> `metricsValueSum` DOUBLE,
> `metricsValueMin` DOUBLE,
> `metricsValueMax` DOUBLE,
> `tag_record_id` String,
> `tag_host_ip` String,
> `tag_instance` String,
> `tag_job_name` String,
> `tag_ai_app_name` String,
> `tag_namespace` String,
> `tag_ai_type` String,
> `tag_host_name` String,
> `tag_alarm_domain` String) WITH (
> 'connector.type' = 'clickhouse',
> 'connector.property-version' = '1',
> 'connector.url' = 'jdbc:clickhouse://xxx:8123/dataeye',
> 'connector.cluster'='ck_cluster',
> 'connector.write.flush.max-rows'='6000',
> 'connector.write.flush.interval'='1000',
> 'connector.table' = 'flink_metric_table_all')
> my sql is :
> insert into
> hive.temp_vipflink.flink_metric_window_table
> select
> cast(HOP_ROWTIME(t, INTERVAL '60' SECOND, INTERVAL '15' MINUTE) AS BIGINT)
> AS `timestamps`,
> sum(COALESCE( `longValue`, 0)) AS longValue,
> metricName,
> sum(IF(IS_DIGIT(metricsValue), cast(metricsValue AS DOUBLE), 0)) AS
> metricsValueSum,
> min(IF(IS_DIGIT(metricsValue), cast(metricsValue AS DOUBLE), 0)) AS
> metricsValueMin,
> max(IF(IS_DIGIT(metricsValue), cast(metricsValue AS DOUBLE), 0)) AS
> metricsValueMax,
> tags ['record_id'],
> tags ['host_ip'],
> tags ['instance'],
> tags ['job_name'],
> tags ['ai_app_name'],
> tags ['namespace'],
> tags ['ai_type'],
> tags ['host_name'],
> tags ['alarm_domain']
> from
> hive.temp_vipflink.metric_source_window_table
> group by
> metricName,
> tags ['record_id'],
> tags ['host_ip'],
> tags ['instance'],
> tags ['job_name'],
> tags ['ai_app_name'],
> tags ['namespace'],
> tags ['ai_type'],
> tags ['host_name'],
> tags ['alarm_domain'],
> HOP(t, INTERVAL '60' SECOND, INTERVAL '15' MINUTE)
>
> when i run this sql for a long hours, it will appear a exception like this:
> [2020-10-22 20:54:52.089] [ERROR] [GroupWindowAggregate(groupBy=[metricName,
> $f1, $f2, $f3, $f4, $f5, $f6, $f7, $f8, $f9], window=[SlidingGroupWindow('w$,
> t, 90, 6)], properties=[w$start, w$end, w$rowtime, w$proctime],
> select=[metricName, $f1, $f2, $f3, $f4, $f5, $f6, $f7, $f8, $f9, SUM($f11) AS
> longValue, SUM($f12) AS metricsValueSum, MIN($f12) AS metricsValueMin,
> MAX($f12) AS metricsValueMax, start('w$) AS w$start, end('w$) AS w$end,
> rowtime('w$) AS w$rowtime, proctime('w$) AS w$proctime]) ->
> Calc(select=[CAST(CAST(w$rowtime)) AS timestamps, longValue, metricName,
> metricsValueSum, metricsValueMin, metricsValueMax, $f1 AS EXPR$6, $f2 AS
> EXPR$7, $f3 AS EXPR$8, $f4 AS EXPR$9, $f5 AS EXPR$10, $f6 AS EXPR$11, $f7 AS
> EXPR$12, $f8 AS EXPR$13, $f9 AS EXPR$14]) -> SinkConversionToTuple2 -> Sink:
> JdbcUpsertTableSink(timestamp, longValue, metricName, metricsValueSum,
> metricsValueMin, metricsValueMax, tag_record_id, tag_host_ip, tag_instance,
> tag_job_name, tag_ai_app_name, tag_namespace, tag_ai_type, tag_host_name,
> tag_alarm_domain) (23/44)]
> [org.apache.flink.streaming.runtime.tasks.StreamTask] >>> Error during
> disposal of stream operator. java.lang.NullPointerException: null at
>
[jira] [Updated] (FLINK-19796) Error results when use constant decimal array
[
https://issues.apache.org/jira/browse/FLINK-19796?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Benchao Li updated FLINK-19796:
---
Affects Version/s: 1.11.2
> Error results when use constant decimal array
> -
>
> Key: FLINK-19796
> URL: https://issues.apache.org/jira/browse/FLINK-19796
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Runtime
>Affects Versions: 1.11.2
>Reporter: Zou
>Assignee: Zou
>Priority: Major
>
> The result is wrong when use constant decimal array with different precisions.
> Here is an example:
> {code:sql}
> create table print_sink(
> data array
> ) with (
> 'connector' = 'print'
> );
> insert into print_sink
> select array[0.12, 0.5, 0.99];
> {code}
> The result will be:
> {code:java}
> +I([0.12, 0.05, 0.99])
> {code}
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[jira] [Commented] (FLINK-19796) Error results when use constant decimal array
[
https://issues.apache.org/jira/browse/FLINK-19796?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17220440#comment-17220440
]
Benchao Li commented on FLINK-19796:
[~FrankZou] assigned to you~
> Error results when use constant decimal array
> -
>
> Key: FLINK-19796
> URL: https://issues.apache.org/jira/browse/FLINK-19796
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Runtime
>Reporter: Zou
>Assignee: Zou
>Priority: Major
>
> The result is wrong when use constant decimal array with different precisions.
> Here is an example:
> {code:sql}
> create table print_sink(
> data array
> ) with (
> 'connector' = 'print'
> );
> insert into print_sink
> select array[0.12, 0.5, 0.99];
> {code}
> The result will be:
> {code:java}
> +I([0.12, 0.05, 0.99])
> {code}
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[jira] [Assigned] (FLINK-19796) Error results when use constant decimal array
[
https://issues.apache.org/jira/browse/FLINK-19796?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Benchao Li reassigned FLINK-19796:
--
Assignee: Zou
> Error results when use constant decimal array
> -
>
> Key: FLINK-19796
> URL: https://issues.apache.org/jira/browse/FLINK-19796
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Runtime
>Reporter: Zou
>Assignee: Zou
>Priority: Major
>
> The result is wrong when use constant decimal array with different precisions.
> Here is an example:
> {code:sql}
> create table print_sink(
> data array
> ) with (
> 'connector' = 'print'
> );
> insert into print_sink
> select array[0.12, 0.5, 0.99];
> {code}
> The result will be:
> {code:java}
> +I([0.12, 0.05, 0.99])
> {code}
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[jira] [Commented] (FLINK-19771) NullPointerException when accessing null array from postgres in JDBC Connector
[
https://issues.apache.org/jira/browse/FLINK-19771?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17220439#comment-17220439
]
Leonard Xu commented on FLINK-19771:
Thanks [~dforciea] for reporting,
I think we should override `createInternalConverter` and reuse null check in
`createNullableInternalConverter`.
The `DerbyRowConverter` and `MySQLRowConverter` have same logic and so use
derby is enough, For `PostgresRowConverter` test, `PostgresCatalogITCase`
cover it, I think we can add null array test in
`PostgresCatalogITCase.testArrayTypes`.
> NullPointerException when accessing null array from postgres in JDBC Connector
> --
>
> Key: FLINK-19771
> URL: https://issues.apache.org/jira/browse/FLINK-19771
> Project: Flink
> Issue Type: Bug
> Components: Connectors / JDBC
>Affects Versions: 1.11.2
>Reporter: Dylan Forciea
>Priority: Major
>
> When trying to utilize the JDBC Connector for Postgres, I tried to read in a
> text array. When a row that was null was attempted to be read in, the
> connector threw an exception and execution stopped. It appears looking at the
> source code that if the row is null that it will still attempt to grab the
> contents out:
> [https://github.com/apache/flink/blob/release-1.11.2/flink-connectors/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/internal/converter/PostgresRowConverter.java#L97]
> The stack trace is as follows:
> {code:java}
> [error] Caused by: java.io.IOException: Couldn't access resultSet
> [error] at
> org.apache.flink.connector.jdbc.table.JdbcRowDataInputFormat.nextRecord(JdbcRowDataInputFormat.java:266)
> [error] at
> org.apache.flink.connector.jdbc.table.JdbcRowDataInputFormat.nextRecord(JdbcRowDataInputFormat.java:57)
> [error] at
> org.apache.flink.streaming.api.functions.source.InputFormatSourceFunction.run(InputFormatSourceFunction.java:91)
> [error] at
> org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:100)
> [error] at
> org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:63)
> [error] at
> org.apache.flink.streaming.runtime.tasks.SourceStreamTask$LegacySourceFunctionThread.run(SourceStreamTask.java:213)
> [error] Caused by: java.lang.NullPointerException
> [error] at
> org.apache.flink.connector.jdbc.internal.converter.PostgresRowConverter.lambda$createPostgresArrayConverter$c06ce9f4$2(PostgresRowConverter.java:97)
> [error] at
> org.apache.flink.connector.jdbc.internal.converter.AbstractJdbcRowConverter.toInternal(AbstractJdbcRowConverter.java:79)
> [error] at
> org.apache.flink.connector.jdbc.table.JdbcRowDataInputFormat.nextRecord(JdbcRowDataInputFormat.java:259)
> [error] ... 5 more {code}
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[GitHub] [flink] flinkbot commented on pull request #13786: [FLINK-19764] Add More Metrics to TaskManager in Web UI
flinkbot commented on pull request #13786: URL: https://github.com/apache/flink/pull/13786#issuecomment-716265458 Thanks a lot for your contribution to the Apache Flink project. I'm the @flinkbot. I help the community to review your pull request. We will use this comment to track the progress of the review. ## Automated Checks Last check on commit f6bedbd6f4cc29777a888be558030b56ba803a54 (Mon Oct 26 02:18:19 UTC 2020) **Warnings:** * No documentation files were touched! Remember to keep the Flink docs up to date! Mention the bot in a comment to re-run the automated checks. ## Review Progress * ❓ 1. The [description] looks good. * ❓ 2. There is [consensus] that the contribution should go into to Flink. * ❓ 3. Needs [attention] from. * ❓ 4. The change fits into the overall [architecture]. * ❓ 5. Overall code [quality] is good. Please see the [Pull Request Review Guide](https://flink.apache.org/contributing/reviewing-prs.html) for a full explanation of the review process. The Bot is tracking the review progress through labels. Labels are applied according to the order of the review items. For consensus, approval by a Flink committer of PMC member is required Bot commands The @flinkbot bot supports the following commands: - `@flinkbot approve description` to approve one or more aspects (aspects: `description`, `consensus`, `architecture` and `quality`) - `@flinkbot approve all` to approve all aspects - `@flinkbot approve-until architecture` to approve everything until `architecture` - `@flinkbot attention @username1 [@username2 ..]` to require somebody's attention - `@flinkbot disapprove architecture` to remove an approval you gave earlier This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[jira] [Updated] (FLINK-19764) Add More Metrics to TaskManager in Web UI
[ https://issues.apache.org/jira/browse/FLINK-19764?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] ASF GitHub Bot updated FLINK-19764: --- Labels: pull-request-available (was: ) > Add More Metrics to TaskManager in Web UI > - > > Key: FLINK-19764 > URL: https://issues.apache.org/jira/browse/FLINK-19764 > Project: Flink > Issue Type: Sub-task > Components: Runtime / Web Frontend >Reporter: Yadong Xie >Assignee: Yadong Xie >Priority: Major > Labels: pull-request-available > > update the UI since https://issues.apache.org/jira/browse/FLINK-14422 and > https://issues.apache.org/jira/browse/FLINK-14406 has been fixed -- This message was sent by Atlassian Jira (v8.3.4#803005)
[GitHub] [flink] vthinkxie opened a new pull request #13786: [FLINK-19764] Add More Metrics to TaskManager in Web UI
vthinkxie opened a new pull request #13786: URL: https://github.com/apache/flink/pull/13786 ## What is the purpose of the change ref https://issues.apache.org/jira/browse/FLINK-19764 ## Brief change log Add More Metrics to TaskManager in Web UI ## Verifying this change - visit the TM metrics page - verify if the TM metrics match the document https://ci.apache.org/projects/flink/flink-docs-release-1.11/ops/memory/mem_setup_tm.html before:  after:  ## Does this pull request potentially affect one of the following parts: - Dependencies (does it add or upgrade a dependency): (no) - The public API, i.e., is any changed class annotated with `@Public(Evolving)`: (no) - The serializers: (no) - The runtime per-record code paths (performance sensitive): (no) - Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Yarn/Mesos, ZooKeeper: (no) - The S3 file system connector: (no) ## Documentation - Does this pull request introduce a new feature? (yes) - If yes, how is the feature documented? (no needed) This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[jira] [Commented] (FLINK-19796) Error results when use constant decimal array
[
https://issues.apache.org/jira/browse/FLINK-19796?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17220437#comment-17220437
]
Zou commented on FLINK-19796:
-
[~libenchao] Yes, I'd like to.
> Error results when use constant decimal array
> -
>
> Key: FLINK-19796
> URL: https://issues.apache.org/jira/browse/FLINK-19796
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Runtime
>Reporter: Zou
>Priority: Major
>
> The result is wrong when use constant decimal array with different precisions.
> Here is an example:
> {code:sql}
> create table print_sink(
> data array
> ) with (
> 'connector' = 'print'
> );
> insert into print_sink
> select array[0.12, 0.5, 0.99];
> {code}
> The result will be:
> {code:java}
> +I([0.12, 0.05, 0.99])
> {code}
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[jira] [Updated] (FLINK-19796) Error results when use constant decimal array
[
https://issues.apache.org/jira/browse/FLINK-19796?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Zou updated FLINK-19796:
Description:
The result is wrong when use constant decimal array with different precisions.
Here is an example:
{code:sql}
create table print_sink(
data array
) with (
'connector' = 'print'
);
insert into print_sink
select array[0.12, 0.5, 0.99];
{code}
The result will be:
{code:java}
+I([0.12, 0.05, 0.99])
{code}
was:
The result is wrong when use constant decimal array with different precisions.
Here is an example:
{code:sql}
create table print_sink(
data array
) with (
'connector' = 'print'
);
insert into print_sink
select array[0.12, 0.5, 0.99];
{code}
The result will be:
{code:java}
+I([0.12, 0.05])
{code}
> Error results when use constant decimal array
> -
>
> Key: FLINK-19796
> URL: https://issues.apache.org/jira/browse/FLINK-19796
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Runtime
>Reporter: Zou
>Priority: Major
>
> The result is wrong when use constant decimal array with different precisions.
> Here is an example:
> {code:sql}
> create table print_sink(
> data array
> ) with (
> 'connector' = 'print'
> );
> insert into print_sink
> select array[0.12, 0.5, 0.99];
> {code}
> The result will be:
> {code:java}
> +I([0.12, 0.05, 0.99])
> {code}
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[jira] [Assigned] (FLINK-11220) Can not Select row time field in JOIN query
[
https://issues.apache.org/jira/browse/FLINK-11220?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Dian Fu reassigned FLINK-11220:
---
Assignee: (was: Dian Fu)
> Can not Select row time field in JOIN query
> ---
>
> Key: FLINK-11220
> URL: https://issues.apache.org/jira/browse/FLINK-11220
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / API
>Affects Versions: 1.8.0
>Reporter: sunjincheng
>Priority: Major
>
> SQL:
> {code:java}
> Orders...toTable(tEnv, 'orderId, 'orderTime.rowtime)
> Payment...toTable(tEnv, 'orderId, 'payTime.rowtime)
> SELECT orderTime, o.orderId, payTime
> FROM Orders AS o JOIN Payment AS p
> ON o.orderId = p.orderId AND
> p.payTime BETWEEN orderTime AND orderTime + INTERVAL '1' HOUR
> {code}
> Execption:
> {code:java}
> org.apache.flink.table.api.TableException: Found more than one rowtime field:
> [orderTime, payTime] in the table that should be converted to a DataStream.
> Please select the rowtime field that should be used as event-time timestamp
> for the DataStream by casting all other fields to TIMESTAMP.
> at
> org.apache.flink.table.api.StreamTableEnvironment.translate(StreamTableEnvironment.scala:906)
> {code}
> The reason for the error is that we have 2 time fields `orderTime` and
> `payTime`. I think we do not need throw the exception, and we can remove
> the logic of `plan.process(new OutputRowtimeProcessFunction[A](conversion,
> rowtimeFields.head.getIndex))`, if we want using the timestamp after
> toDataSteram, we should using `assignTimestampsAndWatermarks()`.
> What do you think ? [~twalthr] [~fhueske]
>
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[jira] [Commented] (FLINK-19791) PartitionRequestClientFactoryTest.testInterruptsNotCached fails with NullPointerException
[
https://issues.apache.org/jira/browse/FLINK-19791?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17220435#comment-17220435
]
Dian Fu commented on FLINK-19791:
-
https://dev.azure.com/apache-flink/apache-flink/_build/results?buildId=8249=logs=f0ac5c25-1168-55a5-07ff-0e88223afed9=0dbaca5d-7c38-52e6-f4fe-2fb69ccb3ada
> PartitionRequestClientFactoryTest.testInterruptsNotCached fails with
> NullPointerException
> -
>
> Key: FLINK-19791
> URL: https://issues.apache.org/jira/browse/FLINK-19791
> Project: Flink
> Issue Type: Bug
> Components: Runtime / Network
>Affects Versions: 1.12.0
>Reporter: Robert Metzger
>Assignee: Roman Khachatryan
>Priority: Major
> Labels: pull-request-available, test-stability
> Fix For: 1.12.0
>
>
> https://dev.azure.com/rmetzger/Flink/_build/results?buildId=8517=logs=6e58d712-c5cc-52fb-0895-6ff7bd56c46b=f30a8e80-b2cf-535c-9952-7f521a4ae374
> {code}
> 2020-10-23T13:25:12.0774554Z [ERROR]
> testInterruptsNotCached(org.apache.flink.runtime.io.network.netty.PartitionRequestClientFactoryTest)
> Time elapsed: 0.762 s <<< ERROR!
> 2020-10-23T13:25:12.0775695Z java.io.IOException:
> java.util.concurrent.ExecutionException:
> org.apache.flink.runtime.io.network.netty.exception.RemoteTransportException:
> Connecting to remote task manager '934dfa03c743/172.18.0.2:8080' has failed.
> This might indicate that the remote task manager has been lost.
> 2020-10-23T13:25:12.0776455Z at
> org.apache.flink.runtime.io.network.netty.PartitionRequestClientFactory.createPartitionRequestClient(PartitionRequestClientFactory.java:95)
> 2020-10-23T13:25:12.0777038Z at
> org.apache.flink.runtime.io.network.netty.PartitionRequestClientFactoryTest.testInterruptsNotCached(PartitionRequestClientFactoryTest.java:72)
> 2020-10-23T13:25:12.0777465Z at
> sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
> 2020-10-23T13:25:12.0777815Z at
> sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
> 2020-10-23T13:25:12.0778221Z at
> sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
> 2020-10-23T13:25:12.0778581Z at
> java.lang.reflect.Method.invoke(Method.java:498)
> 2020-10-23T13:25:12.0778921Z at
> org.junit.runners.model.FrameworkMethod$1.runReflectiveCall(FrameworkMethod.java:50)
> 2020-10-23T13:25:12.0779331Z at
> org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:12)
> 2020-10-23T13:25:12.0779733Z at
> org.junit.runners.model.FrameworkMethod.invokeExplosively(FrameworkMethod.java:47)
> 2020-10-23T13:25:12.0780117Z at
> org.junit.internal.runners.statements.InvokeMethod.evaluate(InvokeMethod.java:17)
> 2020-10-23T13:25:12.0780484Z at
> org.junit.runners.ParentRunner.runLeaf(ParentRunner.java:325)
> 2020-10-23T13:25:12.0780851Z at
> org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:78)
> 2020-10-23T13:25:12.0781236Z at
> org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:57)
> 2020-10-23T13:25:12.0781600Z at
> org.junit.runners.ParentRunner$3.run(ParentRunner.java:290)
> 2020-10-23T13:25:12.0781937Z at
> org.junit.runners.ParentRunner$1.schedule(ParentRunner.java:71)
> 2020-10-23T13:25:12.0782431Z at
> org.junit.runners.ParentRunner.runChildren(ParentRunner.java:288)
> 2020-10-23T13:25:12.0782877Z at
> org.junit.runners.ParentRunner.access$000(ParentRunner.java:58)
> 2020-10-23T13:25:12.0783223Z at
> org.junit.runners.ParentRunner$2.evaluate(ParentRunner.java:268)
> 2020-10-23T13:25:12.0783541Z at
> org.junit.runners.ParentRunner.run(ParentRunner.java:363)
> 2020-10-23T13:25:12.0783905Z at
> org.apache.maven.surefire.junit4.JUnit4Provider.execute(JUnit4Provider.java:365)
> 2020-10-23T13:25:12.0784315Z at
> org.apache.maven.surefire.junit4.JUnit4Provider.executeWithRerun(JUnit4Provider.java:273)
> 2020-10-23T13:25:12.0784718Z at
> org.apache.maven.surefire.junit4.JUnit4Provider.executeTestSet(JUnit4Provider.java:238)
> 2020-10-23T13:25:12.0785125Z at
> org.apache.maven.surefire.junit4.JUnit4Provider.invoke(JUnit4Provider.java:159)
> 2020-10-23T13:25:12.0785552Z at
> org.apache.maven.surefire.booter.ForkedBooter.invokeProviderInSameClassLoader(ForkedBooter.java:384)
> 2020-10-23T13:25:12.0785980Z at
> org.apache.maven.surefire.booter.ForkedBooter.runSuitesInProcess(ForkedBooter.java:345)
> 2020-10-23T13:25:12.0786379Z at
> org.apache.maven.surefire.booter.ForkedBooter.execute(ForkedBooter.java:126)
> 2020-10-23T13:25:12.0786763Z at
> org.apache.maven.surefire.booter.ForkedBooter.main(ForkedBooter.java:418)
> 2020-10-23T13:25:12.0787922Z Caused by:
> java.util.concurrent.ExecutionException:
>
[jira] [Commented] (FLINK-19793) KafkaTableITCase.testKafkaSourceSinkWithMetadata fails on AZP
[
https://issues.apache.org/jira/browse/FLINK-19793?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17220434#comment-17220434
]
Dian Fu commented on FLINK-19793:
-
https://dev.azure.com/apache-flink/apache-flink/_build/results?buildId=8249=logs=1fc6e7bf-633c-5081-c32a-9dea24b05730=80a658d1-f7f6-5d93-2758-53ac19fd5b19
> KafkaTableITCase.testKafkaSourceSinkWithMetadata fails on AZP
> -
>
> Key: FLINK-19793
> URL: https://issues.apache.org/jira/browse/FLINK-19793
> Project: Flink
> Issue Type: Bug
> Components: Connectors / Kafka, Table SQL / Ecosystem
>Affects Versions: 1.12.0
>Reporter: Till Rohrmann
>Priority: Critical
> Labels: test-stability
> Fix For: 1.12.0
>
>
> The {{KafkaTableITCase.testKafkaSourceSinkWithMetadata}} seems to fail on AZP:
> https://dev.azure.com/apache-flink/apache-flink/_build/results?buildId=8197=logs=c5f0071e-1851-543e-9a45-9ac140befc32=1fb1a56f-e8b5-5a82-00a0-a2db7757b4f5
> {code}
> Expected: k2=[B@7c9ecd9e},0,metadata_topic_avro,true>
> 2,2,CreateTime,2020-03-09T13:12:11.123,1,0,{},0,metadata_topic_avro,false>
> k2=[B@4af44e42},0,metadata_topic_avro,true>
> but: was 1,1,CreateTime,2020-03-08T13:12:11.123,0,0,{k1=[B@4ea4e0f3,
> k2=[B@7c9ecd9e},0,metadata_topic_avro,true>
> 2,2,CreateTime,2020-03-09T13:12:11.123,1,0,{},0,metadata_topic_avro,false>
> k2=[B@4af44e42},0,metadata_topic_avro,true>
> {code}
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[jira] [Commented] (FLINK-19201) PyFlink e2e tests is instable and failed with "Connection broken: OSError"
[
https://issues.apache.org/jira/browse/FLINK-19201?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17220433#comment-17220433
]
Dian Fu commented on FLINK-19201:
-
https://dev.azure.com/apache-flink/apache-flink/_build/results?buildId=8250=logs=33dd8067-7758-552f-a1cf-a8b8ff0e44cd=789348ee-cf3e-5c4b-7c78-355970e5f360
> PyFlink e2e tests is instable and failed with "Connection broken: OSError"
> --
>
> Key: FLINK-19201
> URL: https://issues.apache.org/jira/browse/FLINK-19201
> Project: Flink
> Issue Type: Bug
> Components: API / Python
>Affects Versions: 1.11.0, 1.12.0
>Reporter: Dian Fu
>Priority: Major
> Labels: test-stability
>
> [https://dev.azure.com/apache-flink/apache-flink/_build/results?buildId=6452=logs=ff2e2ea5-07e3-5521-7b04-a4fc3ad765e9=6945d9e3-ebef-5993-0c44-838d8ad079c0]
> {code}
> 2020-09-10T21:37:42.9988117Z install conda ... [SUCCESS]
> 2020-09-10T21:37:43.0018449Z install miniconda... [SUCCESS]
> 2020-09-10T21:37:43.0082244Z installing python environment...
> 2020-09-10T21:37:43.0100408Z installing python3.5...
> 2020-09-10T21:37:58.7214400Z install python3.5... [SUCCESS]
> 2020-09-10T21:37:58.7253792Z installing python3.6...
> 2020-09-10T21:38:06.5855143Z install python3.6... [SUCCESS]
> 2020-09-10T21:38:06.5903358Z installing python3.7...
> 2020-09-10T21:38:11.5444706Z
> 2020-09-10T21:38:11.5484852Z ('Connection broken: OSError("(104,
> \'ECONNRESET\')")', OSError("(104, 'ECONNRESET')"))
> 2020-09-10T21:38:11.5513130Z
> 2020-09-10T21:38:11.8044086Z conda install 3.7 failed.You can
> retry to exec the script.
> {code}
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
