[flink] branch master updated: [FLINK-31424][table-planner] Fix NPE produced by multiple sink with local-global window aggregate
This is an automated email from the ASF dual-hosted git repository.
lincoln pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 824259ecf53 [FLINK-31424][table-planner] Fix NPE produced by multiple
sink with local-global window aggregate
824259ecf53 is described below
commit 824259ecf53fd82061d177b508594e6510e69060
Author: Jane Chan <55568005+ladyfor...@users.noreply.github.com>
AuthorDate: Wed Mar 22 16:30:58 2023 +0800
[FLINK-31424][table-planner] Fix NPE produced by multiple sink with
local-global window aggregate
This closes #2
---
.../plan/metadata/FlinkRelMdWindowProperties.scala | 40 ++-
.../table/planner/plan/stats/FlinkStatistic.scala | 6 +-
.../plan/stream/sql/join/WindowJoinTest.xml| 72 +
.../plan/metadata/FlinkRelMdHandlerTestBase.scala | 318 -
.../metadata/FlinkRelMdWindowPropertiesTest.scala | 184
.../planner/plan/metadata/MetadataTestUtil.scala | 36 +++
.../plan/stream/sql/join/WindowJoinTest.scala | 60
7 files changed, 708 insertions(+), 8 deletions(-)
diff --git
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdWindowProperties.scala
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdWindowProperties.scala
index 30087585974..a688302e914 100644
---
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdWindowProperties.scala
+++
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdWindowProperties.scala
@@ -19,6 +19,7 @@ package org.apache.flink.table.planner.plan.metadata
import org.apache.flink.table.planner.{JArrayList, JHashMap, JList}
import org.apache.flink.table.planner.plan.`trait`.RelWindowProperties
+import org.apache.flink.table.planner.plan.logical.WindowSpec
import org.apache.flink.table.planner.plan.nodes.calcite.{Expand,
WatermarkAssigner}
import
org.apache.flink.table.planner.plan.nodes.logical.{FlinkLogicalAggregate,
FlinkLogicalCorrelate, FlinkLogicalJoin, FlinkLogicalRank}
import
org.apache.flink.table.planner.plan.nodes.physical.common.CommonPhysicalLookupJoin
@@ -27,6 +28,7 @@ import
org.apache.flink.table.planner.plan.schema.FlinkPreparingTableBase
import
org.apache.flink.table.planner.plan.utils.WindowJoinUtil.satisfyWindowJoin
import
org.apache.flink.table.planner.plan.utils.WindowUtil.{convertToWindowingStrategy,
groupingContainsWindowStartEnd, isWindowTableFunctionCall}
import org.apache.flink.table.runtime.groupwindow._
+import org.apache.flink.table.types.logical.LogicalType
import org.apache.calcite.plan.hep.HepRelVertex
import org.apache.calcite.plan.volcano.RelSubset
@@ -254,11 +256,34 @@ class FlinkRelMdWindowProperties private extends
MetadataHandler[FlinkMetadata.W
def getWindowProperties(
rel: StreamPhysicalWindowAggregate,
mq: RelMetadataQuery): RelWindowProperties = {
+getWindowAggregateWindowProperties(
+ rel.grouping.length + rel.aggCalls.size(),
+ rel.namedWindowProperties,
+ rel.windowing.getWindow,
+ rel.windowing.getTimeAttributeType
+)
+ }
+
+ def getWindowProperties(
+ rel: StreamPhysicalGlobalWindowAggregate,
+ mq: RelMetadataQuery): RelWindowProperties = {
+getWindowAggregateWindowProperties(
+ rel.grouping.length + rel.aggCalls.size(),
+ rel.namedWindowProperties,
+ rel.windowing.getWindow,
+ rel.windowing.getTimeAttributeType
+)
+ }
+
+ private def getWindowAggregateWindowProperties(
+ propertyOffset: Int,
+ windowProperties: Seq[NamedWindowProperty],
+ windowSpec: WindowSpec,
+ timeAttributeType: LogicalType): RelWindowProperties = {
val starts = ArrayBuffer[Int]()
val ends = ArrayBuffer[Int]()
val times = ArrayBuffer[Int]()
-val propertyOffset = rel.grouping.length + rel.aggCalls.size()
-rel.namedWindowProperties.map(_.getProperty).zipWithIndex.foreach {
+windowProperties.map(_.getProperty).zipWithIndex.foreach {
case (p, index) =>
p match {
case _: WindowStart =>
@@ -275,11 +300,18 @@ class FlinkRelMdWindowProperties private extends
MetadataHandler[FlinkMetadata.W
ImmutableBitSet.of(starts: _*),
ImmutableBitSet.of(ends: _*),
ImmutableBitSet.of(times: _*),
- rel.windowing.getWindow,
- rel.windowing.getTimeAttributeType
+ windowSpec,
+ timeAttributeType
)
}
+ def getWindowProperties(
+ rel: StreamPhysicalLocalWindowAggregate,
+ mq: RelMetadataQuery): RelWindowProperties = {
+val fmq = FlinkRelMetadataQuery.reuseOrCreate(mq)
+fmq.getRelWindowProperties(rel.getInput)
+ }
+
def getWindowProperties(
rel: StreamPhysicalWindowRank,
mq: RelMetadataQuery): Re
[flink-connector-cassandra] branch main updated: [FLINK-26822] Add Cassandra Source
This is an automated email from the ASF dual-hosted git repository.
chesnay pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-connector-cassandra.git
The following commit(s) were added to refs/heads/main by this push:
new 72e3bef [FLINK-26822] Add Cassandra Source
72e3bef is described below
commit 72e3bef1fb9ee6042955b5e9871a9f70a8837cca
Author: Etienne Chauchot
AuthorDate: Wed Mar 22 09:38:17 2023 +0100
[FLINK-26822] Add Cassandra Source
---
flink-connector-cassandra/pom.xml | 7 +
.../cassandra/source/CassandraSource.java | 203 +
.../enumerator/CassandraEnumeratorState.java | 167 ++
.../CassandraEnumeratorStateSerializer.java| 106 +
.../enumerator/CassandraSplitEnumerator.java | 148 +
.../source/reader/CassandraRecordEmitter.java | 138
.../cassandra/source/reader/CassandraRow.java | 46
.../source/reader/CassandraSourceReader.java | 104 +
.../reader/CassandraSourceReaderFactory.java | 55 +
.../source/reader/CassandraSplitReader.java| 205 +
.../cassandra/source/split/CassandraSplit.java | 75 +++
.../source/split/CassandraSplitSerializer.java | 63 ++
.../cassandra/source/split/SplitsGenerator.java| 242 +
.../source/utils/BigIntegerSerializationUtils.java | 40
.../cassandra/example/BatchPojoExample.java| 1 +
.../cassandra/source/CassandraSourceITCase.java| 234
.../cassandra/source/CassandraTestContext.java | 161 ++
.../cassandra/source/CassandraTestEnvironment.java | 196 +
.../CassandraEnumeratorStateSerializerTest.java| 58 +
.../source/reader/CassandraQueryTest.java | 119 ++
.../source/split/CassandraSplitSerializerTest.java | 43
.../cassandra/utils}/Pojo.java | 26 ++-
tools/maven/suppressions.xml | 2 +-
23 files changed, 2437 insertions(+), 2 deletions(-)
diff --git a/flink-connector-cassandra/pom.xml
b/flink-connector-cassandra/pom.xml
index fa78a65..58d70b4 100644
--- a/flink-connector-cassandra/pom.xml
+++ b/flink-connector-cassandra/pom.xml
@@ -78,6 +78,13 @@ under the License.
provided
+
+ org.apache.flink
+ flink-connector-base
+ ${flink.version}
+ provided
+
+
org.scala-lang
scala-library
diff --git
a/flink-connector-cassandra/src/main/java/org/apache/flink/connector/cassandra/source/CassandraSource.java
b/flink-connector-cassandra/src/main/java/org/apache/flink/connector/cassandra/source/CassandraSource.java
new file mode 100644
index 000..dd45913
--- /dev/null
+++
b/flink-connector-cassandra/src/main/java/org/apache/flink/connector/cassandra/source/CassandraSource.java
@@ -0,0 +1,203 @@
+/*
+ * 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.connector.cassandra.source;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.connector.source.Boundedness;
+import org.apache.flink.api.connector.source.Source;
+import org.apache.flink.api.connector.source.SourceReader;
+import org.apache.flink.api.connector.source.SourceReaderContext;
+import org.apache.flink.api.connector.source.SplitEnumerator;
+import org.apache.flink.api.connector.source.SplitEnumeratorContext;
+import org.apache.flink.api.java.ClosureCleaner;
+import org.apache.flink.api.java.typeutils.ResultTypeQueryable;
+import org.apache.flink.configuration.MemorySize;
+import
org.apache.flink.connector.cassandra.source.enumerator.CassandraEnumeratorState;
+import
org.apache.flink.connector.cassandra.source.enumerator.CassandraEnumeratorStateSerializer;
+imp
[flink-web] branch asf-site updated: Add flink-connector-parent 1.0.0
This is an automated email from the ASF dual-hosted git repository. chesnay pushed a commit to branch asf-site in repository https://gitbox.apache.org/repos/asf/flink-web.git The following commit(s) were added to refs/heads/asf-site by this push: new df678144b Add flink-connector-parent 1.0.0 df678144b is described below commit df678144b162d42e2c2b38546584b8fa32451db5 Author: Chesnay Schepler AuthorDate: Wed Mar 15 16:38:51 2023 +0100 Add flink-connector-parent 1.0.0 --- docs/data/additional_components.yml | 6 ++ 1 file changed, 6 insertions(+) diff --git a/docs/data/additional_components.yml b/docs/data/additional_components.yml index e406e67a4..e5ae97d20 100644 --- a/docs/data/additional_components.yml +++ b/docs/data/additional_components.yml @@ -15,6 +15,12 @@ # specific language governing permissions and limitations # under the License +flink-connector-parent: + name: "Apache Flink-connector-parent 1.0.0 Source release" + source_release_url: "https://www.apache.org/dyn/closer.lua/flink/flink-connector-parent-1.0.0/flink-connector-parent-1.0.0-src.tgz"; + source_release_asc_url: "https://downloads.apache.org/flink/flink-connector-parent-1.0.0/flink-connector-parent-1.0.0-src.tgz.asc"; + source_release_sha512_url: "https://downloads.apache.org/flink/flink-connector-parent-1.0.0/flink-connector-parent-1.0.0-src.tgz.sha512"; + flink-shaded: name: "Apache Flink-shaded 16.1 Source Release" source_release_url: "https://www.apache.org/dyn/closer.lua/flink/flink-shaded-16.1/flink-shaded-16.1-src.tgz";
[flink-web] branch asf-site updated: Rebuild website
This is an automated email from the ASF dual-hosted git repository. chesnay pushed a commit to branch asf-site in repository https://gitbox.apache.org/repos/asf/flink-web.git The following commit(s) were added to refs/heads/asf-site by this push: new 756d107b4 Rebuild website 756d107b4 is described below commit 756d107b4d37273f60cac9e2eeb05d73e5438cfa Author: Chesnay Schepler AuthorDate: Wed Mar 22 10:15:51 2023 +0100 Rebuild website --- content/downloads/index.html| 7 +++ content/zh/downloads/index.html | 7 +++ 2 files changed, 14 insertions(+) diff --git a/content/downloads/index.html b/content/downloads/index.html index ed28e9a5d..12116a96f 100644 --- a/content/downloads/index.html +++ b/content/downloads/index.html @@ -945,6 +945,7 @@ https://github.com/alex-shpak/hugo-book Pre-bundled Hadoop 2.6.5 Pre-bundled Hadoop 2.4.1 Apache Flink-shaded 16.1 Source Release +Apache Flink-connector-parent 1.0.0 Source release Verifying Hashes and Signatures @@ -1264,6 +1265,11 @@ under the License. # https://www.apache.org/dyn/closer.lua/flink/flink-shaded-16.1/flink-shaded-16.1-src.tgz";>Apache Flink-shaded 16.1 Source Release Source Release https://downloads.apache.org/flink/flink-shaded-16.1/flink-shaded-16.1-src.tgz.asc";>(asc, https://downloads.apache.org/flink/flink-shaded-16.1/flink-shaded-16.1-src.tgz.sha512";>sha512) + + Apache Flink-connector-parent 1.0.0 Source release + # + +https://www.apache.org/dyn/closer.lua/flink/flink-connector-parent-1.0.0/flink-connector-parent-1.0.0-src.tgz";>Apache Flink-connector-parent 1.0.0 Source release Source Release https://downloads.apache.org/flink/flink-connector-parent-1.0.0/flink-connector-parent-1.0.0-src.tgz.asc";>(asc, https://downloads.apache.org/flink/flink-connector-parent-1.0.0/flink-connector-parent-1.0.0-src.tgz.sha512";>sha512) Verifying Hashes and Signatures # @@ -1625,6 +1631,7 @@ The statefun-flink-harness dependency includes a local execution en Pre-bundled Hadoop 2.6.5 Pre-bundled Hadoop 2.4.1 Apache Flink-shaded 16.1 Source Release +Apache Flink-connector-parent 1.0.0 Source release Verifying Hashes and Signatures diff --git a/content/zh/downloads/index.html b/content/zh/downloads/index.html index 4fbc133d2..1e4f553b1 100644 --- a/content/zh/downloads/index.html +++ b/content/zh/downloads/index.html @@ -945,6 +945,7 @@ https://github.com/alex-shpak/hugo-book Pre-bundled Hadoop 2.6.5 Pre-bundled Hadoop 2.4.1 Apache Flink-shaded 16.1 Source Release +Apache Flink-connector-parent 1.0.0 Source release 验证哈希和签名 @@ -1264,6 +1265,11 @@ under the License. # https://www.apache.org/dyn/closer.lua/flink/flink-shaded-16.1/flink-shaded-16.1-src.tgz";>Apache Flink-shaded 16.1 Source Release Source Release https://downloads.apache.org/flink/flink-shaded-16.1/flink-shaded-16.1-src.tgz.asc";>(asc, https://downloads.apache.org/flink/flink-shaded-16.1/flink-shaded-16.1-src.tgz.sha512";>sha512) + + Apache Flink-connector-parent 1.0.0 Source release + # + +https://www.apache.org/dyn/closer.lua/flink/flink-connector-parent-1.0.0/flink-connector-parent-1.0.0-src.tgz";>Apache Flink-connector-parent 1.0.0 Source release Source Release https://downloads.apache.org/flink/flink-connector-parent-1.0.0/flink-connector-parent-1.0.0-src.tgz.asc";>(asc, https://downloads.apache.org/flink/flink-connector-parent-1.0.0/flink-connector-parent-1.0.0-src.tgz.sha512";>sha512) 验证哈希和签名 # @@ -1626,6 +1632,7 @@ under the License. Pre-bundled Hadoop 2.6.5 Pre-bundled Hadoop 2.4.1 Apache Flink-shaded 16.1 Source Release +Apache Flink-connector-parent 1.0.0 Source release 验证哈希和签名
[flink-connector-cassandra] branch main updated: [hotfix] Update to Apache flink-connector-parent pom.
This is an automated email from the ASF dual-hosted git repository. chesnay pushed a commit to branch main in repository https://gitbox.apache.org/repos/asf/flink-connector-cassandra.git The following commit(s) were added to refs/heads/main by this push: new a250223 [hotfix] Update to Apache flink-connector-parent pom. a250223 is described below commit a2502236e87b3ec65b7b69c1e34a213927f4036f Author: Chesnay Schepler AuthorDate: Wed Mar 22 10:22:35 2023 +0100 [hotfix] Update to Apache flink-connector-parent pom. --- pom.xml | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/pom.xml b/pom.xml index 5b721b2..85ce634 100644 --- a/pom.xml +++ b/pom.xml @@ -20,14 +20,13 @@ under the License. xmlns="http://maven.apache.org/POM/4.0.0"; xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance";> - io.github.zentol.flink + org.apache.flink flink-connector-parent - 1.0 + 1.0.0 4.0.0 - org.apache.flink flink-connector-cassandra-parent 4.0-SNAPSHOT Flink : Connectors : Cassandra : Parent
[flink] branch master updated: [FLINK-31513][doc] Fix incorrect path in tpcds-tool README (#22212)
This is an automated email from the ASF dual-hosted git repository.
guoweijie pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 18869f6dfd0 [FLINK-31513][doc] Fix incorrect path in tpcds-tool README
(#22212)
18869f6dfd0 is described below
commit 18869f6dfd04abd017cf110932c48a33e91c6bb5
Author: leixin <1403342...@qq.com>
AuthorDate: Wed Mar 22 18:22:26 2023 +0800
[FLINK-31513][doc] Fix incorrect path in tpcds-tool README (#22212)
This closes #22212
---
flink-end-to-end-tests/flink-tpcds-test/tpcds-tool/README.md | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/flink-end-to-end-tests/flink-tpcds-test/tpcds-tool/README.md
b/flink-end-to-end-tests/flink-tpcds-test/tpcds-tool/README.md
index 999ab8524da..b3039f3144d 100644
--- a/flink-end-to-end-tests/flink-tpcds-test/tpcds-tool/README.md
+++ b/flink-end-to-end-tests/flink-tpcds-test/tpcds-tool/README.md
@@ -21,7 +21,7 @@ $ export
FLINK_DIR=/flink/flink-dist/target/flink-${flink-ve
$ export END_TO_END_DIR=/flink/flink-end-to-end-tests
# set end to end tests directory
$ mkdir -p /dir_to_save_genarator
$ mkdir -p /dir_to_save_data
-$ cd /flink/flink-end-to-end-tests/tpcds-tool
+$ cd
/flink/flink-end-to-end-tests/flink-tpcds-test/tpcds-tool
$ sh data_generator.sh /dir_to_save_genarator 1
/dir_to_save_data
/flink/flink-end-to-end-tests/test-scripts
```
The downloaded generator will be saved to
`/dir_to_save_genarator`, the generated data will be saved
to `/dir_to_save_data`.
[flink] branch master updated: [FLINK-31527][tests] Stabilize ChangelogRescalingITCase
This is an automated email from the ASF dual-hosted git repository.
roman pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 9ba3b1a5863 [FLINK-31527][tests] Stabilize ChangelogRescalingITCase
9ba3b1a5863 is described below
commit 9ba3b1a5863c1aeeca0be25b4bb375abfe02b940
Author: Roman Khachatryan
AuthorDate: Tue Mar 21 21:08:27 2023 +
[FLINK-31527][tests] Stabilize ChangelogRescalingITCase
---
.../org/apache/flink/test/state/ChangelogRescalingITCase.java | 10 --
1 file changed, 8 insertions(+), 2 deletions(-)
diff --git
a/flink-tests/src/test/java/org/apache/flink/test/state/ChangelogRescalingITCase.java
b/flink-tests/src/test/java/org/apache/flink/test/state/ChangelogRescalingITCase.java
index 6891eafadae..10ba869f6a8 100644
---
a/flink-tests/src/test/java/org/apache/flink/test/state/ChangelogRescalingITCase.java
+++
b/flink-tests/src/test/java/org/apache/flink/test/state/ChangelogRescalingITCase.java
@@ -329,7 +329,7 @@ public class ChangelogRescalingITCase extends TestLogger {
private String checkpointAndCancel(JobID jobID) throws Exception {
waitForCheckpoint(jobID, cluster.getMiniCluster(), 1);
cluster.getClusterClient().cancel(jobID).get();
-checkStatus(jobID);
+waitForSuccessfulTermination(jobID);
return CommonTestUtils.getLatestCompletedCheckpointPath(jobID,
cluster.getMiniCluster())
.orElseThrow(
() -> {
@@ -337,7 +337,13 @@ public class ChangelogRescalingITCase extends TestLogger {
});
}
-private void checkStatus(JobID jobID) throws InterruptedException,
ExecutionException {
+private void waitForSuccessfulTermination(JobID jobID) throws Exception {
+CommonTestUtils.waitUntilCondition(
+() ->
+cluster.getClusterClient()
+.getJobStatus(jobID)
+.get()
+.isGloballyTerminalState());
if
(cluster.getClusterClient().getJobStatus(jobID).get().isGloballyTerminalState())
{
cluster.getClusterClient()
.requestJobResult(jobID)
[flink] branch release-1.16 updated (e891bfc156d -> 80ee512f00a)
This is an automated email from the ASF dual-hosted git repository. roman pushed a change to branch release-1.16 in repository https://gitbox.apache.org/repos/asf/flink.git from e891bfc156d [python][examples] Add Python streaming word count examples add 80ee512f00a [FLINK-31527][tests] Stabilize ChangelogRescalingITCase No new revisions were added by this update. Summary of changes: .../org/apache/flink/test/state/ChangelogRescalingITCase.java | 10 -- 1 file changed, 8 insertions(+), 2 deletions(-)
[flink-kubernetes-operator] branch main updated: [FLINK-31407] Bump fabric8 version to 6.5.0
This is an automated email from the ASF dual-hosted git repository.
mbalassi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-kubernetes-operator.git
The following commit(s) were added to refs/heads/main by this push:
new 91cf2338 [FLINK-31407] Bump fabric8 version to 6.5.0
91cf2338 is described below
commit 91cf2338c9bd54b1bad7680c231a59facc12c8b9
Author: Márton Balassi
AuthorDate: Wed Mar 22 22:25:26 2023 +0100
[FLINK-31407] Bump fabric8 version to 6.5.0
---
.../operator/autoscaler/AutoScalerInfo.java| 2 +-
.../operator/service/AbstractFlinkService.java | 1 -
.../operator/utils/KubernetesClientUtils.java | 4 +-
.../kubernetes/operator/utils/StatusRecorder.java | 4 +-
.../flink/kubernetes/operator/TestUtils.java | 2 +-
.../metrics/KubernetesClientMetricsTest.java | 2 +-
.../operator/utils/KubernetesClientUtilsTest.java | 20 +--
.../Fabric8FlinkStandaloneKubeClient.java | 5 +-
.../crds/flinkdeployments.flink.apache.org-v1.yml | 153 +
pom.xml| 2 +-
10 files changed, 169 insertions(+), 26 deletions(-)
diff --git
a/flink-kubernetes-operator-autoscaler/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/AutoScalerInfo.java
b/flink-kubernetes-operator-autoscaler/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/AutoScalerInfo.java
index 8850b95a..a5462219 100644
---
a/flink-kubernetes-operator-autoscaler/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/AutoScalerInfo.java
+++
b/flink-kubernetes-operator-autoscaler/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/AutoScalerInfo.java
@@ -181,7 +181,7 @@ public class AutoScalerInfo {
public void replaceInKubernetes(KubernetesClient client) throws Exception {
trimHistoryToMaxCmSize();
-client.resource(configMap).replace();
+client.resource(configMap).update();
}
@VisibleForTesting
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkService.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkService.java
index 4c5d9fcc..97262daf 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkService.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkService.java
@@ -808,7 +808,6 @@ public abstract class AbstractFlinkService implements
FlinkService {
.inNamespace(namespace)
.withName(
ExternalServiceDecorator.getExternalServiceName(clusterId))
-.fromServer()
.get();
if (service == null) {
serviceRunning = false;
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/KubernetesClientUtils.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/KubernetesClientUtils.java
index d52bfc67..d6724adf 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/KubernetesClientUtils.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/KubernetesClientUtils.java
@@ -73,12 +73,12 @@ public class KubernetesClientUtils {
public static > void applyToStoredCr(
KubernetesClient kubernetesClient, T cr, Consumer function) {
-var inKube = kubernetesClient.resource(cr).fromServer().get();
+var inKube = kubernetesClient.resource(cr).get();
Long localGeneration = cr.getMetadata().getGeneration();
Long serverGeneration = inKube.getMetadata().getGeneration();
if (serverGeneration.equals(localGeneration)) {
function.accept(inKube);
-kubernetesClient.resource(inKube).lockResourceVersion().replace();
+kubernetesClient.resource(inKube).lockResourceVersion().update();
} else {
LOG.info(
"Spec already upgrading in kube (generation - local: {}
server: {}), skipping scale operation.",
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/StatusRecorder.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/StatusRecorder.java
index b1bec2f7..e009a7ce 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/StatusRecorder.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/StatusRecorder.java
@@ -120,7 +120,7 @@ public class StatusRecorder<
int retries = 0;
while (true) {
try {
-var updated =
[flink] branch master updated (9ba3b1a5863 -> e91eb5ec2fe)
This is an automated email from the ASF dual-hosted git repository. snuyanzin pushed a change to branch master in repository https://gitbox.apache.org/repos/asf/flink.git from 9ba3b1a5863 [FLINK-31527][tests] Stabilize ChangelogRescalingITCase add e91eb5ec2fe [FLINK-31466][table] Apply Calcite's optimization of pushing filter to semijoin for FilterJoinRule No new revisions were added by this update. Summary of changes: .../plan/rules/logical/FlinkFilterJoinRule.java| 27 +-- .../sql/join/BroadcastHashSemiAntiJoinTest.xml | 38 +++ .../batch/sql/join/NestedLoopSemiAntiJoinTest.xml | 50 +-- .../plan/batch/sql/join/SemiAntiJoinTest.xml | 52 ++-- .../sql/join/ShuffledHashSemiAntiJoinTest.xml | 42 .../batch/sql/join/SortMergeSemiAntiJoinTest.xml | 42 .../logical/CalcPruneAggregateCallRuleTest.xml | 16 +++ .../plan/rules/logical/FlinkFilterJoinRuleTest.xml | 9 ++-- .../logical/ProjectPruneAggregateCallRuleTest.xml | 56 -- .../plan/stream/sql/join/SemiAntiJoinTest.xml | 54 ++--- .../stream/sql/SemiAntiJoinStreamITCase.scala | 27 ++- 11 files changed, 219 insertions(+), 194 deletions(-)
[flink] branch master updated: [FLINK-31230][yarn] Improve YarnClusterDescriptor memory unit display.
This is an automated email from the ASF dual-hosted git repository.
fanrui pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 753da61bb36 [FLINK-31230][yarn] Improve YarnClusterDescriptor memory
unit display.
753da61bb36 is described below
commit 753da61bb366df35df4bc2c2ebbbe7f75e8237d3
Author: slfan1989
AuthorDate: Wed Mar 22 19:33:34 2023 +0800
[FLINK-31230][yarn] Improve YarnClusterDescriptor memory unit display.
---
.../apache/flink/yarn/YARNSessionFIFOITCase.java | 2 +-
.../apache/flink/yarn/YarnClusterDescriptor.java | 28 ++
2 files changed, 19 insertions(+), 11 deletions(-)
diff --git
a/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNSessionFIFOITCase.java
b/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNSessionFIFOITCase.java
index a57b9b75bc3..8c19fd17e35 100644
---
a/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNSessionFIFOITCase.java
+++
b/flink-yarn-tests/src/test/java/org/apache/flink/yarn/YARNSessionFIFOITCase.java
@@ -244,7 +244,7 @@ class YARNSessionFIFOITCase extends YarnTestBase {
log.info("Starting testQueryCluster()");
runWithArgs(
new String[] {"-q"},
-"Summary: totalMemory 8192 totalCores 1332",
+"Summary: totalMemory 8.000gb (8589934592 bytes)
totalCores 1332",
null,
RunTypes.YARN_SESSION,
0); // we have 666*2 cores.
diff --git
a/flink-yarn/src/main/java/org/apache/flink/yarn/YarnClusterDescriptor.java
b/flink-yarn/src/main/java/org/apache/flink/yarn/YarnClusterDescriptor.java
index 95fe09f9c06..8231fd6fa59 100644
--- a/flink-yarn/src/main/java/org/apache/flink/yarn/YarnClusterDescriptor.java
+++ b/flink-yarn/src/main/java/org/apache/flink/yarn/YarnClusterDescriptor.java
@@ -38,6 +38,7 @@ import org.apache.flink.configuration.CoreOptions;
import org.apache.flink.configuration.HighAvailabilityOptions;
import org.apache.flink.configuration.IllegalConfigurationException;
import org.apache.flink.configuration.JobManagerOptions;
+import org.apache.flink.configuration.MemorySize;
import org.apache.flink.configuration.PipelineOptions;
import org.apache.flink.configuration.ResourceManagerOptions;
import org.apache.flink.configuration.RestOptions;
@@ -97,6 +98,7 @@ import org.apache.hadoop.yarn.client.api.YarnClient;
import org.apache.hadoop.yarn.client.api.YarnClientApplication;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.apache.hadoop.yarn.exceptions.YarnException;
+import org.apache.hadoop.yarn.util.ConverterUtils;
import org.apache.hadoop.yarn.util.Records;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -844,7 +846,7 @@ public class YarnClusterDescriptor implements
ClusterDescriptor {
final ApplicationId appId = appContext.getApplicationId();
-// -- Add Zookeeper namespace to local
flinkConfiguration --
+// -- Add Zookeeper namespace to local
flinkConfiguraton --
setHAClusterIdIfNotSet(configuration, appId);
if
(HighAvailabilityMode.isHighAvailabilityModeActivated(configuration)) {
@@ -1229,7 +1231,6 @@ public class YarnClusterDescriptor implements
ClusterDescriptor {
LOG.info("Waiting for the cluster to be allocated");
final long startTime = System.currentTimeMillis();
-long lastLogTime = System.currentTimeMillis();
ApplicationReport report;
YarnApplicationState lastAppState = YarnApplicationState.NEW;
loop:
@@ -1265,11 +1266,9 @@ public class YarnClusterDescriptor implements
ClusterDescriptor {
if (appState != lastAppState) {
LOG.info("Deploying cluster, current state " +
appState);
}
-if (System.currentTimeMillis() - lastLogTime > 6) {
-lastLogTime = System.currentTimeMillis();
+if (System.currentTimeMillis() - startTime > 6) {
LOG.info(
-"Deployment took more than {} seconds. Please
check if the requested resources are available in the YARN cluster",
-(lastLogTime - startTime) / 1000);
+"Deployment took more than 60 seconds. Please
check if the requested resources are available in the YARN cluster");
}
}
lastAppState = appState;
@@ -1443,13 +1442,17 @@ public class YarnClusterDescriptor implements
ClusterDescriptor {
totalMemory += res.getMemory();
totalCores += res.getVirtualCores();
ps.format(format, "NodeID", rep.getNodeId());
-
[flink] branch release-1.17 updated: [FLINK-31424][table-planner] Fix NPE produced by multiple sink with local-global window aggregate
This is an automated email from the ASF dual-hosted git repository.
lincoln pushed a commit to branch release-1.17
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.17 by this push:
new 71f23066562 [FLINK-31424][table-planner] Fix NPE produced by multiple
sink with local-global window aggregate
71f23066562 is described below
commit 71f23066562018910e5599719d2a1bc0eb17f430
Author: Jane Chan <55568005+ladyfor...@users.noreply.github.com>
AuthorDate: Thu Mar 23 12:04:42 2023 +0800
[FLINK-31424][table-planner] Fix NPE produced by multiple sink with
local-global window aggregate
This closes #22238
---
.../plan/metadata/FlinkRelMdWindowProperties.scala | 40 ++-
.../table/planner/plan/stats/FlinkStatistic.scala | 6 +-
.../plan/stream/sql/join/WindowJoinTest.xml| 72 +
.../plan/metadata/FlinkRelMdHandlerTestBase.scala | 318 -
.../metadata/FlinkRelMdWindowPropertiesTest.scala | 184
.../planner/plan/metadata/MetadataTestUtil.scala | 36 +++
.../plan/stream/sql/join/WindowJoinTest.scala | 60
7 files changed, 708 insertions(+), 8 deletions(-)
diff --git
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdWindowProperties.scala
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdWindowProperties.scala
index 30087585974..a688302e914 100644
---
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdWindowProperties.scala
+++
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdWindowProperties.scala
@@ -19,6 +19,7 @@ package org.apache.flink.table.planner.plan.metadata
import org.apache.flink.table.planner.{JArrayList, JHashMap, JList}
import org.apache.flink.table.planner.plan.`trait`.RelWindowProperties
+import org.apache.flink.table.planner.plan.logical.WindowSpec
import org.apache.flink.table.planner.plan.nodes.calcite.{Expand,
WatermarkAssigner}
import
org.apache.flink.table.planner.plan.nodes.logical.{FlinkLogicalAggregate,
FlinkLogicalCorrelate, FlinkLogicalJoin, FlinkLogicalRank}
import
org.apache.flink.table.planner.plan.nodes.physical.common.CommonPhysicalLookupJoin
@@ -27,6 +28,7 @@ import
org.apache.flink.table.planner.plan.schema.FlinkPreparingTableBase
import
org.apache.flink.table.planner.plan.utils.WindowJoinUtil.satisfyWindowJoin
import
org.apache.flink.table.planner.plan.utils.WindowUtil.{convertToWindowingStrategy,
groupingContainsWindowStartEnd, isWindowTableFunctionCall}
import org.apache.flink.table.runtime.groupwindow._
+import org.apache.flink.table.types.logical.LogicalType
import org.apache.calcite.plan.hep.HepRelVertex
import org.apache.calcite.plan.volcano.RelSubset
@@ -254,11 +256,34 @@ class FlinkRelMdWindowProperties private extends
MetadataHandler[FlinkMetadata.W
def getWindowProperties(
rel: StreamPhysicalWindowAggregate,
mq: RelMetadataQuery): RelWindowProperties = {
+getWindowAggregateWindowProperties(
+ rel.grouping.length + rel.aggCalls.size(),
+ rel.namedWindowProperties,
+ rel.windowing.getWindow,
+ rel.windowing.getTimeAttributeType
+)
+ }
+
+ def getWindowProperties(
+ rel: StreamPhysicalGlobalWindowAggregate,
+ mq: RelMetadataQuery): RelWindowProperties = {
+getWindowAggregateWindowProperties(
+ rel.grouping.length + rel.aggCalls.size(),
+ rel.namedWindowProperties,
+ rel.windowing.getWindow,
+ rel.windowing.getTimeAttributeType
+)
+ }
+
+ private def getWindowAggregateWindowProperties(
+ propertyOffset: Int,
+ windowProperties: Seq[NamedWindowProperty],
+ windowSpec: WindowSpec,
+ timeAttributeType: LogicalType): RelWindowProperties = {
val starts = ArrayBuffer[Int]()
val ends = ArrayBuffer[Int]()
val times = ArrayBuffer[Int]()
-val propertyOffset = rel.grouping.length + rel.aggCalls.size()
-rel.namedWindowProperties.map(_.getProperty).zipWithIndex.foreach {
+windowProperties.map(_.getProperty).zipWithIndex.foreach {
case (p, index) =>
p match {
case _: WindowStart =>
@@ -275,11 +300,18 @@ class FlinkRelMdWindowProperties private extends
MetadataHandler[FlinkMetadata.W
ImmutableBitSet.of(starts: _*),
ImmutableBitSet.of(ends: _*),
ImmutableBitSet.of(times: _*),
- rel.windowing.getWindow,
- rel.windowing.getTimeAttributeType
+ windowSpec,
+ timeAttributeType
)
}
+ def getWindowProperties(
+ rel: StreamPhysicalLocalWindowAggregate,
+ mq: RelMetadataQuery): RelWindowProperties = {
+val fmq = FlinkRelMetadataQuery.reuseOrCreate(mq)
+fmq.getRelWindowProperties(rel.getInput)
+ }
+
def getWindowProperties(
rel: StreamPhysicalWindowRank,
mq: RelMetada
[flink] branch master updated (753da61bb36 -> 103e5f9fd0f)
This is an automated email from the ASF dual-hosted git repository. renqs pushed a change to branch master in repository https://gitbox.apache.org/repos/asf/flink.git from 753da61bb36 [FLINK-31230][yarn] Improve YarnClusterDescriptor memory unit display. add 103e5f9fd0f [FLINK-31157][release] Add release note for version 1.17 (#22146) No new revisions were added by this update. Summary of changes: docs/content.zh/_index.md | 12 +- docs/content.zh/release-notes/flink-1.17.md | 213 docs/content/_index.md | 13 +- docs/content/release-notes/flink-1.17.md| 213 4 files changed, 449 insertions(+), 2 deletions(-) create mode 100644 docs/content.zh/release-notes/flink-1.17.md create mode 100644 docs/content/release-notes/flink-1.17.md
[flink] branch master updated: [FLINK-31149][docs] Update compatibility matrix for release 1.17 (#22244)
This is an automated email from the ASF dual-hosted git repository. renqs pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/flink.git The following commit(s) were added to refs/heads/master by this push: new 1009196d1ff [FLINK-31149][docs] Update compatibility matrix for release 1.17 (#22244) 1009196d1ff is described below commit 1009196d1ffe2d83ccab46608c3581e25d12fcc3 Author: Qingsheng Ren AuthorDate: Thu Mar 23 14:58:06 2023 +0800 [FLINK-31149][docs] Update compatibility matrix for release 1.17 (#22244) --- docs/content.zh/docs/ops/upgrading.md | 38 +++ docs/content/docs/ops/upgrading.md| 38 +++ 2 files changed, 76 insertions(+) diff --git a/docs/content.zh/docs/ops/upgrading.md b/docs/content.zh/docs/ops/upgrading.md index 3bb53660250..6e6540bb649 100644 --- a/docs/content.zh/docs/ops/upgrading.md +++ b/docs/content.zh/docs/ops/upgrading.md @@ -262,6 +262,7 @@ $ bin/flink run -s :savepointPath [:runArgs] 1.14.x 1.15.x 1.16.x + 1.17.x 限制 @@ -284,6 +285,7 @@ $ bin/flink run -s :savepointPath [:runArgs] + 从 Flink 1.1.x 迁移到 1.2.x+ 的作业的最大并行度目前固定为作业的并行度。 这意味着迁移后无法增加并行度。 在未来的错误修复版本中可能会删除此限制。 @@ -306,6 +308,7 @@ $ bin/flink run -s :savepointPath [:runArgs] O O O + 从 Flink 1.2.x 迁移到 Flink 1.3.x+ 时，不支持同时更改并行度。 迁移到 Flink 1.3.x+ 后， @@ -334,6 +337,7 @@ $ bin/flink run -s :savepointPath [:runArgs] O O O + M如果 Savepoint 包含 Scala 案例类，则从 Flink 1.3.0 迁移到 Flink 1.4.[0,1] 将失败。用户必须直接迁移到 1.4.2+。 @@ -354,6 +358,7 @@ $ bin/flink run -s :savepointPath [:runArgs] O O O + @@ -374,6 +379,7 @@ $ bin/flink run -s :savepointPath [:runArgs] O O O + 在 1.6.x 到 1.6.2 和 1.7.0 版本中恢复使用 1.5.x 创建的广播状态存在一个已知问题：https://issues.apache.org/jira/browse/FLINK-11087";>FLINK-11087. 升级到 1.6.x 或 1.7.x 系列的用户需要 直接迁移到次要版本分别高于 1.6.2 和 1.7.0。 @@ -396,6 +402,7 @@ $ bin/flink run -s :savepointPath [:runArgs] O O O + @@ -416,6 +423,7 @@ $ bin/flink run -s :savepointPath [:runArgs] O O O + @@ -436,6 +444,7 @@ $ bin/flink run -s :savepointPath [:runArgs] O O O + O @@ -456,6 +465,7 @@ $ bin/flink run -s :savepointPath [:runArgs] O O O + O @@ -476,6 +486,7 @@ $ bin/flink run -s :savepointPath [:runArgs] O O O + O @@ -496,6 +507,7 @@ $ bin/flink run -s :savepointPath [:runArgs] O O O + O @@ -516,6 +528,7 @@ $ bin/flink run -s :savepointPath [:runArgs] O O O + O @@ -536,6 +549,7 @@ $ bin/flink run -s :savepointPath [:runArgs] O O O + O Don't upgrade from 1.12.x to 1.13.x with an unaligned checkpoint. Please use a savepoint for migrating. @@ -556,6 +570,7 @@ $ bin/flink run -s :savepointPath [:runArgs] O O O + O @@ -576,6 +591,7 @@ $ bin/flink run -s :savepointPath [:runArgs] O O + O For Table API: 1.15.0 and 1.15.1 generated non-deterministic UIDs for operators that make it difficult/impossible to restore state or upgrade to next patch version. A new @@ -604,6 +620,28 @@ $ bin/flink run -s :savepointPath [:runArgs] O + O + + + + 1.17.x + + + + + + + + + + + + + + + + + O diff --git a/docs/content/docs/ops/upgrading.md b/docs/content/docs/ops/upgrading.md index 827bab87eac..c79bf808125 100644 --- a/docs/content/docs/ops/upgrading.md +++ b/docs/content/docs/ops/upgrading.md @@ -264,6 +264,7 @@ Savepoints are compatible across Flink versions as indicated by the table below: 1.14.x 1.15.x 1.16.x + 1.17.x Limitations @@ -286,6 +287,7 @@ Savepoints are compatible across Flink versions as indicated by the table below: + The maximum parallelism of a job that was migrated from Flink 1.1.x to 1.2.x+ is currentl
