You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by se...@apache.org on 2019/12/08 01:29:30 UTC
[flink] branch master updated (fc37b6d -> 56cd59f)
This is an automated email from the ASF dual-hosted git repository.
sewen pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git.
from fc37b6d [FLINK-9495][kubernetes] Implement ResourceManager for Kubernetes
new 5f5cdeb [hotfix][runtime] Checkstyle fixes in JobVertex
new 8caea1a [FLINK-14454][tests] Re-enable SavepointSerializers.setFailWhenLegacyStateDetected after tests
new 56cd59f [FLINK-14907] [filesystems] Add EnvironmentVariableKeyProvider for Azure Blob Storage
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
docs/ops/filesystems/azure.md | 19 ++++++---
.../fs/azurefs/EnvironmentVariableKeyProvider.java | 48 ++++++++++++++++++++++
.../apache/flink/runtime/jobgraph/JobVertex.java | 4 +-
.../utils/SavepointMigrationTestBase.java | 17 +++++---
.../restore/AbstractOperatorRestoreTestBase.java | 6 +++
5 files changed, 81 insertions(+), 13 deletions(-)
create mode 100644 flink-filesystems/flink-azure-fs-hadoop/src/main/java/org/apache/flink/fs/azurefs/EnvironmentVariableKeyProvider.java
[flink] 02/03: [FLINK-14454][tests] Re-enable
SavepointSerializers.setFailWhenLegacyStateDetected after tests
Posted by se...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
sewen pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 8caea1a8f16a35217c343d867d2b45addc9a0abe
Author: Jiayi <bu...@163.com>
AuthorDate: Sun Oct 20 18:31:18 2019 +0800
[FLINK-14454][tests] Re-enable SavepointSerializers.setFailWhenLegacyStateDetected after tests
This closes #9944
---
.../checkpointing/utils/SavepointMigrationTestBase.java | 17 ++++++++++++-----
.../restore/AbstractOperatorRestoreTestBase.java | 6 ++++++
2 files changed, 18 insertions(+), 5 deletions(-)
diff --git a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/utils/SavepointMigrationTestBase.java b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/utils/SavepointMigrationTestBase.java
index d549c39..eac68b1 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/utils/SavepointMigrationTestBase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/utils/SavepointMigrationTestBase.java
@@ -40,6 +40,7 @@ import org.apache.flink.test.util.TestBaseUtils;
import org.apache.flink.testutils.junit.category.AlsoRunWithLegacyScheduler;
import org.apache.commons.io.FileUtils;
+import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.ClassRule;
import org.junit.Rule;
@@ -66,11 +67,6 @@ import static org.junit.Assert.fail;
@Category(AlsoRunWithLegacyScheduler.class)
public abstract class SavepointMigrationTestBase extends TestBaseUtils {
- @BeforeClass
- public static void before() {
- SavepointSerializers.setFailWhenLegacyStateDetected(false);
- }
-
@ClassRule
public static final TemporaryFolder TEMP_FOLDER = new TemporaryFolder();
@@ -78,8 +74,19 @@ public abstract class SavepointMigrationTestBase extends TestBaseUtils {
public final MiniClusterWithClientResource miniClusterResource;
private static final Logger LOG = LoggerFactory.getLogger(SavepointMigrationTestBase.class);
+
protected static final int DEFAULT_PARALLELISM = 4;
+ @BeforeClass
+ public static void before() {
+ SavepointSerializers.setFailWhenLegacyStateDetected(false);
+ }
+
+ @AfterClass
+ public static void after() {
+ SavepointSerializers.setFailWhenLegacyStateDetected(true);
+ }
+
protected static String getResourceFilename(String filename) {
ClassLoader cl = SavepointMigrationTestBase.class.getClassLoader();
URL resource = cl.getResource(filename);
diff --git a/flink-tests/src/test/java/org/apache/flink/test/state/operator/restore/AbstractOperatorRestoreTestBase.java b/flink-tests/src/test/java/org/apache/flink/test/state/operator/restore/AbstractOperatorRestoreTestBase.java
index 3035c7d..9b00d49 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/state/operator/restore/AbstractOperatorRestoreTestBase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/state/operator/restore/AbstractOperatorRestoreTestBase.java
@@ -41,6 +41,7 @@ import org.apache.flink.testutils.junit.category.AlsoRunWithLegacyScheduler;
import org.apache.flink.util.ExceptionUtils;
import org.apache.flink.util.TestLogger;
+import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Rule;
import org.junit.Test;
@@ -108,6 +109,11 @@ public abstract class AbstractOperatorRestoreTestBase extends TestLogger {
SavepointSerializers.setFailWhenLegacyStateDetected(false);
}
+ @AfterClass
+ public static void after() {
+ SavepointSerializers.setFailWhenLegacyStateDetected(true);
+ }
+
@Test
public void testMigrationAndRestore() throws Throwable {
ClusterClient<?> clusterClient = cluster.getClusterClient();
[flink] 03/03: [FLINK-14907] [filesystems] Add
EnvironmentVariableKeyProvider for Azure Blob Storage
Posted by se...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
sewen pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 56cd59f47e31e58e112e260f0f4b43c45c1aa74a
Author: Konstantin Knauf <kn...@gmail.com>
AuthorDate: Fri Nov 29 16:20:29 2019 +0100
[FLINK-14907] [filesystems] Add EnvironmentVariableKeyProvider for Azure Blob Storage
Also updates the documentation on credential configuration for ABS
This closes #10376
---
docs/ops/filesystems/azure.md | 19 ++++++---
.../fs/azurefs/EnvironmentVariableKeyProvider.java | 48 ++++++++++++++++++++++
2 files changed, 61 insertions(+), 6 deletions(-)
diff --git a/docs/ops/filesystems/azure.md b/docs/ops/filesystems/azure.md
index d721be5..d63c9d6 100644
--- a/docs/ops/filesystems/azure.md
+++ b/docs/ops/filesystems/azure.md
@@ -62,17 +62,24 @@ cp ./opt/flink-azure-fs-hadoop-{{ site.version }}.jar ./plugins/azure-fs-hadoop/
`flink-azure-fs-hadoop` registers default FileSystem wrappers for URIs with the *wasb://* and *wasbs://* (SSL encrypted access) scheme.
-#### Configurations setup
-After setting up the Azure Blob Storage FileSystem wrapper, you need to configure credentials to make sure that Flink is allowed to access Azure Blob Storage.
+### Credentials Configuration
-To allow for easy adoption, you can use the same configuration keys in `flink-conf.yaml` as in Hadoop's `core-site.xml`
+Hadoop's Azure Filesystem supports configuration of credentials via the Hadoop configuration as
+outlined in the [Hadoop Azure Blob Storage documentation](https://hadoop.apache.org/docs/current/hadoop-azure/index.html#Configuring_Credentials).
+For convenience Flink forwards all Flink configurations with a key prefix of `fs.azure` to the
+Hadoop configuration of the filesystem. Consequentially, the azure blob storage key can be configured
+in `flink-conf.yaml` via:
-You can see the configuration keys in the [Hadoop Azure Blob Storage documentation](https://hadoop.apache.org/docs/current/hadoop-azure/index.html#Configuring_Credentials).
+{% highlight yaml %}
+fs.azure.account.key.<account_name>.blob.core.windows.net: <azure_storage_key>
+{% endhighlight %}
-There are some required configurations that must be added to `flink-conf.yaml`:
+Alternatively, the the filesystem can be configured to read the Azure Blob Storage key from an
+environment variable `AZURE_STORAGE_KEY` by setting the following configuration keys in
+`flink-conf.yaml`.
{% highlight yaml %}
-fs.azure.account.key.youraccount.blob.core.windows.net: Azure Blob Storage access key
+fs.azure.account.keyprovider.<account_name>.blob.core.windows.net: org.apache.flink.fs.azurefs.EnvironmentVariableKeyProvider
{% endhighlight %}
{% top %}
diff --git a/flink-filesystems/flink-azure-fs-hadoop/src/main/java/org/apache/flink/fs/azurefs/EnvironmentVariableKeyProvider.java b/flink-filesystems/flink-azure-fs-hadoop/src/main/java/org/apache/flink/fs/azurefs/EnvironmentVariableKeyProvider.java
new file mode 100644
index 0000000..a7cd3fa
--- /dev/null
+++ b/flink-filesystems/flink-azure-fs-hadoop/src/main/java/org/apache/flink/fs/azurefs/EnvironmentVariableKeyProvider.java
@@ -0,0 +1,48 @@
+/*
+ * 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.fs.azurefs;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.azure.KeyProvider;
+import org.apache.hadoop.fs.azure.KeyProviderException;
+
+/**
+ * An implementation of {@link org.apache.hadoop.fs.azure.KeyProvider}, which reads the Azure
+ * storage key from an environment variable named "AZURE_STORAGE_KEY".
+ *
+ */
+public class EnvironmentVariableKeyProvider implements KeyProvider {
+
+ public static final String AZURE_STORAGE_KEY_ENV_VARIABLE = "AZURE_STORAGE_KEY";
+
+ @Override
+ public String getStorageAccountKey(
+ final String s,
+ final Configuration configuration) throws KeyProviderException {
+
+ String azureStorageKey = System.getenv(AZURE_STORAGE_KEY_ENV_VARIABLE);
+
+ if (azureStorageKey != null) {
+ return azureStorageKey;
+ } else {
+ throw new KeyProviderException("Unable to retrieve Azure storage key from environment. \""
+ + AZURE_STORAGE_KEY_ENV_VARIABLE + "\" not set.");
+ }
+ }
+}
[flink] 01/03: [hotfix][runtime] Checkstyle fixes in JobVertex
Posted by se...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
sewen pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 5f5cdeb782ef2596d900de5b90ff2cc7bb62e735
Author: Stephan Ewen <se...@apache.org>
AuthorDate: Fri Nov 15 20:24:39 2019 +0100
[hotfix][runtime] Checkstyle fixes in JobVertex
---
.../src/main/java/org/apache/flink/runtime/jobgraph/JobVertex.java | 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/JobVertex.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/JobVertex.java
index e8c7528..79b6dd3 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/JobVertex.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/JobVertex.java
@@ -62,10 +62,10 @@ public class JobVertex implements java.io.Serializable {
private final ArrayList<OperatorID> operatorIdsAlternatives = new ArrayList<>();
/** List of produced data sets, one per writer. */
- private final ArrayList<IntermediateDataSet> results = new ArrayList<IntermediateDataSet>();
+ private final ArrayList<IntermediateDataSet> results = new ArrayList<>();
/** List of edges with incoming data. One per Reader. */
- private final ArrayList<JobEdge> inputs = new ArrayList<JobEdge>();
+ private final ArrayList<JobEdge> inputs = new ArrayList<>();
/** Number of subtasks to split this task into at runtime. */
private int parallelism = ExecutionConfig.PARALLELISM_DEFAULT;