You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by tz...@apache.org on 2017/08/04 02:33:35 UTC
[1/4] flink git commit: [FLINK-7356] [config] misleading s3 file uri
in configuration file
Repository: flink
Updated Branches:
refs/heads/master 1fabe0cd0 -> 875a1369c
[FLINK-7356] [config] misleading s3 file uri in configuration file
This closes #4466.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/40901ed5
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/40901ed5
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/40901ed5
Branch: refs/heads/master
Commit: 40901ed5866b95b19ba056575a7480fa95b40c68
Parents: 1fabe0c
Author: Bowen Li <bo...@gmail.com>
Authored: Wed Aug 2 12:01:36 2017 -0700
Committer: Tzu-Li (Gordon) Tai <tz...@apache.org>
Committed: Thu Aug 3 23:03:17 2017 +0800
----------------------------------------------------------------------
flink-dist/src/main/resources/flink-conf.yaml | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/40901ed5/flink-dist/src/main/resources/flink-conf.yaml
----------------------------------------------------------------------
diff --git a/flink-dist/src/main/resources/flink-conf.yaml b/flink-dist/src/main/resources/flink-conf.yaml
index 23eb525..2fb1559 100644
--- a/flink-dist/src/main/resources/flink-conf.yaml
+++ b/flink-dist/src/main/resources/flink-conf.yaml
@@ -118,7 +118,7 @@ jobmanager.web.port: 8081
# Directory for storing checkpoints in a Flink-supported filesystem
# Note: State backend must be accessible from the JobManager and all TaskManagers.
# Use "hdfs://" for HDFS setups, "file://" for UNIX/POSIX-compliant file systems,
-# (or any local file system under Windows), or "S3://" for S3 file system.
+# (or any local file system under Windows), or "s3://" with lower case 's' for S3 file system.
#
# state.backend.fs.checkpointdir: hdfs://namenode-host:port/flink-checkpoints
[2/4] flink git commit: [FLINK-7127] [runtime] Add null check for
state meta info snapshotting in heap backend
Posted by tz...@apache.org.
[FLINK-7127] [runtime] Add null check for state meta info snapshotting in heap backend
This closes #4349.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/bef74846
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/bef74846
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/bef74846
Branch: refs/heads/master
Commit: bef7484610dd206d7c749e32f5d91a0f67ef405a
Parents: 40901ed
Author: Dmitrii Kniazev <my...@ya.ru>
Authored: Mon Jul 10 21:41:32 2017 +0300
Committer: Tzu-Li (Gordon) Tai <tz...@apache.org>
Committed: Thu Aug 3 23:03:37 2017 +0800
----------------------------------------------------------------------
.../org/apache/flink/runtime/state/heap/HeapKeyedStateBackend.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/bef74846/flink-runtime/src/main/java/org/apache/flink/runtime/state/heap/HeapKeyedStateBackend.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/heap/HeapKeyedStateBackend.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/heap/HeapKeyedStateBackend.java
index 055274f..d1c0466 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/heap/HeapKeyedStateBackend.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/heap/HeapKeyedStateBackend.java
@@ -314,10 +314,10 @@ public class HeapKeyedStateBackend<K> extends AbstractKeyedStateBackend<K> {
final Map<StateTable<K, ?, ?>, StateTableSnapshot> cowStateStableSnapshots = new HashedMap(stateTables.size());
for (Map.Entry<String, StateTable<K, ?, ?>> kvState : stateTables.entrySet()) {
- metaInfoSnapshots.add(kvState.getValue().getMetaInfo().snapshot());
kVStateToId.put(kvState.getKey(), kVStateToId.size());
StateTable<K, ?, ?> stateTable = kvState.getValue();
if (null != stateTable) {
+ metaInfoSnapshots.add(stateTable.getMetaInfo().snapshot());
cowStateStableSnapshots.put(stateTable, stateTable.createSnapshot());
}
}
[4/4] flink git commit: [FLINK-6493] Fix ineffective null check in
equality of RegisteredKeyedBackendStateMetaInfo snapshots
Posted by tz...@apache.org.
[FLINK-6493] Fix ineffective null check in equality of RegisteredKeyedBackendStateMetaInfo snapshots
This closes #4328.
This closes #4408.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/875a1369
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/875a1369
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/875a1369
Branch: refs/heads/master
Commit: 875a1369c2ecfd721e3797f59f7ae29e7c522840
Parents: 2248103
Author: Tzu-Li (Gordon) Tai <tz...@apache.org>
Authored: Thu Aug 3 23:14:34 2017 +0800
Committer: Tzu-Li (Gordon) Tai <tz...@apache.org>
Committed: Thu Aug 3 23:14:34 2017 +0800
----------------------------------------------------------------------
.../state/RegisteredKeyedBackendStateMetaInfo.java | 15 ++++++---------
.../RegisteredOperatorBackendStateMetaInfo.java | 7 ++++---
2 files changed, 10 insertions(+), 12 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/875a1369/flink-runtime/src/main/java/org/apache/flink/runtime/state/RegisteredKeyedBackendStateMetaInfo.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/RegisteredKeyedBackendStateMetaInfo.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/RegisteredKeyedBackendStateMetaInfo.java
index e1a7e06..342bc7f 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/RegisteredKeyedBackendStateMetaInfo.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/RegisteredKeyedBackendStateMetaInfo.java
@@ -23,6 +23,8 @@ import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.api.common.typeutils.TypeSerializerConfigSnapshot;
import org.apache.flink.util.Preconditions;
+import java.util.Objects;
+
import static org.apache.flink.util.Preconditions.checkNotNull;
/**
@@ -220,15 +222,10 @@ public class RegisteredKeyedBackendStateMetaInfo<N, S> {
}
// need to check for nulls because serializer and config snapshots may be null on restore
- return
- ((getStateSerializer() == null && that.getStateSerializer() == null)
- || getStateSerializer().equals(that.getStateSerializer()))
- && ((getNamespaceSerializer() == null && that.getNamespaceSerializer() == null)
- || getNamespaceSerializer().equals(that.getNamespaceSerializer()))
- && ((getNamespaceSerializerConfigSnapshot() == null && that.getNamespaceSerializerConfigSnapshot() == null)
- || getNamespaceSerializerConfigSnapshot().equals(that.getNamespaceSerializerConfigSnapshot()))
- && ((getStateSerializerConfigSnapshot() == null && that.getStateSerializerConfigSnapshot() == null)
- || getStateSerializerConfigSnapshot().equals(that.getStateSerializerConfigSnapshot()));
+ return Objects.equals(getStateSerializer(), that.getStateSerializer())
+ && Objects.equals(getNamespaceSerializer(), that.getNamespaceSerializer())
+ && Objects.equals(getNamespaceSerializerConfigSnapshot(), that.getNamespaceSerializerConfigSnapshot())
+ && Objects.equals(getStateSerializerConfigSnapshot(), that.getStateSerializerConfigSnapshot());
}
@Override
http://git-wip-us.apache.org/repos/asf/flink/blob/875a1369/flink-runtime/src/main/java/org/apache/flink/runtime/state/RegisteredOperatorBackendStateMetaInfo.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/RegisteredOperatorBackendStateMetaInfo.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/RegisteredOperatorBackendStateMetaInfo.java
index 7c56851..af289f9 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/RegisteredOperatorBackendStateMetaInfo.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/RegisteredOperatorBackendStateMetaInfo.java
@@ -18,11 +18,12 @@
package org.apache.flink.runtime.state;
-import java.util.Objects;
import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.api.common.typeutils.TypeSerializerConfigSnapshot;
import org.apache.flink.util.Preconditions;
+import java.util.Objects;
+
/**
* Compound meta information for a registered state in an operator state backend.
* This contains the state name, assignment mode, and state partition serializer.
@@ -185,8 +186,8 @@ public class RegisteredOperatorBackendStateMetaInfo<S> {
// need to check for nulls because serializer and config snapshots may be null on restore
return name.equals(snapshot.getName())
&& assignmentMode.equals(snapshot.getAssignmentMode())
- && (Objects.equals(partitionStateSerializer, snapshot.getPartitionStateSerializer()))
- && (Objects.equals(partitionStateSerializerConfigSnapshot, snapshot.getPartitionStateSerializerConfigSnapshot()));
+ && Objects.equals(partitionStateSerializer, snapshot.getPartitionStateSerializer())
+ && Objects.equals(partitionStateSerializerConfigSnapshot, snapshot.getPartitionStateSerializerConfigSnapshot());
}
@Override
[3/4] flink git commit: [FLINK-6493] Fix ineffective null check in
RegisteredOperatorBackendStateMetaInfo#equals
Posted by tz...@apache.org.
[FLINK-6493] Fix ineffective null check in RegisteredOperatorBackendStateMetaInfo#equals
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/2248103e
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/2248103e
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/2248103e
Branch: refs/heads/master
Commit: 2248103e5b374507f2fb67c180c928352ae7f02b
Parents: bef7484
Author: zhangminglei <zm...@163.com>
Authored: Thu Aug 3 20:32:35 2017 +0800
Committer: Tzu-Li (Gordon) Tai <tz...@apache.org>
Committed: Thu Aug 3 23:14:04 2017 +0800
----------------------------------------------------------------------
.../RegisteredOperatorBackendStateMetaInfo.java | 18 +++++++++++-------
1 file changed, 11 insertions(+), 7 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/2248103e/flink-runtime/src/main/java/org/apache/flink/runtime/state/RegisteredOperatorBackendStateMetaInfo.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/RegisteredOperatorBackendStateMetaInfo.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/RegisteredOperatorBackendStateMetaInfo.java
index b43fc9c..7c56851 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/RegisteredOperatorBackendStateMetaInfo.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/RegisteredOperatorBackendStateMetaInfo.java
@@ -18,6 +18,7 @@
package org.apache.flink.runtime.state;
+import java.util.Objects;
import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.api.common.typeutils.TypeSerializerConfigSnapshot;
import org.apache.flink.util.Preconditions;
@@ -175,14 +176,17 @@ public class RegisteredOperatorBackendStateMetaInfo<S> {
return false;
}
+ if (!(obj instanceof Snapshot)) {
+ return false;
+ }
+
+ Snapshot snapshot = (Snapshot)obj;
+
// need to check for nulls because serializer and config snapshots may be null on restore
- return (obj instanceof Snapshot)
- && name.equals(((Snapshot) obj).getName())
- && assignmentMode.equals(((Snapshot) obj).getAssignmentMode())
- && ((partitionStateSerializer == null && ((Snapshot) obj).getPartitionStateSerializer() == null)
- || partitionStateSerializer.equals(((Snapshot) obj).getPartitionStateSerializer()))
- && ((partitionStateSerializerConfigSnapshot == null && ((Snapshot) obj).getPartitionStateSerializerConfigSnapshot() == null)
- || partitionStateSerializerConfigSnapshot.equals(((Snapshot) obj).getPartitionStateSerializerConfigSnapshot()));
+ return name.equals(snapshot.getName())
+ && assignmentMode.equals(snapshot.getAssignmentMode())
+ && (Objects.equals(partitionStateSerializer, snapshot.getPartitionStateSerializer()))
+ && (Objects.equals(partitionStateSerializerConfigSnapshot, snapshot.getPartitionStateSerializerConfigSnapshot()));
}
@Override