You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ch...@apache.org on 2017/06/25 06:44:44 UTC
[01/21] flink git commit: [FLINK-6541] Improve tmp dir setup in
TM/WebMonitor
Repository: flink
Updated Branches:
refs/heads/master a0b781461 -> 1ae4ac28b
[FLINK-6541] Improve tmp dir setup in TM/WebMonitor
This closes #3894.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/72164077
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/72164077
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/72164077
Branch: refs/heads/master
Commit: 721640778aa88b6491baf837278a873493f283f0
Parents: 60222a2
Author: zentol <ch...@apache.org>
Authored: Sat May 13 18:09:30 2017 +0200
Committer: zentol <ch...@apache.org>
Committed: Fri Jun 23 14:14:29 2017 +0200
----------------------------------------------------------------------
.../apache/flink/runtime/webmonitor/WebRuntimeMonitor.java | 2 +-
.../flink/runtime/taskexecutor/TaskManagerServices.java | 8 +++++---
2 files changed, 6 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/72164077/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/WebRuntimeMonitor.java
----------------------------------------------------------------------
diff --git a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/WebRuntimeMonitor.java b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/WebRuntimeMonitor.java
index 535cc71..dbaaaf2 100644
--- a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/WebRuntimeMonitor.java
+++ b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/WebRuntimeMonitor.java
@@ -175,7 +175,7 @@ public class WebRuntimeMonitor implements WebMonitor {
// create storage for uploads
this.uploadDir = getUploadDir(config);
// the upload directory should either 1. exist and writable or 2. can be created and writable
- if (!(uploadDir.exists() && uploadDir.canWrite()) && !(uploadDir.mkdir() && uploadDir.canWrite())) {
+ if (!(uploadDir.exists() && uploadDir.canWrite()) && !(uploadDir.mkdirs() && uploadDir.canWrite())) {
throw new IOException(
String.format("Jar upload directory %s cannot be created or is not writable.",
uploadDir.getAbsolutePath()));
http://git-wip-us.apache.org/repos/asf/flink/blob/72164077/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerServices.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerServices.java b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerServices.java
index 86a2fdf..1c30ff6 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerServices.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerServices.java
@@ -603,11 +603,11 @@ public class TaskManagerServices {
}
/**
- * Validates that all the directories denoted by the strings do actually exist, are proper
+ * Validates that all the directories denoted by the strings do actually exist or can be created, are proper
* directories (not files), and are writable.
*
* @param tmpDirs The array of directory paths to check.
- * @throws IOException Thrown if any of the directories does not exist or is not writable
+ * @throws IOException Thrown if any of the directories does not exist and cannot be created or is not writable
* or is a file, rather than a directory.
*/
private static void checkTempDirs(String[] tmpDirs) throws IOException {
@@ -615,7 +615,9 @@ public class TaskManagerServices {
if (dir != null && !dir.equals("")) {
File file = new File(dir);
if (!file.exists()) {
- throw new IOException("Temporary file directory " + file.getAbsolutePath() + " does not exist.");
+ if (!file.mkdirs()) {
+ throw new IOException("Temporary file directory " + file.getAbsolutePath() + " does not exist and could not be created.");
+ }
}
if (!file.isDirectory()) {
throw new IOException("Temporary file directory " + file.getAbsolutePath() + " is not a directory.");
[20/21] flink git commit: [FLINK-6389] [connector] Upgrade hbase
dependency to 1.3.1
Posted by ch...@apache.org.
[FLINK-6389] [connector] Upgrade hbase dependency to 1.3.1
This closes #4151.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/5604930c
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/5604930c
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/5604930c
Branch: refs/heads/master
Commit: 5604930c010b32200facc4c86dfb26c64120349e
Parents: b863b90
Author: zhangminglei <zm...@163.com>
Authored: Wed Jun 21 09:31:06 2017 +0800
Committer: zentol <ch...@apache.org>
Committed: Sat Jun 24 22:09:07 2017 +0200
----------------------------------------------------------------------
flink-connectors/flink-hbase/pom.xml | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/5604930c/flink-connectors/flink-hbase/pom.xml
----------------------------------------------------------------------
diff --git a/flink-connectors/flink-hbase/pom.xml b/flink-connectors/flink-hbase/pom.xml
index 949e27a..b900d23 100644
--- a/flink-connectors/flink-hbase/pom.xml
+++ b/flink-connectors/flink-hbase/pom.xml
@@ -34,7 +34,7 @@ under the License.
<packaging>jar</packaging>
<properties>
- <hbase.version>1.3.0</hbase.version>
+ <hbase.version>1.3.1</hbase.version>
</properties>
<build>
[05/21] flink git commit: [FLINK-6788] Remove unsused
GenericFlatTypePostPass/AbstractSchema classes
Posted by ch...@apache.org.
[FLINK-6788] Remove unsused GenericFlatTypePostPass/AbstractSchema classes
This closes #4118.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/af0ced98
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/af0ced98
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/af0ced98
Branch: refs/heads/master
Commit: af0ced984ba9f13302f1d4c4d4b064bd580cc29c
Parents: e1269ed
Author: wangmiao1981 <wm...@hotmail.com>
Authored: Tue Jun 13 21:25:02 2017 +0800
Committer: zentol <ch...@apache.org>
Committed: Fri Jun 23 14:14:29 2017 +0200
----------------------------------------------------------------------
.../optimizer/postpass/AbstractSchema.java | 39 --
.../postpass/GenericFlatTypePostPass.java | 579 -------------------
2 files changed, 618 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/af0ced98/flink-optimizer/src/main/java/org/apache/flink/optimizer/postpass/AbstractSchema.java
----------------------------------------------------------------------
diff --git a/flink-optimizer/src/main/java/org/apache/flink/optimizer/postpass/AbstractSchema.java b/flink-optimizer/src/main/java/org/apache/flink/optimizer/postpass/AbstractSchema.java
deleted file mode 100644
index f2b736c..0000000
--- a/flink-optimizer/src/main/java/org/apache/flink/optimizer/postpass/AbstractSchema.java
+++ /dev/null
@@ -1,39 +0,0 @@
-/*
- * 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.optimizer.postpass;
-
-import java.util.Map;
-
-public abstract class AbstractSchema<X> implements Iterable<Map.Entry<Integer, X>> {
-
- private int numConnectionsThatContributed;
-
-
- public int getNumConnectionsThatContributed() {
- return this.numConnectionsThatContributed;
- }
-
- public void increaseNumConnectionsThatContributed() {
- this.numConnectionsThatContributed++;
- }
-
- public abstract void addType(int pos, X type) throws ConflictingFieldTypeInfoException;
-
- public abstract X getType(int field);
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/af0ced98/flink-optimizer/src/main/java/org/apache/flink/optimizer/postpass/GenericFlatTypePostPass.java
----------------------------------------------------------------------
diff --git a/flink-optimizer/src/main/java/org/apache/flink/optimizer/postpass/GenericFlatTypePostPass.java b/flink-optimizer/src/main/java/org/apache/flink/optimizer/postpass/GenericFlatTypePostPass.java
deleted file mode 100644
index 2d8377e..0000000
--- a/flink-optimizer/src/main/java/org/apache/flink/optimizer/postpass/GenericFlatTypePostPass.java
+++ /dev/null
@@ -1,579 +0,0 @@
-/*
- * 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.optimizer.postpass;
-
-import java.util.Map;
-
-import org.apache.flink.api.common.operators.SemanticProperties;
-import org.apache.flink.api.common.operators.util.FieldList;
-import org.apache.flink.api.common.typeutils.TypeComparatorFactory;
-import org.apache.flink.api.common.typeutils.TypePairComparatorFactory;
-import org.apache.flink.api.common.typeutils.TypeSerializerFactory;
-import org.apache.flink.optimizer.CompilerException;
-import org.apache.flink.optimizer.CompilerPostPassException;
-import org.apache.flink.optimizer.dag.OptimizerNode;
-import org.apache.flink.optimizer.dag.SingleInputNode;
-import org.apache.flink.optimizer.dag.TwoInputNode;
-import org.apache.flink.optimizer.dag.WorksetIterationNode;
-import org.apache.flink.optimizer.plan.BulkIterationPlanNode;
-import org.apache.flink.optimizer.plan.BulkPartialSolutionPlanNode;
-import org.apache.flink.optimizer.plan.Channel;
-import org.apache.flink.optimizer.plan.DualInputPlanNode;
-import org.apache.flink.optimizer.plan.NAryUnionPlanNode;
-import org.apache.flink.optimizer.plan.OptimizedPlan;
-import org.apache.flink.optimizer.plan.PlanNode;
-import org.apache.flink.optimizer.plan.SingleInputPlanNode;
-import org.apache.flink.optimizer.plan.SinkPlanNode;
-import org.apache.flink.optimizer.plan.SolutionSetPlanNode;
-import org.apache.flink.optimizer.plan.SourcePlanNode;
-import org.apache.flink.optimizer.plan.WorksetIterationPlanNode;
-import org.apache.flink.optimizer.plan.WorksetPlanNode;
-
-/**
- *
- */
-public abstract class GenericFlatTypePostPass<X, T extends AbstractSchema<X>> implements OptimizerPostPass {
-
- private boolean propagateParentSchemaDown = true;
-
-
- public boolean isPropagateParentSchemaDown() {
- return propagateParentSchemaDown;
- }
-
- public void setPropagateParentSchemaDown(boolean propagateParentSchemaDown) {
- this.propagateParentSchemaDown = propagateParentSchemaDown;
- }
-
- // --------------------------------------------------------------------------------------------
- // Generic schema inferring traversal
- // --------------------------------------------------------------------------------------------
-
- @Override
- public void postPass(OptimizedPlan plan) {
- for (SinkPlanNode sink : plan.getDataSinks()) {
- traverse(sink, null, true);
- }
- }
-
- @SuppressWarnings("unchecked")
- protected void traverse(PlanNode node, T parentSchema, boolean createUtilities) {
- // distinguish the node types
- if (node instanceof SinkPlanNode) {
- SinkPlanNode sn = (SinkPlanNode) node;
- Channel inchannel = sn.getInput();
-
- T schema = createEmptySchema();
- sn.postPassHelper = schema;
-
- // add the sinks information to the schema
- try {
- getSinkSchema(sn, schema);
- }
- catch (ConflictingFieldTypeInfoException e) {
- throw new CompilerPostPassException("Conflicting type infomation for the data sink '" +
- sn.getSinkNode().getOperator().getName() + "'.");
- }
-
- // descend to the input channel
- try {
- propagateToChannel(schema, inchannel, createUtilities);
- }
- catch (MissingFieldTypeInfoException ex) {
- throw new CompilerPostPassException("Missing type infomation for the channel that inputs to the data sink '" +
- sn.getSinkNode().getOperator().getName() + "'.");
- }
- }
- else if (node instanceof SourcePlanNode) {
- if (createUtilities) {
- ((SourcePlanNode) node).setSerializer(createSerializer(parentSchema, node));
- // nothing else to be done here. the source has no input and no strategy itself
- }
- }
- else if (node instanceof BulkIterationPlanNode) {
- BulkIterationPlanNode iterationNode = (BulkIterationPlanNode) node;
-
- // get the nodes current schema
- T schema;
- if (iterationNode.postPassHelper == null) {
- schema = createEmptySchema();
- iterationNode.postPassHelper = schema;
- } else {
- schema = (T) iterationNode.postPassHelper;
- }
- schema.increaseNumConnectionsThatContributed();
-
- // add the parent schema to the schema
- if (propagateParentSchemaDown) {
- addSchemaToSchema(parentSchema, schema, iterationNode.getProgramOperator().getName());
- }
-
- // check whether all outgoing channels have not yet contributed. come back later if not.
- if (schema.getNumConnectionsThatContributed() < iterationNode.getOutgoingChannels().size()) {
- return;
- }
-
- if (iterationNode.getRootOfStepFunction() instanceof NAryUnionPlanNode) {
- throw new CompilerException("Optimizer cannot compile an iteration step function where next partial solution is created by a Union node.");
- }
-
- // traverse the termination criterion for the first time. create schema only, no utilities. Needed in case of intermediate termination criterion
- if (iterationNode.getRootOfTerminationCriterion() != null) {
- SingleInputPlanNode addMapper = (SingleInputPlanNode) iterationNode.getRootOfTerminationCriterion();
- traverse(addMapper.getInput().getSource(), createEmptySchema(), false);
- try {
- addMapper.getInput().setSerializer(createSerializer(createEmptySchema()));
- } catch (MissingFieldTypeInfoException e) {
- throw new RuntimeException(e);
- }
- }
-
- // traverse the step function for the first time. create schema only, no utilities
- traverse(iterationNode.getRootOfStepFunction(), schema, false);
-
- T pss = (T) iterationNode.getPartialSolutionPlanNode().postPassHelper;
- if (pss == null) {
- throw new CompilerException("Error in Optimizer Post Pass: Partial solution schema is null after first traversal of the step function.");
- }
-
- // traverse the step function for the second time, taking the schema of the partial solution
- traverse(iterationNode.getRootOfStepFunction(), pss, createUtilities);
-
- if (iterationNode.getRootOfTerminationCriterion() != null) {
- SingleInputPlanNode addMapper = (SingleInputPlanNode) iterationNode.getRootOfTerminationCriterion();
- traverse(addMapper.getInput().getSource(), createEmptySchema(), createUtilities);
- try {
- addMapper.getInput().setSerializer(createSerializer(createEmptySchema()));
- } catch (MissingFieldTypeInfoException e) {
- throw new RuntimeException(e);
- }
- }
-
- // take the schema from the partial solution node and add its fields to the iteration result schema.
- // input and output schema need to be identical, so this is essentially a sanity check
- addSchemaToSchema(pss, schema, iterationNode.getProgramOperator().getName());
-
- // set the serializer
- if (createUtilities) {
- iterationNode.setSerializerForIterationChannel(createSerializer(pss, iterationNode.getPartialSolutionPlanNode()));
- }
-
- // done, we can now propagate our info down
- try {
- propagateToChannel(schema, iterationNode.getInput(), createUtilities);
- } catch (MissingFieldTypeInfoException e) {
- throw new CompilerPostPassException("Could not set up runtime strategy for input channel to node '"
- + iterationNode.getProgramOperator().getName() + "'. Missing type information for key field " +
- e.getFieldNumber());
- }
- }
- else if (node instanceof WorksetIterationPlanNode) {
- WorksetIterationPlanNode iterationNode = (WorksetIterationPlanNode) node;
-
- // get the nodes current schema
- T schema;
- if (iterationNode.postPassHelper == null) {
- schema = createEmptySchema();
- iterationNode.postPassHelper = schema;
- } else {
- schema = (T) iterationNode.postPassHelper;
- }
- schema.increaseNumConnectionsThatContributed();
-
- // add the parent schema to the schema (which refers to the solution set schema)
- if (propagateParentSchemaDown) {
- addSchemaToSchema(parentSchema, schema, iterationNode.getProgramOperator().getName());
- }
-
- // check whether all outgoing channels have not yet contributed. come back later if not.
- if (schema.getNumConnectionsThatContributed() < iterationNode.getOutgoingChannels().size()) {
- return;
- }
- if (iterationNode.getNextWorkSetPlanNode() instanceof NAryUnionPlanNode) {
- throw new CompilerException("Optimizer cannot compile a workset iteration step function where the next workset is produced by a Union node.");
- }
- if (iterationNode.getSolutionSetDeltaPlanNode() instanceof NAryUnionPlanNode) {
- throw new CompilerException("Optimizer cannot compile a workset iteration step function where the solution set delta is produced by a Union node.");
- }
-
- // traverse the step function
- // pass an empty schema to the next workset and the parent schema to the solution set delta
- // these first traversals are schema only
- traverse(iterationNode.getNextWorkSetPlanNode(), createEmptySchema(), false);
- traverse(iterationNode.getSolutionSetDeltaPlanNode(), schema, false);
-
- T wss = (T) iterationNode.getWorksetPlanNode().postPassHelper;
- T sss = (T) iterationNode.getSolutionSetPlanNode().postPassHelper;
-
- if (wss == null) {
- throw new CompilerException("Error in Optimizer Post Pass: Workset schema is null after first traversal of the step function.");
- }
- if (sss == null) {
- throw new CompilerException("Error in Optimizer Post Pass: Solution set schema is null after first traversal of the step function.");
- }
-
- // make the second pass and instantiate the utilities
- traverse(iterationNode.getNextWorkSetPlanNode(), wss, createUtilities);
- traverse(iterationNode.getSolutionSetDeltaPlanNode(), sss, createUtilities);
-
- // add the types from the solution set schema to the iteration's own schema. since
- // the solution set input and the result must have the same schema, this acts as a sanity check.
- try {
- for (Map.Entry<Integer, X> entry : sss) {
- Integer pos = entry.getKey();
- schema.addType(pos, entry.getValue());
- }
- } catch (ConflictingFieldTypeInfoException e) {
- throw new CompilerPostPassException("Conflicting type information for field " + e.getFieldNumber()
- + " in node '" + iterationNode.getProgramOperator().getName() + "'. Contradicting types between the " +
- "result of the iteration and the solution set schema: " + e.getPreviousType() +
- " and " + e.getNewType() + ". Most probable cause: Invalid constant field annotations.");
- }
-
- // set the serializers and comparators
- if (createUtilities) {
- WorksetIterationNode optNode = iterationNode.getIterationNode();
- iterationNode.setWorksetSerializer(createSerializer(wss, iterationNode.getWorksetPlanNode()));
- iterationNode.setSolutionSetSerializer(createSerializer(sss, iterationNode.getSolutionSetPlanNode()));
- try {
- iterationNode.setSolutionSetComparator(createComparator(optNode.getSolutionSetKeyFields(), null, sss));
- } catch (MissingFieldTypeInfoException ex) {
- throw new CompilerPostPassException("Could not set up the solution set for workset iteration '" +
- optNode.getOperator().getName() + "'. Missing type information for key field " + ex.getFieldNumber() + '.');
- }
- }
-
- // done, we can now propagate our info down
- try {
- propagateToChannel(schema, iterationNode.getInitialSolutionSetInput(), createUtilities);
- propagateToChannel(wss, iterationNode.getInitialWorksetInput(), createUtilities);
- } catch (MissingFieldTypeInfoException ex) {
- throw new CompilerPostPassException("Could not set up runtime strategy for input channel to node '"
- + iterationNode.getProgramOperator().getName() + "'. Missing type information for key field " +
- ex.getFieldNumber());
- }
- }
- else if (node instanceof SingleInputPlanNode) {
- SingleInputPlanNode sn = (SingleInputPlanNode) node;
-
- // get the nodes current schema
- T schema;
- if (sn.postPassHelper == null) {
- schema = createEmptySchema();
- sn.postPassHelper = schema;
- } else {
- schema = (T) sn.postPassHelper;
- }
- schema.increaseNumConnectionsThatContributed();
- SingleInputNode optNode = sn.getSingleInputNode();
-
- // add the parent schema to the schema
- if (propagateParentSchemaDown) {
- addSchemaToSchema(parentSchema, schema, optNode, 0);
- }
-
- // check whether all outgoing channels have not yet contributed. come back later if not.
- if (schema.getNumConnectionsThatContributed() < sn.getOutgoingChannels().size()) {
- return;
- }
-
- // add the nodes local information
- try {
- getSingleInputNodeSchema(sn, schema);
- } catch (ConflictingFieldTypeInfoException e) {
- throw new CompilerPostPassException(getConflictingTypeErrorMessage(e, optNode.getOperator().getName()));
- }
-
- if (createUtilities) {
- // parameterize the node's driver strategy
- for(int i=0;i<sn.getDriverStrategy().getNumRequiredComparators();i++) {
- try {
- sn.setComparator(createComparator(sn.getKeys(i), sn.getSortOrders(i), schema),i);
- } catch (MissingFieldTypeInfoException e) {
- throw new CompilerPostPassException("Could not set up runtime strategy for node '" +
- optNode.getOperator().getName() + "'. Missing type information for key field " +
- e.getFieldNumber());
- }
- }
- }
-
- // done, we can now propagate our info down
- try {
- propagateToChannel(schema, sn.getInput(), createUtilities);
- } catch (MissingFieldTypeInfoException e) {
- throw new CompilerPostPassException("Could not set up runtime strategy for input channel to node '" +
- optNode.getOperator().getName() + "'. Missing type information for field " + e.getFieldNumber());
- }
-
- // don't forget the broadcast inputs
- for (Channel c: sn.getBroadcastInputs()) {
- try {
- propagateToChannel(createEmptySchema(), c, createUtilities);
- } catch (MissingFieldTypeInfoException e) {
- throw new CompilerPostPassException("Could not set up runtime strategy for broadcast channel in node '" +
- optNode.getOperator().getName() + "'. Missing type information for field " + e.getFieldNumber());
- }
- }
- }
- else if (node instanceof DualInputPlanNode) {
- DualInputPlanNode dn = (DualInputPlanNode) node;
-
- // get the nodes current schema
- T schema1;
- T schema2;
- if (dn.postPassHelper1 == null) {
- schema1 = createEmptySchema();
- schema2 = createEmptySchema();
- dn.postPassHelper1 = schema1;
- dn.postPassHelper2 = schema2;
- } else {
- schema1 = (T) dn.postPassHelper1;
- schema2 = (T) dn.postPassHelper2;
- }
-
- schema1.increaseNumConnectionsThatContributed();
- schema2.increaseNumConnectionsThatContributed();
- TwoInputNode optNode = dn.getTwoInputNode();
-
- // add the parent schema to the schema
- if (propagateParentSchemaDown) {
- addSchemaToSchema(parentSchema, schema1, optNode, 0);
- addSchemaToSchema(parentSchema, schema2, optNode, 1);
- }
-
- // check whether all outgoing channels have not yet contributed. come back later if not.
- if (schema1.getNumConnectionsThatContributed() < dn.getOutgoingChannels().size()) {
- return;
- }
-
- // add the nodes local information
- try {
- getDualInputNodeSchema(dn, schema1, schema2);
- } catch (ConflictingFieldTypeInfoException e) {
- throw new CompilerPostPassException(getConflictingTypeErrorMessage(e, optNode.getOperator().getName()));
- }
-
- // parameterize the node's driver strategy
- if (createUtilities) {
- if (dn.getDriverStrategy().getNumRequiredComparators() > 0) {
- // set the individual comparators
- try {
- dn.setComparator1(createComparator(dn.getKeysForInput1(), dn.getSortOrders(), schema1));
- dn.setComparator2(createComparator(dn.getKeysForInput2(), dn.getSortOrders(), schema2));
- } catch (MissingFieldTypeInfoException e) {
- throw new CompilerPostPassException("Could not set up runtime strategy for node '" +
- optNode.getOperator().getName() + "'. Missing type information for field " + e.getFieldNumber());
- }
-
- // set the pair comparator
- try {
- dn.setPairComparator(createPairComparator(dn.getKeysForInput1(), dn.getKeysForInput2(),
- dn.getSortOrders(), schema1, schema2));
- } catch (MissingFieldTypeInfoException e) {
- throw new CompilerPostPassException("Could not set up runtime strategy for node '" +
- optNode.getOperator().getName() + "'. Missing type information for field " + e.getFieldNumber());
- }
-
- }
- }
-
- // done, we can now propagate our info down
- try {
- propagateToChannel(schema1, dn.getInput1(), createUtilities);
- } catch (MissingFieldTypeInfoException e) {
- throw new CompilerPostPassException("Could not set up runtime strategy for the first input channel to node '"
- + optNode.getOperator().getName() + "'. Missing type information for field " + e.getFieldNumber());
- }
- try {
- propagateToChannel(schema2, dn.getInput2(), createUtilities);
- } catch (MissingFieldTypeInfoException e) {
- throw new CompilerPostPassException("Could not set up runtime strategy for the second input channel to node '"
- + optNode.getOperator().getName() + "'. Missing type information for field " + e.getFieldNumber());
- }
-
- // don't forget the broadcast inputs
- for (Channel c: dn.getBroadcastInputs()) {
- try {
- propagateToChannel(createEmptySchema(), c, createUtilities);
- } catch (MissingFieldTypeInfoException e) {
- throw new CompilerPostPassException("Could not set up runtime strategy for broadcast channel in node '" +
- optNode.getOperator().getName() + "'. Missing type information for field " + e.getFieldNumber());
- }
- }
- }
- else if (node instanceof NAryUnionPlanNode) {
- // only propagate the info down
- try {
- for (Channel channel : node.getInputs()) {
- propagateToChannel(parentSchema, channel, createUtilities);
- }
- } catch (MissingFieldTypeInfoException ex) {
- throw new CompilerPostPassException("Could not set up runtime strategy for the input channel to " +
- " a union node. Missing type information for field " + ex.getFieldNumber());
- }
- }
- // catch the sources of the iterative step functions
- else if (node instanceof BulkPartialSolutionPlanNode ||
- node instanceof SolutionSetPlanNode ||
- node instanceof WorksetPlanNode)
- {
- // get the nodes current schema
- T schema;
- String name;
- if (node instanceof BulkPartialSolutionPlanNode) {
- BulkPartialSolutionPlanNode psn = (BulkPartialSolutionPlanNode) node;
- if (psn.postPassHelper == null) {
- schema = createEmptySchema();
- psn.postPassHelper = schema;
- } else {
- schema = (T) psn.postPassHelper;
- }
- name = "partial solution of bulk iteration '" +
- psn.getPartialSolutionNode().getIterationNode().getOperator().getName() + "'";
- }
- else if (node instanceof SolutionSetPlanNode) {
- SolutionSetPlanNode ssn = (SolutionSetPlanNode) node;
- if (ssn.postPassHelper == null) {
- schema = createEmptySchema();
- ssn.postPassHelper = schema;
- } else {
- schema = (T) ssn.postPassHelper;
- }
- name = "solution set of workset iteration '" +
- ssn.getSolutionSetNode().getIterationNode().getOperator().getName() + "'";
- }
- else if (node instanceof WorksetPlanNode) {
- WorksetPlanNode wsn = (WorksetPlanNode) node;
- if (wsn.postPassHelper == null) {
- schema = createEmptySchema();
- wsn.postPassHelper = schema;
- } else {
- schema = (T) wsn.postPassHelper;
- }
- name = "workset of workset iteration '" +
- wsn.getWorksetNode().getIterationNode().getOperator().getName() + "'";
- } else {
- throw new CompilerException();
- }
-
- schema.increaseNumConnectionsThatContributed();
-
- // add the parent schema to the schema
- addSchemaToSchema(parentSchema, schema, name);
- }
- else {
- throw new CompilerPostPassException("Unknown node type encountered: " + node.getClass().getName());
- }
- }
-
- private void propagateToChannel(T schema, Channel channel, boolean createUtilities) throws MissingFieldTypeInfoException {
- if (createUtilities) {
- // the serializer always exists
- channel.setSerializer(createSerializer(schema));
-
- // parameterize the ship strategy
- if (channel.getShipStrategy().requiresComparator()) {
- channel.setShipStrategyComparator(
- createComparator(channel.getShipStrategyKeys(), channel.getShipStrategySortOrder(), schema));
- }
-
- // parameterize the local strategy
- if (channel.getLocalStrategy().requiresComparator()) {
- channel.setLocalStrategyComparator(
- createComparator(channel.getLocalStrategyKeys(), channel.getLocalStrategySortOrder(), schema));
- }
- }
-
- // propagate the channel's source model
- traverse(channel.getSource(), schema, createUtilities);
- }
-
- private void addSchemaToSchema(T sourceSchema, T targetSchema, String opName) {
- try {
- for (Map.Entry<Integer, X> entry : sourceSchema) {
- Integer pos = entry.getKey();
- targetSchema.addType(pos, entry.getValue());
- }
- } catch (ConflictingFieldTypeInfoException e) {
- throw new CompilerPostPassException("Conflicting type information for field " + e.getFieldNumber()
- + " in node '" + opName + "' propagated from successor node. " +
- "Conflicting types: " + e.getPreviousType() + " and " + e.getNewType() +
- ". Most probable cause: Invalid constant field annotations.");
- }
- }
-
- private void addSchemaToSchema(T sourceSchema, T targetSchema, OptimizerNode optNode, int input) {
- try {
- for (Map.Entry<Integer, X> entry : sourceSchema) {
- Integer pos = entry.getKey();
- SemanticProperties sprops = optNode.getSemanticProperties();
-
- if (sprops != null && sprops.getForwardingTargetFields(input, pos) != null && sprops.getForwardingTargetFields(input, pos).contains(pos)) {
- targetSchema.addType(pos, entry.getValue());
- }
- }
- } catch (ConflictingFieldTypeInfoException e) {
- throw new CompilerPostPassException("Conflicting type information for field " + e.getFieldNumber()
- + " in node '" + optNode.getOperator().getName() + "' propagated from successor node. " +
- "Conflicting types: " + e.getPreviousType() + " and " + e.getNewType() +
- ". Most probable cause: Invalid constant field annotations.");
- }
- }
-
- private String getConflictingTypeErrorMessage(ConflictingFieldTypeInfoException e, String operatorName) {
- return "Conflicting type information for field " + e.getFieldNumber()
- + " in node '" + operatorName + "' between types declared in the node's "
- + "contract and types inferred from successor contracts. Conflicting types: "
- + e.getPreviousType() + " and " + e.getNewType()
- + ". Most probable cause: Invalid constant field annotations.";
- }
-
- private TypeSerializerFactory<?> createSerializer(T schema, PlanNode node) {
- try {
- return createSerializer(schema);
- } catch (MissingFieldTypeInfoException e) {
- throw new CompilerPostPassException("Missing type information while creating serializer for '" +
- node.getProgramOperator().getName() + "'.");
- }
- }
-
- // --------------------------------------------------------------------------------------------
- // Type specific methods that extract schema information
- // --------------------------------------------------------------------------------------------
-
- protected abstract T createEmptySchema();
-
- protected abstract void getSinkSchema(SinkPlanNode sink, T schema) throws CompilerPostPassException, ConflictingFieldTypeInfoException;
-
- protected abstract void getSingleInputNodeSchema(SingleInputPlanNode node, T schema) throws CompilerPostPassException, ConflictingFieldTypeInfoException;
-
- protected abstract void getDualInputNodeSchema(DualInputPlanNode node, T input1Schema, T input2Schema) throws CompilerPostPassException, ConflictingFieldTypeInfoException;
-
- // --------------------------------------------------------------------------------------------
- // Methods to create serializers and comparators
- // --------------------------------------------------------------------------------------------
-
- protected abstract TypeSerializerFactory<?> createSerializer(T schema) throws MissingFieldTypeInfoException;
-
- protected abstract TypeComparatorFactory<?> createComparator(FieldList fields, boolean[] directions, T schema) throws MissingFieldTypeInfoException;
-
- protected abstract TypePairComparatorFactory<?, ?> createPairComparator(FieldList fields1, FieldList fields2, boolean[] sortDirections,
- T schema1, T schema2) throws MissingFieldTypeInfoException;
-}
[21/21] flink git commit: [FLINK-6774][build] set missing
build-helper-maven-plugin version
Posted by ch...@apache.org.
[FLINK-6774][build] set missing build-helper-maven-plugin version
This closes #4017.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/1ae4ac28
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/1ae4ac28
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/1ae4ac28
Branch: refs/heads/master
Commit: 1ae4ac28b102c1a33d071dd1184782eb12625bbc
Parents: 5604930
Author: Nico Kruber <ni...@data-artisans.com>
Authored: Tue May 30 14:52:23 2017 +0200
Committer: zentol <ch...@apache.org>
Committed: Sun Jun 25 06:41:42 2017 +0200
----------------------------------------------------------------------
flink-connectors/flink-connector-kafka-base/pom.xml | 1 +
pom.xml | 1 +
2 files changed, 2 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/1ae4ac28/flink-connectors/flink-connector-kafka-base/pom.xml
----------------------------------------------------------------------
diff --git a/flink-connectors/flink-connector-kafka-base/pom.xml b/flink-connectors/flink-connector-kafka-base/pom.xml
index fc0045e..8ed7dfb 100644
--- a/flink-connectors/flink-connector-kafka-base/pom.xml
+++ b/flink-connectors/flink-connector-kafka-base/pom.xml
@@ -216,6 +216,7 @@ under the License.
<plugin>
<groupId>org.codehaus.mojo</groupId>
<artifactId>build-helper-maven-plugin</artifactId>
+ <version>1.7</version>
<executions>
<execution>
<phase>generate-test-sources</phase>
http://git-wip-us.apache.org/repos/asf/flink/blob/1ae4ac28/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index 5fb895d..042e407 100644
--- a/pom.xml
+++ b/pom.xml
@@ -663,6 +663,7 @@ under the License.
<plugin>
<groupId>org.codehaus.mojo</groupId>
<artifactId>build-helper-maven-plugin</artifactId>
+ <version>1.7</version>
<executions>
<execution>
<phase>generate-sources</phase>
[04/21] flink git commit: [FLINK-6798][docs] update old network
buffer notices
Posted by ch...@apache.org.
[FLINK-6798][docs] update old network buffer notices
This closes #4080.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/d713ed15
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/d713ed15
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/d713ed15
Branch: refs/heads/master
Commit: d713ed154240c1678505b22ae260bdc8b7b652d7
Parents: b2062ee
Author: Nico Kruber <ni...@data-artisans.com>
Authored: Tue Jun 6 14:24:14 2017 +0200
Committer: zentol <ch...@apache.org>
Committed: Fri Jun 23 14:14:29 2017 +0200
----------------------------------------------------------------------
docs/monitoring/large_state_tuning.md | 20 +++++---------------
.../flink/configuration/TaskManagerOptions.java | 4 ++--
.../runtime/io/network/NetworkEnvironment.java | 4 ++--
3 files changed, 9 insertions(+), 19 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/d713ed15/docs/monitoring/large_state_tuning.md
----------------------------------------------------------------------
diff --git a/docs/monitoring/large_state_tuning.md b/docs/monitoring/large_state_tuning.md
index 9e1ecc7..a356970 100644
--- a/docs/monitoring/large_state_tuning.md
+++ b/docs/monitoring/large_state_tuning.md
@@ -94,21 +94,11 @@ When a savepoint is manually triggered, it may be in process concurrently with a
## Tuning Network Buffers
-The number of network buffers is a parameter that can currently have an effect on checkpointing at large scale.
-The Flink community is working on eliminating that parameter in the next versions of Flink.
-
-The number of network buffers defines how much data a TaskManager can hold in-flight before back-pressure kicks in.
-A very high number of network buffers means that a lot of data may be in the stream network channels when a checkpoint
-is started. Because the checkpoint barriers travel with that data (see [description of how checkpointing works](../internals/stream_checkpointing.html)),
-a lot of in-flight data means that the barriers have to wait for that data to be transported/processed before arriving
-at the target operator.
-
-Having a lot of data in-flight also does not speed up the data processing as a whole. It only means that data is picked up faster
-from the data source (log, files, message queue) and buffered longer in Flink. Having fewer network buffers means that
-data is picked up from the source more immediately before it is actually being processed, which is generally desirable.
-The number of network buffers should hence not be set arbitrarily large, but to a low multiple (such as 2x) of the
-minimum number of required buffers.
-
+Before Flink 1.3, an increased number of network buffers also caused increased checkpointing times since
+keeping more in-flight data meant that checkpoint barriers got delayed. Since Flink 1.3, the
+number of network buffers used per outgoing/incoming channel is limited and thus network buffers
+may be configured without affecting checkpoint times
+(see [network buffer configuration](../setup/config.html#configuring-the-network-buffers)).
## Make state checkpointing Asynchronous where possible
http://git-wip-us.apache.org/repos/asf/flink/blob/d713ed15/flink-core/src/main/java/org/apache/flink/configuration/TaskManagerOptions.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/configuration/TaskManagerOptions.java b/flink-core/src/main/java/org/apache/flink/configuration/TaskManagerOptions.java
index 8480045..bde564a 100644
--- a/flink-core/src/main/java/org/apache/flink/configuration/TaskManagerOptions.java
+++ b/flink-core/src/main/java/org/apache/flink/configuration/TaskManagerOptions.java
@@ -137,7 +137,7 @@ public class TaskManagerOptions {
.defaultValue(1024L << 20); // 1 GB
/**
- * Number of network buffers to use for each outgoing/ingoing channel (subpartition/input channel).
+ * Number of network buffers to use for each outgoing/incoming channel (subpartition/input channel).
*
* Reasoning: 1 buffer for in-flight data in the subpartition + 1 buffer for parallel serialization
*/
@@ -146,7 +146,7 @@ public class TaskManagerOptions {
.defaultValue(2);
/**
- * Number of extra network buffers to use for each outgoing/ingoing gate (result partition/input gate).
+ * Number of extra network buffers to use for each outgoing/incoming gate (result partition/input gate).
*/
public static final ConfigOption<Integer> NETWORK_EXTRA_BUFFERS_PER_GATE =
key("taskmanager.network.memory.floating-buffers-per-gate")
http://git-wip-us.apache.org/repos/asf/flink/blob/d713ed15/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/NetworkEnvironment.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/NetworkEnvironment.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/NetworkEnvironment.java
index cc4cb77..4269af6 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/NetworkEnvironment.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/NetworkEnvironment.java
@@ -73,9 +73,9 @@ public class NetworkEnvironment {
private final int partitionRequestMaxBackoff;
- /** Number of network buffers to use for each outgoing/ingoing channel (subpartition/input channel). */
+ /** Number of network buffers to use for each outgoing/incoming channel (subpartition/input channel). */
private final int networkBuffersPerChannel;
- /** Number of extra network buffers to use for each outgoing/ingoing gate (result partition/input gate). */
+ /** Number of extra network buffers to use for each outgoing/incoming gate (result partition/input gate). */
private final int extraNetworkBuffersPerGate;
private boolean isShutdown;
[09/21] flink git commit: [hotfix] Improve readability in
SPV2#convertToOperatorStateSavepointV2
Posted by ch...@apache.org.
[hotfix] Improve readability in SPV2#convertToOperatorStateSavepointV2
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/7b97371a
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/7b97371a
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/7b97371a
Branch: refs/heads/master
Commit: 7b97371a091d1a096145b373f20e35b7de18483b
Parents: a0b7814
Author: zentol <ch...@apache.org>
Authored: Wed Jun 7 12:03:57 2017 +0200
Committer: zentol <ch...@apache.org>
Committed: Fri Jun 23 14:14:29 2017 +0200
----------------------------------------------------------------------
.../flink/runtime/checkpoint/savepoint/SavepointV2.java | 12 ++++++------
1 file changed, 6 insertions(+), 6 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/7b97371a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/savepoint/SavepointV2.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/savepoint/SavepointV2.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/savepoint/SavepointV2.java
index 6a3b57f..1b2963d 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/savepoint/SavepointV2.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/savepoint/SavepointV2.java
@@ -188,8 +188,8 @@ public class SavepointV2 implements Savepoint {
for (int chainIndex = 0; chainIndex < taskState.getChainLength(); chainIndex++) {
// task consists of multiple operators so we have to break the state apart
- for (int o = 0; o < operatorIDs.size(); o++) {
- OperatorID operatorID = operatorIDs.get(o);
+ for (int operatorIndex = 0; operatorIndex < operatorIDs.size(); operatorIndex++) {
+ OperatorID operatorID = operatorIDs.get(operatorIndex);
OperatorState operatorState = operatorStates.get(operatorID);
if (operatorState == null) {
@@ -204,15 +204,15 @@ public class SavepointV2 implements Savepoint {
KeyedStateHandle rawKeyedState = null;
// only the head operator retains the keyed state
- if (o == operatorIDs.size() - 1) {
+ if (operatorIndex == operatorIDs.size() - 1) {
managedKeyedState = subtaskState.getManagedKeyedState();
rawKeyedState = subtaskState.getRawKeyedState();
}
OperatorSubtaskState operatorSubtaskState = new OperatorSubtaskState(
- nonPartitionedState != null ? nonPartitionedState.get(o) : null,
- partitioneableState != null ? partitioneableState.get(o) : null,
- rawOperatorState != null ? rawOperatorState.get(o) : null,
+ nonPartitionedState != null ? nonPartitionedState.get(operatorIndex) : null,
+ partitioneableState != null ? partitioneableState.get(operatorIndex) : null,
+ rawOperatorState != null ? rawOperatorState.get(operatorIndex) : null,
managedKeyedState,
rawKeyedState);
[02/21] flink git commit: [hotfix][docs] update Checkpoint docs with
correct code example
Posted by ch...@apache.org.
[hotfix][docs] update Checkpoint docs with correct code example
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/071524ef
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/071524ef
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/071524ef
Branch: refs/heads/master
Commit: 071524ef837fa902343d9ea157376ac96afe91f1
Parents: af0ced9
Author: Nico Kruber <ni...@data-artisans.com>
Authored: Wed May 31 10:50:39 2017 +0200
Committer: zentol <ch...@apache.org>
Committed: Fri Jun 23 14:14:29 2017 +0200
----------------------------------------------------------------------
.../org/apache/flink/streaming/api/checkpoint/Checkpointed.java | 2 +-
.../flink/streaming/api/checkpoint/CheckpointedAsynchronously.java | 2 +-
2 files changed, 2 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/071524ef/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/checkpoint/Checkpointed.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/checkpoint/Checkpointed.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/checkpoint/Checkpointed.java
index 7dedfbf..cb3c7cc 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/checkpoint/Checkpointed.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/checkpoint/Checkpointed.java
@@ -42,7 +42,7 @@ import java.io.Serializable;
* }
*
* public void restoreState(List<Integer> state) throws Exception {
- * this.value = state.count.isEmpty() ? 0 : state.get(0);
+ * this.value = state.isEmpty() ? 0 : state.get(0);
* }
*
* public T map(T value) {
http://git-wip-us.apache.org/repos/asf/flink/blob/071524ef/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/checkpoint/CheckpointedAsynchronously.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/checkpoint/CheckpointedAsynchronously.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/checkpoint/CheckpointedAsynchronously.java
index 2230f59..5138b49 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/checkpoint/CheckpointedAsynchronously.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/checkpoint/CheckpointedAsynchronously.java
@@ -43,7 +43,7 @@ import java.io.Serializable;
* }
*
* public void restoreState(List<Integer> state) throws Exception {
- * this.value = state.count.isEmpty() ? 0 : state.get(0);
+ * this.value = state.isEmpty() ? 0 : state.get(0);
* }
*
* public T map(T value) {
[14/21] flink git commit: [FLINK-6682] [checkpoints] Improve error
message in case parallelism exceeds maxParallelism
Posted by ch...@apache.org.
[FLINK-6682] [checkpoints] Improve error message in case parallelism exceeds maxParallelism
This closes #4125.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/c736ba2e
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/c736ba2e
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/c736ba2e
Branch: refs/heads/master
Commit: c736ba2ef15e9e81a54a3fc02ccffadcbf594767
Parents: 7216407
Author: zhangminglei <zm...@163.com>
Authored: Tue Jun 20 19:43:44 2017 +0800
Committer: zentol <ch...@apache.org>
Committed: Fri Jun 23 14:14:30 2017 +0200
----------------------------------------------------------------------
.../flink/runtime/checkpoint/StateAssignmentOperation.java | 8 ++++++++
1 file changed, 8 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/c736ba2e/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/StateAssignmentOperation.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/StateAssignmentOperation.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/StateAssignmentOperation.java
index 1042d5a..5712ea1 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/StateAssignmentOperation.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/StateAssignmentOperation.java
@@ -464,6 +464,14 @@ public class StateAssignmentOperation {
private static void checkParallelismPreconditions(OperatorState operatorState, ExecutionJobVertex executionJobVertex) {
//----------------------------------------max parallelism preconditions-------------------------------------
+ if (operatorState.getMaxParallelism() < executionJobVertex.getParallelism()) {
+ throw new IllegalStateException("The state for task " + executionJobVertex.getJobVertexId() +
+ " can not be restored. The maximum parallelism (" + operatorState.getMaxParallelism() +
+ ") of the restored state is lower than the configured parallelism (" + executionJobVertex.getParallelism() +
+ "). Please reduce the parallelism of the task to be lower or equal to the maximum parallelism."
+ );
+ }
+
// check that the number of key groups have not changed or if we need to override it to satisfy the restored state
if (operatorState.getMaxParallelism() != executionJobVertex.getMaxParallelism()) {
[03/21] flink git commit: [FLINK-6784][docs] update externalized
checkpoints documentation
Posted by ch...@apache.org.
[FLINK-6784][docs] update externalized checkpoints documentation
This closes #4033.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/e1269ed8
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/e1269ed8
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/e1269ed8
Branch: refs/heads/master
Commit: e1269ed8fa2d66978e2816697d6e3fe3dfabf7f6
Parents: d713ed1
Author: Nico Kruber <ni...@data-artisans.com>
Authored: Wed May 31 16:22:22 2017 +0200
Committer: zentol <ch...@apache.org>
Committed: Fri Jun 23 14:14:29 2017 +0200
----------------------------------------------------------------------
docs/dev/stream/checkpointing.md | 21 ++++++++++++++
docs/setup/checkpoints.md | 54 +++++++++++++++++++++++++++++++----
2 files changed, 70 insertions(+), 5 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/e1269ed8/docs/dev/stream/checkpointing.md
----------------------------------------------------------------------
diff --git a/docs/dev/stream/checkpointing.md b/docs/dev/stream/checkpointing.md
index 3a0a1ae..4fe06c1 100644
--- a/docs/dev/stream/checkpointing.md
+++ b/docs/dev/stream/checkpointing.md
@@ -72,6 +72,8 @@ Other parameters for checkpointing include:
This option cannot be used when a minimum time between checkpoints is defined.
+ - *externalized checkpoints*: You can configure periodic checkpoints to be persisted externally. Externalized checkpoints write their meta data out to persistent storage and are *not* automatically cleaned up when the job fails. This way, you will have a checkpoint around to resume from if your job fails. There are more details in the [deployment notes on externalized checkpoints](../../setup/checkpoints.html#externalized-checkpoints).
+
<div class="codetabs" markdown="1">
<div data-lang="java" markdown="1">
{% highlight java %}
@@ -93,6 +95,9 @@ env.getCheckpointConfig().setCheckpointTimeout(60000);
// allow only one checkpoint to be in progress at the same time
env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
+
+// enable externalized checkpoints which are retained after job cancellation
+env.getCheckpointConfig().enableExternalizedCheckpoints(ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
{% endhighlight %}
</div>
<div data-lang="scala" markdown="1">
@@ -119,6 +124,22 @@ env.getCheckpointConfig.setMaxConcurrentCheckpoints(1)
</div>
</div>
+### Related Config Options
+
+Some more parameters and/or defaults may be set via `conf/flink-conf.yaml` (see [configuration](config.html) for a full guide):
+
+- `state.backend`: The backend that will be used to store operator state checkpoints if checkpointing is enabled. Supported backends:
+ - `jobmanager`: In-memory state, backup to JobManager's/ZooKeeper's memory. Should be used only for minimal state (Kafka offsets) or testing and local debugging.
+ - `filesystem`: State is in-memory on the TaskManagers, and state snapshots are stored in a file system. Supported are all filesystems supported by Flink, for example HDFS, S3, ...
+
+- `state.backend.fs.checkpointdir`: Directory for storing checkpoints in a Flink supported filesystem. Note: State backend must be accessible from the JobManager, use `file://` only for local setups.
+
+- `state.backend.rocksdb.checkpointdir`: The local directory for storing RocksDB files, or a list of directories separated by the systems directory delimiter (for example ‘:’ (colon) on Linux/Unix). (DEFAULT value is `taskmanager.tmp.dirs`)
+
+- `state.checkpoints.dir`: The target directory for meta data of [externalized checkpoints](../../setup/checkpoints.html#externalized-checkpoints).
+
+- `state.checkpoints.num-retained`: The number of completed checkpoint instances to retain. Having more than one allows recovery fallback to an earlier checkpoints if the latest checkpoint is corrupt. (Default: 1)
+
{% top %}
http://git-wip-us.apache.org/repos/asf/flink/blob/e1269ed8/docs/setup/checkpoints.md
----------------------------------------------------------------------
diff --git a/docs/setup/checkpoints.md b/docs/setup/checkpoints.md
index 65d14b7..0070cb1 100644
--- a/docs/setup/checkpoints.md
+++ b/docs/setup/checkpoints.md
@@ -28,12 +28,22 @@ under the License.
## Overview
-TBD
+Checkpoints make state in Flink fault tolerant by allowing state and the
+corresponding stream positions to be recovered, thereby giving the application
+the same semantics as a failure-free execution.
+See [Checkpointing](../dev/stream/checkpointing.html) for how to enable and
+configure checkpoints for your program.
## Externalized Checkpoints
-You can configure periodic checkpoints to be persisted externally. Externalized checkpoints write their meta data out to persistent storage and are *not* automatically cleaned up when the job fails. This way, you will have a checkpoint around to resume from if your job fails.
+Checkpoints are by default not persisted externally and are only used to
+resume a job from failures. They are deleted when a program is cancelled.
+You can, however, configure periodic checkpoints to be persisted externally
+similarly to [savepoints](savepoints.html). These *externalized checkpoints*
+write their meta data out to persistent storage and are *not* automatically
+cleaned up when the job fails. This way, you will have a checkpoint around
+to resume from if your job fails.
```java
CheckpointConfig config = env.getCheckpointConfig();
@@ -46,12 +56,46 @@ The `ExternalizedCheckpointCleanup` mode configures what happens with externaliz
- **`ExternalizedCheckpointCleanup.DELETE_ON_CANCELLATION`**: Delete the externalized checkpoint when the job is cancelled. The checkpoint state will only be available if the job fails.
-The **target directory** for the checkpoint is determined from the default checkpoint directory configuration. This is configured via the configuration key `state.checkpoints.dir`, which should point to the desired target directory:
+### Directory Structure
+
+Similarly to [savepoints](savepoints.html), an externalized checkpoint consists
+of a meta data file and, depending on the state back-end, some additional data
+files. The **target directory** for the externalized checkpoint's meta data is
+determined from the configuration key `state.checkpoints.dir` which, currently,
+can only be set via the configuration files.
```
state.checkpoints.dir: hdfs:///checkpoints/
```
-This directory will then contain the checkpoint meta data required to restore the checkpoint. The actual checkpoint files will still be available in their configured directory. You currently can only set this via the configuration files.
+This directory will then contain the checkpoint meta data required to restore
+the checkpoint. For the `MemoryStateBackend`, this meta data file will be
+self-contained and no further files are needed.
+
+`FsStateBackend` and `RocksDBStateBackend` write separate data files
+and only write the paths to these files into the meta data file. These data
+files are stored at the path given to the state back-end during construction.
+
+```java
+env.setStateBackend(new RocksDBStateBackend("hdfs:///checkpoints-data/");
+```
+
+### Difference to Savepoints
+
+Externalized checkpoints have a few differences from [savepoints](savepoints.html). They
+- use a state backend specific (low-level) data format,
+- may be incremental,
+- do not support Flink specific features like rescaling.
+
+### Resuming from an externalized checkpoint
-Follow the [savepoint guide]({{ site.baseurl }}/setup/cli.html#savepoints) when you want to resume from a specific checkpoint.
+A job may be resumed from an externalized checkpoint just as from a savepoint
+by using the checkpoint's meta data file instead (see the
+[savepoint restore guide](cli.html#restore-a-savepoint)). Note that if the
+meta data file is not self-contained, the jobmanager needs to have access to
+the data files it refers to (see [Directory Structure](#directory-structure)
+above).
+
+```sh
+$ bin/flink run -s :checkpointMetaDataPath [:runArgs]
+```
[18/21] flink git commit: [FLINK-6956] Make table example POJO
classes public
Posted by ch...@apache.org.
[FLINK-6956] Make table example POJO classes public
This closes #4148.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/1f6be837
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/1f6be837
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/1f6be837
Branch: refs/heads/master
Commit: 1f6be8372706c1531844857291bc166bf091176b
Parents: 9f77073
Author: zentol <ch...@apache.org>
Authored: Tue Jun 20 15:42:13 2017 +0200
Committer: zentol <ch...@apache.org>
Committed: Sat Jun 24 22:09:05 2017 +0200
----------------------------------------------------------------------
.../java/org/apache/flink/table/examples/java/WordCountSQL.java | 5 ++++-
.../org/apache/flink/table/examples/java/WordCountTable.java | 5 ++++-
2 files changed, 8 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/1f6be837/flink-examples/flink-examples-table/src/main/java/org/apache/flink/table/examples/java/WordCountSQL.java
----------------------------------------------------------------------
diff --git a/flink-examples/flink-examples-table/src/main/java/org/apache/flink/table/examples/java/WordCountSQL.java b/flink-examples/flink-examples-table/src/main/java/org/apache/flink/table/examples/java/WordCountSQL.java
index c4a36f0..65efc17 100644
--- a/flink-examples/flink-examples-table/src/main/java/org/apache/flink/table/examples/java/WordCountSQL.java
+++ b/flink-examples/flink-examples-table/src/main/java/org/apache/flink/table/examples/java/WordCountSQL.java
@@ -65,7 +65,10 @@ public class WordCountSQL {
// USER DATA TYPES
// *************************************************************************
- private static class WC {
+ /**
+ * Simple POJO containing a word and its respective count.
+ */
+ public static class WC {
public String word;
public long frequency;
http://git-wip-us.apache.org/repos/asf/flink/blob/1f6be837/flink-examples/flink-examples-table/src/main/java/org/apache/flink/table/examples/java/WordCountTable.java
----------------------------------------------------------------------
diff --git a/flink-examples/flink-examples-table/src/main/java/org/apache/flink/table/examples/java/WordCountTable.java b/flink-examples/flink-examples-table/src/main/java/org/apache/flink/table/examples/java/WordCountTable.java
index 8606cd8..5fb1beb 100644
--- a/flink-examples/flink-examples-table/src/main/java/org/apache/flink/table/examples/java/WordCountTable.java
+++ b/flink-examples/flink-examples-table/src/main/java/org/apache/flink/table/examples/java/WordCountTable.java
@@ -62,7 +62,10 @@ public class WordCountTable {
// USER DATA TYPES
// *************************************************************************
- private static class WC {
+ /**
+ * Simple POJO containing a word and its respective count.
+ */
+ public static class WC {
public String word;
public long frequency;
[17/21] flink git commit: [FLINK-6967] Remove batch-examples
dependency from storm examples
Posted by ch...@apache.org.
[FLINK-6967] Remove batch-examples dependency from storm examples
This closes #4159.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/9f77073a
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/9f77073a
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/9f77073a
Branch: refs/heads/master
Commit: 9f77073a56cccf06399dae24de1c38dc27dafcb6
Parents: 2274bf7
Author: zentol <ch...@apache.org>
Authored: Wed Jun 21 17:06:00 2017 +0200
Committer: zentol <ch...@apache.org>
Committed: Sat Jun 24 22:09:05 2017 +0200
----------------------------------------------------------------------
flink-contrib/flink-storm-examples/pom.xml | 6 --
.../storm/exclamation/ExclamationTopology.java | 2 +-
.../storm/exclamation/ExclamationWithBolt.java | 2 +-
.../storm/exclamation/ExclamationWithSpout.java | 2 +-
.../storm/wordcount/BoltTokenizerWordCount.java | 2 +-
.../wordcount/BoltTokenizerWordCountPojo.java | 2 +-
.../BoltTokenizerWordCountWithNames.java | 2 +-
.../storm/wordcount/SpoutSourceWordCount.java | 2 +-
.../flink/storm/wordcount/WordCountLocal.java | 2 +-
.../storm/wordcount/WordCountLocalByName.java | 2 +-
.../wordcount/WordCountRemoteByClient.java | 2 +-
.../wordcount/WordCountRemoteBySubmitter.java | 2 +-
.../storm/wordcount/WordCountTopology.java | 2 +-
.../wordcount/operators/WordCountDataPojos.java | 2 +-
.../wordcount/operators/WordCountDataTuple.java | 2 +-
.../operators/WordCountInMemorySpout.java | 2 +-
.../storm/wordcount/util/WordCountData.java | 64 ++++++++++++++++++++
17 files changed, 79 insertions(+), 21 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/9f77073a/flink-contrib/flink-storm-examples/pom.xml
----------------------------------------------------------------------
diff --git a/flink-contrib/flink-storm-examples/pom.xml b/flink-contrib/flink-storm-examples/pom.xml
index 0296ff3..85d21bf 100644
--- a/flink-contrib/flink-storm-examples/pom.xml
+++ b/flink-contrib/flink-storm-examples/pom.xml
@@ -67,12 +67,6 @@ under the License.
</dependency>
<dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-examples-batch_${scala.binary.version}</artifactId>
- <version>${project.version}</version>
- </dependency>
-
- <dependency>
<groupId>org.apache.storm</groupId>
<artifactId>storm-starter</artifactId>
<version>1.0.0</version>
http://git-wip-us.apache.org/repos/asf/flink/blob/9f77073a/flink-contrib/flink-storm-examples/src/main/java/org/apache/flink/storm/exclamation/ExclamationTopology.java
----------------------------------------------------------------------
diff --git a/flink-contrib/flink-storm-examples/src/main/java/org/apache/flink/storm/exclamation/ExclamationTopology.java b/flink-contrib/flink-storm-examples/src/main/java/org/apache/flink/storm/exclamation/ExclamationTopology.java
index 51edd1f..9de1684 100644
--- a/flink-contrib/flink-storm-examples/src/main/java/org/apache/flink/storm/exclamation/ExclamationTopology.java
+++ b/flink-contrib/flink-storm-examples/src/main/java/org/apache/flink/storm/exclamation/ExclamationTopology.java
@@ -17,7 +17,6 @@
package org.apache.flink.storm.exclamation;
-import org.apache.flink.examples.java.wordcount.util.WordCountData;
import org.apache.flink.storm.exclamation.operators.ExclamationBolt;
import org.apache.flink.storm.util.BoltFileSink;
import org.apache.flink.storm.util.BoltPrintSink;
@@ -25,6 +24,7 @@ import org.apache.flink.storm.util.FiniteFileSpout;
import org.apache.flink.storm.util.FiniteInMemorySpout;
import org.apache.flink.storm.util.OutputFormatter;
import org.apache.flink.storm.util.SimpleOutputFormatter;
+import org.apache.flink.storm.wordcount.util.WordCountData;
import org.apache.storm.topology.TopologyBuilder;
http://git-wip-us.apache.org/repos/asf/flink/blob/9f77073a/flink-contrib/flink-storm-examples/src/main/java/org/apache/flink/storm/exclamation/ExclamationWithBolt.java
----------------------------------------------------------------------
diff --git a/flink-contrib/flink-storm-examples/src/main/java/org/apache/flink/storm/exclamation/ExclamationWithBolt.java b/flink-contrib/flink-storm-examples/src/main/java/org/apache/flink/storm/exclamation/ExclamationWithBolt.java
index a838e69..c31c36a 100644
--- a/flink-contrib/flink-storm-examples/src/main/java/org/apache/flink/storm/exclamation/ExclamationWithBolt.java
+++ b/flink-contrib/flink-storm-examples/src/main/java/org/apache/flink/storm/exclamation/ExclamationWithBolt.java
@@ -20,9 +20,9 @@ package org.apache.flink.storm.exclamation;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.typeutils.TypeExtractor;
-import org.apache.flink.examples.java.wordcount.util.WordCountData;
import org.apache.flink.storm.exclamation.operators.ExclamationBolt;
import org.apache.flink.storm.util.StormConfig;
+import org.apache.flink.storm.wordcount.util.WordCountData;
import org.apache.flink.storm.wrappers.BoltWrapper;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
http://git-wip-us.apache.org/repos/asf/flink/blob/9f77073a/flink-contrib/flink-storm-examples/src/main/java/org/apache/flink/storm/exclamation/ExclamationWithSpout.java
----------------------------------------------------------------------
diff --git a/flink-contrib/flink-storm-examples/src/main/java/org/apache/flink/storm/exclamation/ExclamationWithSpout.java b/flink-contrib/flink-storm-examples/src/main/java/org/apache/flink/storm/exclamation/ExclamationWithSpout.java
index b165f00..c86ab69 100644
--- a/flink-contrib/flink-storm-examples/src/main/java/org/apache/flink/storm/exclamation/ExclamationWithSpout.java
+++ b/flink-contrib/flink-storm-examples/src/main/java/org/apache/flink/storm/exclamation/ExclamationWithSpout.java
@@ -20,10 +20,10 @@ package org.apache.flink.storm.exclamation;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.typeutils.TypeExtractor;
-import org.apache.flink.examples.java.wordcount.util.WordCountData;
import org.apache.flink.storm.util.FiniteFileSpout;
import org.apache.flink.storm.util.FiniteInMemorySpout;
import org.apache.flink.storm.util.StormConfig;
+import org.apache.flink.storm.wordcount.util.WordCountData;
import org.apache.flink.storm.wrappers.SpoutWrapper;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
http://git-wip-us.apache.org/repos/asf/flink/blob/9f77073a/flink-contrib/flink-storm-examples/src/main/java/org/apache/flink/storm/wordcount/BoltTokenizerWordCount.java
----------------------------------------------------------------------
diff --git a/flink-contrib/flink-storm-examples/src/main/java/org/apache/flink/storm/wordcount/BoltTokenizerWordCount.java b/flink-contrib/flink-storm-examples/src/main/java/org/apache/flink/storm/wordcount/BoltTokenizerWordCount.java
index 6f7addf..54a6080 100644
--- a/flink-contrib/flink-storm-examples/src/main/java/org/apache/flink/storm/wordcount/BoltTokenizerWordCount.java
+++ b/flink-contrib/flink-storm-examples/src/main/java/org/apache/flink/storm/wordcount/BoltTokenizerWordCount.java
@@ -19,8 +19,8 @@ package org.apache.flink.storm.wordcount;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.typeutils.TypeExtractor;
-import org.apache.flink.examples.java.wordcount.util.WordCountData;
import org.apache.flink.storm.wordcount.operators.BoltTokenizer;
+import org.apache.flink.storm.wordcount.util.WordCountData;
import org.apache.flink.storm.wrappers.BoltWrapper;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
http://git-wip-us.apache.org/repos/asf/flink/blob/9f77073a/flink-contrib/flink-storm-examples/src/main/java/org/apache/flink/storm/wordcount/BoltTokenizerWordCountPojo.java
----------------------------------------------------------------------
diff --git a/flink-contrib/flink-storm-examples/src/main/java/org/apache/flink/storm/wordcount/BoltTokenizerWordCountPojo.java b/flink-contrib/flink-storm-examples/src/main/java/org/apache/flink/storm/wordcount/BoltTokenizerWordCountPojo.java
index 125a044..06c6a03 100644
--- a/flink-contrib/flink-storm-examples/src/main/java/org/apache/flink/storm/wordcount/BoltTokenizerWordCountPojo.java
+++ b/flink-contrib/flink-storm-examples/src/main/java/org/apache/flink/storm/wordcount/BoltTokenizerWordCountPojo.java
@@ -23,10 +23,10 @@ import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.typeutils.PojoTypeInfo;
import org.apache.flink.api.java.typeutils.TypeExtractor;
import org.apache.flink.core.fs.Path;
-import org.apache.flink.examples.java.wordcount.util.WordCountData;
import org.apache.flink.storm.wordcount.operators.BoltTokenizerByName;
import org.apache.flink.storm.wordcount.operators.WordCountDataPojos;
import org.apache.flink.storm.wordcount.operators.WordCountDataPojos.Sentence;
+import org.apache.flink.storm.wordcount.util.WordCountData;
import org.apache.flink.storm.wrappers.BoltWrapper;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
http://git-wip-us.apache.org/repos/asf/flink/blob/9f77073a/flink-contrib/flink-storm-examples/src/main/java/org/apache/flink/storm/wordcount/BoltTokenizerWordCountWithNames.java
----------------------------------------------------------------------
diff --git a/flink-contrib/flink-storm-examples/src/main/java/org/apache/flink/storm/wordcount/BoltTokenizerWordCountWithNames.java b/flink-contrib/flink-storm-examples/src/main/java/org/apache/flink/storm/wordcount/BoltTokenizerWordCountWithNames.java
index f469bab..b0bb85a 100644
--- a/flink-contrib/flink-storm-examples/src/main/java/org/apache/flink/storm/wordcount/BoltTokenizerWordCountWithNames.java
+++ b/flink-contrib/flink-storm-examples/src/main/java/org/apache/flink/storm/wordcount/BoltTokenizerWordCountWithNames.java
@@ -25,9 +25,9 @@ import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.typeutils.TupleTypeInfo;
import org.apache.flink.api.java.typeutils.TypeExtractor;
import org.apache.flink.core.fs.Path;
-import org.apache.flink.examples.java.wordcount.util.WordCountData;
import org.apache.flink.storm.wordcount.operators.BoltTokenizerByName;
import org.apache.flink.storm.wordcount.operators.WordCountDataTuple;
+import org.apache.flink.storm.wordcount.util.WordCountData;
import org.apache.flink.storm.wrappers.BoltWrapper;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
http://git-wip-us.apache.org/repos/asf/flink/blob/9f77073a/flink-contrib/flink-storm-examples/src/main/java/org/apache/flink/storm/wordcount/SpoutSourceWordCount.java
----------------------------------------------------------------------
diff --git a/flink-contrib/flink-storm-examples/src/main/java/org/apache/flink/storm/wordcount/SpoutSourceWordCount.java b/flink-contrib/flink-storm-examples/src/main/java/org/apache/flink/storm/wordcount/SpoutSourceWordCount.java
index f0cfd7a..7109c4a 100644
--- a/flink-contrib/flink-storm-examples/src/main/java/org/apache/flink/storm/wordcount/SpoutSourceWordCount.java
+++ b/flink-contrib/flink-storm-examples/src/main/java/org/apache/flink/storm/wordcount/SpoutSourceWordCount.java
@@ -20,9 +20,9 @@ package org.apache.flink.storm.wordcount;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.typeutils.TypeExtractor;
-import org.apache.flink.examples.java.wordcount.util.WordCountData;
import org.apache.flink.storm.wordcount.operators.WordCountFileSpout;
import org.apache.flink.storm.wordcount.operators.WordCountInMemorySpout;
+import org.apache.flink.storm.wordcount.util.WordCountData;
import org.apache.flink.storm.wrappers.SpoutWrapper;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
http://git-wip-us.apache.org/repos/asf/flink/blob/9f77073a/flink-contrib/flink-storm-examples/src/main/java/org/apache/flink/storm/wordcount/WordCountLocal.java
----------------------------------------------------------------------
diff --git a/flink-contrib/flink-storm-examples/src/main/java/org/apache/flink/storm/wordcount/WordCountLocal.java b/flink-contrib/flink-storm-examples/src/main/java/org/apache/flink/storm/wordcount/WordCountLocal.java
index 82c8ae3..7198327 100644
--- a/flink-contrib/flink-storm-examples/src/main/java/org/apache/flink/storm/wordcount/WordCountLocal.java
+++ b/flink-contrib/flink-storm-examples/src/main/java/org/apache/flink/storm/wordcount/WordCountLocal.java
@@ -17,9 +17,9 @@
package org.apache.flink.storm.wordcount;
-import org.apache.flink.examples.java.wordcount.util.WordCountData;
import org.apache.flink.storm.api.FlinkLocalCluster;
import org.apache.flink.storm.api.FlinkTopology;
+import org.apache.flink.storm.wordcount.util.WordCountData;
import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
http://git-wip-us.apache.org/repos/asf/flink/blob/9f77073a/flink-contrib/flink-storm-examples/src/main/java/org/apache/flink/storm/wordcount/WordCountLocalByName.java
----------------------------------------------------------------------
diff --git a/flink-contrib/flink-storm-examples/src/main/java/org/apache/flink/storm/wordcount/WordCountLocalByName.java b/flink-contrib/flink-storm-examples/src/main/java/org/apache/flink/storm/wordcount/WordCountLocalByName.java
index b960b79..dcad567 100644
--- a/flink-contrib/flink-storm-examples/src/main/java/org/apache/flink/storm/wordcount/WordCountLocalByName.java
+++ b/flink-contrib/flink-storm-examples/src/main/java/org/apache/flink/storm/wordcount/WordCountLocalByName.java
@@ -17,9 +17,9 @@
package org.apache.flink.storm.wordcount;
-import org.apache.flink.examples.java.wordcount.util.WordCountData;
import org.apache.flink.storm.api.FlinkLocalCluster;
import org.apache.flink.storm.api.FlinkTopology;
+import org.apache.flink.storm.wordcount.util.WordCountData;
import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
http://git-wip-us.apache.org/repos/asf/flink/blob/9f77073a/flink-contrib/flink-storm-examples/src/main/java/org/apache/flink/storm/wordcount/WordCountRemoteByClient.java
----------------------------------------------------------------------
diff --git a/flink-contrib/flink-storm-examples/src/main/java/org/apache/flink/storm/wordcount/WordCountRemoteByClient.java b/flink-contrib/flink-storm-examples/src/main/java/org/apache/flink/storm/wordcount/WordCountRemoteByClient.java
index 8dff6d7..9ad0d63 100644
--- a/flink-contrib/flink-storm-examples/src/main/java/org/apache/flink/storm/wordcount/WordCountRemoteByClient.java
+++ b/flink-contrib/flink-storm-examples/src/main/java/org/apache/flink/storm/wordcount/WordCountRemoteByClient.java
@@ -17,9 +17,9 @@
package org.apache.flink.storm.wordcount;
-import org.apache.flink.examples.java.wordcount.util.WordCountData;
import org.apache.flink.storm.api.FlinkClient;
import org.apache.flink.storm.api.FlinkTopology;
+import org.apache.flink.storm.wordcount.util.WordCountData;
import org.apache.storm.Config;
import org.apache.storm.generated.AlreadyAliveException;
http://git-wip-us.apache.org/repos/asf/flink/blob/9f77073a/flink-contrib/flink-storm-examples/src/main/java/org/apache/flink/storm/wordcount/WordCountRemoteBySubmitter.java
----------------------------------------------------------------------
diff --git a/flink-contrib/flink-storm-examples/src/main/java/org/apache/flink/storm/wordcount/WordCountRemoteBySubmitter.java b/flink-contrib/flink-storm-examples/src/main/java/org/apache/flink/storm/wordcount/WordCountRemoteBySubmitter.java
index 745ec85..2d1a0b5 100644
--- a/flink-contrib/flink-storm-examples/src/main/java/org/apache/flink/storm/wordcount/WordCountRemoteBySubmitter.java
+++ b/flink-contrib/flink-storm-examples/src/main/java/org/apache/flink/storm/wordcount/WordCountRemoteBySubmitter.java
@@ -17,10 +17,10 @@
package org.apache.flink.storm.wordcount;
-import org.apache.flink.examples.java.wordcount.util.WordCountData;
import org.apache.flink.storm.api.FlinkClient;
import org.apache.flink.storm.api.FlinkSubmitter;
import org.apache.flink.storm.api.FlinkTopology;
+import org.apache.flink.storm.wordcount.util.WordCountData;
import org.apache.storm.Config;
import org.apache.storm.StormSubmitter;
http://git-wip-us.apache.org/repos/asf/flink/blob/9f77073a/flink-contrib/flink-storm-examples/src/main/java/org/apache/flink/storm/wordcount/WordCountTopology.java
----------------------------------------------------------------------
diff --git a/flink-contrib/flink-storm-examples/src/main/java/org/apache/flink/storm/wordcount/WordCountTopology.java b/flink-contrib/flink-storm-examples/src/main/java/org/apache/flink/storm/wordcount/WordCountTopology.java
index 8627145..3ae7c5c 100644
--- a/flink-contrib/flink-storm-examples/src/main/java/org/apache/flink/storm/wordcount/WordCountTopology.java
+++ b/flink-contrib/flink-storm-examples/src/main/java/org/apache/flink/storm/wordcount/WordCountTopology.java
@@ -17,7 +17,6 @@
package org.apache.flink.storm.wordcount;
-import org.apache.flink.examples.java.wordcount.util.WordCountData;
import org.apache.flink.storm.util.BoltFileSink;
import org.apache.flink.storm.util.BoltPrintSink;
import org.apache.flink.storm.util.NullTerminatingSpout;
@@ -29,6 +28,7 @@ import org.apache.flink.storm.wordcount.operators.BoltTokenizer;
import org.apache.flink.storm.wordcount.operators.BoltTokenizerByName;
import org.apache.flink.storm.wordcount.operators.WordCountFileSpout;
import org.apache.flink.storm.wordcount.operators.WordCountInMemorySpout;
+import org.apache.flink.storm.wordcount.util.WordCountData;
import org.apache.storm.generated.StormTopology;
import org.apache.storm.topology.TopologyBuilder;
http://git-wip-us.apache.org/repos/asf/flink/blob/9f77073a/flink-contrib/flink-storm-examples/src/main/java/org/apache/flink/storm/wordcount/operators/WordCountDataPojos.java
----------------------------------------------------------------------
diff --git a/flink-contrib/flink-storm-examples/src/main/java/org/apache/flink/storm/wordcount/operators/WordCountDataPojos.java b/flink-contrib/flink-storm-examples/src/main/java/org/apache/flink/storm/wordcount/operators/WordCountDataPojos.java
index d63974b..7273eca 100644
--- a/flink-contrib/flink-storm-examples/src/main/java/org/apache/flink/storm/wordcount/operators/WordCountDataPojos.java
+++ b/flink-contrib/flink-storm-examples/src/main/java/org/apache/flink/storm/wordcount/operators/WordCountDataPojos.java
@@ -17,7 +17,7 @@
package org.apache.flink.storm.wordcount.operators;
-import org.apache.flink.examples.java.wordcount.util.WordCountData;
+import org.apache.flink.storm.wordcount.util.WordCountData;
import java.io.Serializable;
http://git-wip-us.apache.org/repos/asf/flink/blob/9f77073a/flink-contrib/flink-storm-examples/src/main/java/org/apache/flink/storm/wordcount/operators/WordCountDataTuple.java
----------------------------------------------------------------------
diff --git a/flink-contrib/flink-storm-examples/src/main/java/org/apache/flink/storm/wordcount/operators/WordCountDataTuple.java b/flink-contrib/flink-storm-examples/src/main/java/org/apache/flink/storm/wordcount/operators/WordCountDataTuple.java
index d01d9a2..6adfdb4 100644
--- a/flink-contrib/flink-storm-examples/src/main/java/org/apache/flink/storm/wordcount/operators/WordCountDataTuple.java
+++ b/flink-contrib/flink-storm-examples/src/main/java/org/apache/flink/storm/wordcount/operators/WordCountDataTuple.java
@@ -18,7 +18,7 @@
package org.apache.flink.storm.wordcount.operators;
import org.apache.flink.api.java.tuple.Tuple1;
-import org.apache.flink.examples.java.wordcount.util.WordCountData;
+import org.apache.flink.storm.wordcount.util.WordCountData;
/**
* Input tuples for WordCount programs.
http://git-wip-us.apache.org/repos/asf/flink/blob/9f77073a/flink-contrib/flink-storm-examples/src/main/java/org/apache/flink/storm/wordcount/operators/WordCountInMemorySpout.java
----------------------------------------------------------------------
diff --git a/flink-contrib/flink-storm-examples/src/main/java/org/apache/flink/storm/wordcount/operators/WordCountInMemorySpout.java b/flink-contrib/flink-storm-examples/src/main/java/org/apache/flink/storm/wordcount/operators/WordCountInMemorySpout.java
index be376a9..6bb8ac1 100644
--- a/flink-contrib/flink-storm-examples/src/main/java/org/apache/flink/storm/wordcount/operators/WordCountInMemorySpout.java
+++ b/flink-contrib/flink-storm-examples/src/main/java/org/apache/flink/storm/wordcount/operators/WordCountInMemorySpout.java
@@ -17,8 +17,8 @@
package org.apache.flink.storm.wordcount.operators;
-import org.apache.flink.examples.java.wordcount.util.WordCountData;
import org.apache.flink.storm.util.FiniteInMemorySpout;
+import org.apache.flink.storm.wordcount.util.WordCountData;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.tuple.Fields;
http://git-wip-us.apache.org/repos/asf/flink/blob/9f77073a/flink-contrib/flink-storm-examples/src/main/java/org/apache/flink/storm/wordcount/util/WordCountData.java
----------------------------------------------------------------------
diff --git a/flink-contrib/flink-storm-examples/src/main/java/org/apache/flink/storm/wordcount/util/WordCountData.java b/flink-contrib/flink-storm-examples/src/main/java/org/apache/flink/storm/wordcount/util/WordCountData.java
new file mode 100644
index 0000000..9682072
--- /dev/null
+++ b/flink-contrib/flink-storm-examples/src/main/java/org/apache/flink/storm/wordcount/util/WordCountData.java
@@ -0,0 +1,64 @@
+/*
+ * 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.storm.wordcount.util;
+
+/**
+ * Provides the default data sets used for the WordCount example program.
+ * The default data sets are used, if no parameters are given to the program.
+ */
+public class WordCountData {
+
+ public static final String[] WORDS = new String[]{
+ "To be, or not to be,--that is the question:--",
+ "Whether 'tis nobler in the mind to suffer",
+ "The slings and arrows of outrageous fortune",
+ "Or to take arms against a sea of troubles,",
+ "And by opposing end them?--To die,--to sleep,--",
+ "No more; and by a sleep to say we end",
+ "The heartache, and the thousand natural shocks",
+ "That flesh is heir to,--'tis a consummation",
+ "Devoutly to be wish'd. To die,--to sleep;--",
+ "To sleep! perchance to dream:--ay, there's the rub;",
+ "For in that sleep of death what dreams may come,",
+ "When we have shuffled off this mortal coil,",
+ "Must give us pause: there's the respect",
+ "That makes calamity of so long life;",
+ "For who would bear the whips and scorns of time,",
+ "The oppressor's wrong, the proud man's contumely,",
+ "The pangs of despis'd love, the law's delay,",
+ "The insolence of office, and the spurns",
+ "That patient merit of the unworthy takes,",
+ "When he himself might his quietus make",
+ "With a bare bodkin? who would these fardels bear,",
+ "To grunt and sweat under a weary life,",
+ "But that the dread of something after death,--",
+ "The undiscover'd country, from whose bourn",
+ "No traveller returns,--puzzles the will,",
+ "And makes us rather bear those ills we have",
+ "Than fly to others that we know not of?",
+ "Thus conscience does make cowards of us all;",
+ "And thus the native hue of resolution",
+ "Is sicklied o'er with the pale cast of thought;",
+ "And enterprises of great pith and moment,",
+ "With this regard, their currents turn awry,",
+ "And lose the name of action.--Soft you now!",
+ "The fair Ophelia!--Nymph, in thy orisons",
+ "Be all my sins remember'd."
+ };
+}
[08/21] flink git commit: [FLINK-6742] Improve savepoint migration
failure error message
Posted by ch...@apache.org.
[FLINK-6742] Improve savepoint migration failure error message
This closes #4083.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/72b0ae06
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/72b0ae06
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/72b0ae06
Branch: refs/heads/master
Commit: 72b0ae069f8404a2f8a952e1a20004b9d340c445
Parents: 7b97371
Author: zentol <ch...@apache.org>
Authored: Wed Jun 7 12:03:21 2017 +0200
Committer: zentol <ch...@apache.org>
Committed: Fri Jun 23 14:14:29 2017 +0200
----------------------------------------------------------------------
.../checkpoint/savepoint/SavepointV2.java | 19 ++++++++++++++++++-
1 file changed, 18 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/72b0ae06/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/savepoint/SavepointV2.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/savepoint/SavepointV2.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/savepoint/SavepointV2.java
index 1b2963d..5e46f93 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/savepoint/SavepointV2.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/savepoint/SavepointV2.java
@@ -168,10 +168,27 @@ public class SavepointV2 implements Savepoint {
expandedToLegacyIds = true;
}
+ if (jobVertex == null) {
+ throw new IllegalStateException(
+ "Could not find task for state with ID " + taskState.getJobVertexID() + ". " +
+ "When migrating a savepoint from a version < 1.3 please make sure that the topology was not " +
+ "changed through removal of a stateful operator or modification of a chain containing a stateful " +
+ "operator.");
+ }
+
List<OperatorID> operatorIDs = jobVertex.getOperatorIDs();
for (int subtaskIndex = 0; subtaskIndex < jobVertex.getParallelism(); subtaskIndex++) {
- SubtaskState subtaskState = taskState.getState(subtaskIndex);
+ SubtaskState subtaskState;
+ try {
+ subtaskState = taskState.getState(subtaskIndex);
+ } catch (Exception e) {
+ throw new IllegalStateException(
+ "Could not find subtask with index " + subtaskIndex + " for task " + jobVertex.getJobVertexId() + ". " +
+ "When migrating a savepoint from a version < 1.3 please make sure that no changes were made " +
+ "to the parallelism of stateful operators.",
+ e);
+ }
if (subtaskState == null) {
continue;
[07/21] flink git commit: [FLINK-6782][docs] update snapshot
documentation to reflect flink 1.3
Posted by ch...@apache.org.
[FLINK-6782][docs] update snapshot documentation to reflect flink 1.3
This closes #4024.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/60222a24
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/60222a24
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/60222a24
Branch: refs/heads/master
Commit: 60222a24123379bbbd044a6ca25817e64a9348b9
Parents: 071524e
Author: Nico Kruber <ni...@data-artisans.com>
Authored: Wed May 31 13:51:12 2017 +0200
Committer: zentol <ch...@apache.org>
Committed: Fri Jun 23 14:14:29 2017 +0200
----------------------------------------------------------------------
docs/setup/savepoints.md | 43 +++++++++++++++++++++++++++----------------
1 file changed, 27 insertions(+), 16 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/60222a24/docs/setup/savepoints.md
----------------------------------------------------------------------
diff --git a/docs/setup/savepoints.md b/docs/setup/savepoints.md
index eada9b4..2849423 100644
--- a/docs/setup/savepoints.md
+++ b/docs/setup/savepoints.md
@@ -27,12 +27,15 @@ under the License.
## Overview
-Savepoints are externally stored checkpoints that you can use to stop-and-resume or update your Flink programs. They use Flink's [checkpointing mechanism]({{ site.baseurl }}/internals/stream_checkpointing.html) to create a snapshot of the state of your streaming program and write the checkpoint meta data out to an external file system.
-
-This page covers all steps involved in triggering, restoring, and disposing savepoints. In order to allow upgrades between programs and Flink versions, it is important to check out the section about [assigning IDs to your operators](#assigning-operator-ids).
+Savepoints are externally stored self-contained checkpoints that you can use to stop-and-resume or update your Flink programs. They use Flink's [checkpointing mechanism]({{ site.baseurl }}/internals/stream_checkpointing.html) to create a (non-incremental) snapshot of the state of your streaming program and write the checkpoint data and meta data out to an external file system.
+This page covers all steps involved in triggering, restoring, and disposing savepoints.
For more details on how Flink handles state and failures in general, check out the [State in Streaming Programs]({{ site.baseurl }}/dev/stream/state.html) page.
+<div class="alert alert-warning">
+<strong>Attention:</strong> In order to allow upgrades between programs and Flink versions, it is important to check out the following section about <a href="#assigning-operator-ids">assigning IDs to your operators</a>.
+</div>
+
## Assigning Operator IDs
It is **highly recommended** that you adjust your programs as described in this section in order to be able to upgrade your programs in the future. The main required change is to manually specify operator IDs via the **`uid(String)`** method. These IDs are used to scope the state of each operator.
@@ -73,17 +76,29 @@ With Flink >= 1.2.0 it is also possible to *resume from savepoints* using the we
### Triggering Savepoints
-When triggering a savepoint, a single savepoint file will be created that contains the checkpoint *meta data*. The actual checkpoint state will be kept around in the configured checkpoint directory. For example with a `FsStateBackend` or `RocksDBStateBackend`:
+When triggering a savepoint, a new savepoint directory beneath the target directory is created. In there, the data as well as the meta data will be stored. For example with a `FsStateBackend` or `RocksDBStateBackend`:
```sh
+# Savepoint target directory
+/savepoints/
+
+# Savepoint directory
+/savepoints/savepoint-:shortjobid-:savepointid/
+
# Savepoint file contains the checkpoint meta data
-/savepoints/savepoint-123123
+/savepoints/savepoint-:shortjobid-:savepointid/_metadata
-# Checkpoint directory contains the actual state
-/checkpoints/:jobid/chk-:id/...
+# Savepoint state
+/savepoints/savepoint-:shortjobid-:savepointid/...
```
-The savepoint file is usually much smaller than the actual checkpointed state. Note that if you use the `MemoryStateBackend`, the savepoint file will be self-contained and contain all the state.
+<div class="alert alert-info">
+ <strong>Note:</strong>
+Although it looks as if the savepoints may be moved, it is currently not possible due to absolute paths in the <code>_metadata</code> file.
+Please follow <a href="https://issues.apache.org/jira/browse/FLINK-5778">FLINK-5778</a> for progress on lifting this restriction.
+</div>
+
+Note that if you use the `MemoryStateBackend`, metadata *and* savepoint state will be stored in the `_metadata` file. Since it is self-contained, you may move the file and restore from any location.
#### Trigger a Savepoint
@@ -111,7 +126,7 @@ If you don't specify a target directory, you need to have [configured a default
$ bin/flink run -s :savepointPath [:runArgs]
```
-This submits a job and specifies the savepoint path. The execution will resume from the respective savepoint state. The savepoint file holds the meta data of a checkpoint and points to the actual checkpoint files. This is why the savepoint file is usually much smaller than the actual checkpoint state.
+This submits a job and specifies a savepoint to resume from. You may give a path to either the savepoint's directory or the `_metadata` file.
#### Allowing Non-Restored State
@@ -129,11 +144,11 @@ $ bin/flink savepoint -d :savepointPath
This disposes the savepoint stored in `:savepointPath`.
-Note that since savepoints always go to a file system it is possible to also manually delete the savepoint via a regular file system operation. Keep in mind though that the savepoint only stores meta data that points to the actual checkpoint data. Therefore, if you manually want to delete a savepoint, you would have to include the checkpoint files as well. Since there is currently no straight forward way to figure out how a savepoint maps to a checkpoint, it is recommended to use the savepoint tool for this as described above.
+Note that it is possible to also manually delete a savepoint via regular file system operations without affecting other savepoints or checkpoints (recall that each savepoint is self-contained). Up to Flink 1.2, this was a more tedious task which was performed with the savepoint command above.
### Configuration
-You can configure a default savepoint target directory via the `state.savepoints.dir` key. When triggering savepoints, this directory will be used to store the savepoint meta data. You can overwrite the default by specifying a custom target directory with the trigger commands (see the [`:targetDirectory` argument](#trigger-a-savepoint)).
+You can configure a default savepoint target directory via the `state.savepoints.dir` key. When triggering savepoints, this directory will be used to store the savepoint. You can overwrite the default by specifying a custom target directory with the trigger commands (see the [`:targetDirectory` argument](#trigger-a-savepoint)).
```sh
# Default savepoint target directory
@@ -150,10 +165,6 @@ As a rule of thumb, yes. Strictly speaking, it is sufficient to only assign IDs
In practice, it is recommended to assign it to all operators, because some of Flink's built-in operators like the Window operator are also stateful and it is not obvious which built-in operators are actually stateful and which are not. If you are absolutely certain that an operator is stateless, you can skip the `uid` method.
-### Why is the savepoint file so small?
-
-The savepoint file only contains the meta data of the checkpoint and has pointers to the checkpoint state, which is usually much larger. In case of using the `MemoryStateBackend`, the checkpoint will include all state, but is constrained by the backend to small state.
-
### What happens if I add a new operator that requires state to my job?
When you add a new operator to your job it will be initialized without any state. Savepoints contain the state of each stateful operator. Stateless operators are simply not part of the savepoint. The new operator behaves similar to a stateless operator.
@@ -184,4 +195,4 @@ If you did not assign IDs, the auto generated IDs of the stateful operators will
If the savepoint was triggered with Flink >= 1.2.0 and using no deprecated state API like `Checkpointed`, you can simply restore the program from a savepoint and specify a new parallelism.
-If you are resuming from a savepoint triggered with Flink < 1.2.0 or using now deprecated APIs you first have to migrate your job and savepoint to Flink 1.2.0 before being able to change the parallelism. See the [upgrading jobs and Flink versions guide]({{ site.baseurl }}/ops/upgrading.html).
+If you are resuming from a savepoint triggered with Flink < 1.2.0 or using now deprecated APIs you first have to migrate your job and savepoint to Flink >= 1.2.0 before being able to change the parallelism. See the [upgrading jobs and Flink versions guide]({{ site.baseurl }}/ops/upgrading.html).
[19/21] flink git commit: [FLINK-6786] [metrics] Deduplicate
QueryScopeIntoTest
Posted by ch...@apache.org.
[FLINK-6786] [metrics] Deduplicate QueryScopeIntoTest
This closes #4034.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/b863b90c
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/b863b90c
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/b863b90c
Branch: refs/heads/master
Commit: b863b90c80a63e950106b4460bb81f733819dbc2
Parents: 1f6be83
Author: zentol <ch...@apache.org>
Authored: Wed May 31 16:42:26 2017 +0200
Committer: zentol <ch...@apache.org>
Committed: Sat Jun 24 22:09:06 2017 +0200
----------------------------------------------------------------------
.../metrics/dump/QueryScopeInfoTest.java | 140 ++++++++++++----
.../metrics/groups/QueryScopeInfoTest.java | 161 -------------------
2 files changed, 111 insertions(+), 190 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/b863b90c/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/dump/QueryScopeInfoTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/dump/QueryScopeInfoTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/dump/QueryScopeInfoTest.java
index f4f9515..d09063a 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/dump/QueryScopeInfoTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/dump/QueryScopeInfoTest.java
@@ -20,11 +20,6 @@ package org.apache.flink.runtime.metrics.dump;
import org.junit.Test;
-import static org.apache.flink.runtime.metrics.dump.QueryScopeInfo.INFO_CATEGORY_JM;
-import static org.apache.flink.runtime.metrics.dump.QueryScopeInfo.INFO_CATEGORY_JOB;
-import static org.apache.flink.runtime.metrics.dump.QueryScopeInfo.INFO_CATEGORY_OPERATOR;
-import static org.apache.flink.runtime.metrics.dump.QueryScopeInfo.INFO_CATEGORY_TASK;
-import static org.apache.flink.runtime.metrics.dump.QueryScopeInfo.INFO_CATEGORY_TM;
import static org.junit.Assert.assertEquals;
/**
@@ -32,46 +27,133 @@ import static org.junit.Assert.assertEquals;
*/
public class QueryScopeInfoTest {
@Test
- public void testJobManagerMetricInfo() {
- QueryScopeInfo.JobManagerQueryScopeInfo info = new QueryScopeInfo.JobManagerQueryScopeInfo("abc");
- assertEquals("abc", info.scope);
- assertEquals(INFO_CATEGORY_JM, info.getCategory());
+ public void testJobManagerQueryScopeInfo() {
+ QueryScopeInfo.JobManagerQueryScopeInfo info = new QueryScopeInfo.JobManagerQueryScopeInfo();
+ assertEquals(QueryScopeInfo.INFO_CATEGORY_JM, info.getCategory());
+ assertEquals("", info.scope);
+
+ info = info.copy("world");
+ assertEquals(QueryScopeInfo.INFO_CATEGORY_JM, info.getCategory());
+ assertEquals("world", info.scope);
+
+ info = new QueryScopeInfo.JobManagerQueryScopeInfo("hello");
+ assertEquals(QueryScopeInfo.INFO_CATEGORY_JM, info.getCategory());
+ assertEquals("hello", info.scope);
+
+ info = info.copy("world");
+ assertEquals(QueryScopeInfo.INFO_CATEGORY_JM, info.getCategory());
+ assertEquals("hello.world", info.scope);
}
@Test
- public void testTaskManagerMetricInfo() {
- QueryScopeInfo.TaskManagerQueryScopeInfo info = new QueryScopeInfo.TaskManagerQueryScopeInfo("tmid", "abc");
- assertEquals("abc", info.scope);
+ public void testTaskManagerQueryScopeInfo() {
+ QueryScopeInfo.TaskManagerQueryScopeInfo info = new QueryScopeInfo.TaskManagerQueryScopeInfo("tmid");
+ assertEquals(QueryScopeInfo.INFO_CATEGORY_TM, info.getCategory());
+ assertEquals("", info.scope);
+ assertEquals("tmid", info.taskManagerID);
+
+ info = info.copy("world");
+ assertEquals(QueryScopeInfo.INFO_CATEGORY_TM, info.getCategory());
+ assertEquals("world", info.scope);
+ assertEquals("tmid", info.taskManagerID);
+
+ info = new QueryScopeInfo.TaskManagerQueryScopeInfo("tmid", "hello");
+ assertEquals(QueryScopeInfo.INFO_CATEGORY_TM, info.getCategory());
+ assertEquals("hello", info.scope);
+ assertEquals("tmid", info.taskManagerID);
+
+ info = info.copy("world");
+ assertEquals(QueryScopeInfo.INFO_CATEGORY_TM, info.getCategory());
+ assertEquals("hello.world", info.scope);
assertEquals("tmid", info.taskManagerID);
- assertEquals(INFO_CATEGORY_TM, info.getCategory());
}
@Test
- public void testJobMetricInfo() {
- QueryScopeInfo.JobQueryScopeInfo info = new QueryScopeInfo.JobQueryScopeInfo("jobid", "abc");
- assertEquals("abc", info.scope);
+ public void testJobQueryScopeInfo() {
+ QueryScopeInfo.JobQueryScopeInfo info = new QueryScopeInfo.JobQueryScopeInfo("jobid");
+ assertEquals(QueryScopeInfo.INFO_CATEGORY_JOB, info.getCategory());
+ assertEquals("", info.scope);
+ assertEquals("jobid", info.jobID);
+
+ info = info.copy("world");
+ assertEquals(QueryScopeInfo.INFO_CATEGORY_JOB, info.getCategory());
+ assertEquals("world", info.scope);
+ assertEquals("jobid", info.jobID);
+
+ info = new QueryScopeInfo.JobQueryScopeInfo("jobid", "hello");
+ assertEquals(QueryScopeInfo.INFO_CATEGORY_JOB, info.getCategory());
+ assertEquals("hello", info.scope);
+ assertEquals("jobid", info.jobID);
+
+ info = info.copy("world");
+ assertEquals(QueryScopeInfo.INFO_CATEGORY_JOB, info.getCategory());
+ assertEquals("hello.world", info.scope);
assertEquals("jobid", info.jobID);
- assertEquals(INFO_CATEGORY_JOB, info.getCategory());
}
@Test
- public void testTaskMetricInfo() {
- QueryScopeInfo.TaskQueryScopeInfo info = new QueryScopeInfo.TaskQueryScopeInfo("jid", "vid", 2, "abc");
- assertEquals("abc", info.scope);
- assertEquals("jid", info.jobID);
- assertEquals("vid", info.vertexID);
+ public void testTaskQueryScopeInfo() {
+ QueryScopeInfo.TaskQueryScopeInfo info = new QueryScopeInfo.TaskQueryScopeInfo("jobid", "taskid", 2);
+ assertEquals(QueryScopeInfo.INFO_CATEGORY_TASK, info.getCategory());
+ assertEquals("", info.scope);
+ assertEquals("jobid", info.jobID);
+ assertEquals("taskid", info.vertexID);
+ assertEquals(2, info.subtaskIndex);
+
+ info = info.copy("world");
+ assertEquals(QueryScopeInfo.INFO_CATEGORY_TASK, info.getCategory());
+ assertEquals("world", info.scope);
+ assertEquals("jobid", info.jobID);
+ assertEquals("taskid", info.vertexID);
+ assertEquals(2, info.subtaskIndex);
+
+ info = new QueryScopeInfo.TaskQueryScopeInfo("jobid", "taskid", 2, "hello");
+ assertEquals(QueryScopeInfo.INFO_CATEGORY_TASK, info.getCategory());
+ assertEquals("hello", info.scope);
+ assertEquals("jobid", info.jobID);
+ assertEquals("taskid", info.vertexID);
+ assertEquals(2, info.subtaskIndex);
+
+ info = info.copy("world");
+ assertEquals(QueryScopeInfo.INFO_CATEGORY_TASK, info.getCategory());
+ assertEquals("hello.world", info.scope);
+ assertEquals("jobid", info.jobID);
+ assertEquals("taskid", info.vertexID);
assertEquals(2, info.subtaskIndex);
- assertEquals(INFO_CATEGORY_TASK, info.getCategory());
}
@Test
- public void testOperatorMetricInfo() {
- QueryScopeInfo.OperatorQueryScopeInfo info = new QueryScopeInfo.OperatorQueryScopeInfo("jid", "vid", 2, "opname", "abc");
- assertEquals("abc", info.scope);
- assertEquals("jid", info.jobID);
- assertEquals("vid", info.vertexID);
+ public void testOperatorQueryScopeInfo() {
+ QueryScopeInfo.OperatorQueryScopeInfo info = new QueryScopeInfo.OperatorQueryScopeInfo("jobid", "taskid", 2, "opname");
+ assertEquals(QueryScopeInfo.INFO_CATEGORY_OPERATOR, info.getCategory());
+ assertEquals("", info.scope);
+ assertEquals("jobid", info.jobID);
+ assertEquals("taskid", info.vertexID);
+ assertEquals("opname", info.operatorName);
+ assertEquals(2, info.subtaskIndex);
+
+ info = info.copy("world");
+ assertEquals(QueryScopeInfo.INFO_CATEGORY_OPERATOR, info.getCategory());
+ assertEquals("world", info.scope);
+ assertEquals("jobid", info.jobID);
+ assertEquals("taskid", info.vertexID);
+ assertEquals("opname", info.operatorName);
+ assertEquals(2, info.subtaskIndex);
+
+ info = new QueryScopeInfo.OperatorQueryScopeInfo("jobid", "taskid", 2, "opname", "hello");
+ assertEquals(QueryScopeInfo.INFO_CATEGORY_OPERATOR, info.getCategory());
+ assertEquals("hello", info.scope);
+ assertEquals("jobid", info.jobID);
+ assertEquals("taskid", info.vertexID);
+ assertEquals("opname", info.operatorName);
+ assertEquals(2, info.subtaskIndex);
+
+ info = info.copy("world");
+ assertEquals(QueryScopeInfo.INFO_CATEGORY_OPERATOR, info.getCategory());
+ assertEquals("hello.world", info.scope);
+ assertEquals("jobid", info.jobID);
+ assertEquals("taskid", info.vertexID);
assertEquals("opname", info.operatorName);
assertEquals(2, info.subtaskIndex);
- assertEquals(INFO_CATEGORY_OPERATOR, info.getCategory());
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/b863b90c/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/groups/QueryScopeInfoTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/groups/QueryScopeInfoTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/groups/QueryScopeInfoTest.java
deleted file mode 100644
index f6a1277..0000000
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/groups/QueryScopeInfoTest.java
+++ /dev/null
@@ -1,161 +0,0 @@
-/*
- * 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.metrics.groups;
-
-import org.apache.flink.runtime.metrics.dump.QueryScopeInfo;
-
-import org.junit.Test;
-
-import static org.junit.Assert.assertEquals;
-
-/**
- * Tests for the {@link QueryScopeInfo} classes.
- */
-public class QueryScopeInfoTest {
- @Test
- public void testJobManagerQueryScopeInfo() {
- QueryScopeInfo.JobManagerQueryScopeInfo info = new QueryScopeInfo.JobManagerQueryScopeInfo();
- assertEquals(QueryScopeInfo.INFO_CATEGORY_JM, info.getCategory());
- assertEquals("", info.scope);
-
- info = info.copy("world");
- assertEquals(QueryScopeInfo.INFO_CATEGORY_JM, info.getCategory());
- assertEquals("world", info.scope);
-
- info = new QueryScopeInfo.JobManagerQueryScopeInfo("hello");
- assertEquals(QueryScopeInfo.INFO_CATEGORY_JM, info.getCategory());
- assertEquals("hello", info.scope);
-
- info = info.copy("world");
- assertEquals(QueryScopeInfo.INFO_CATEGORY_JM, info.getCategory());
- assertEquals("hello.world", info.scope);
- }
-
- @Test
- public void testTaskManagerQueryScopeInfo() {
- QueryScopeInfo.TaskManagerQueryScopeInfo info = new QueryScopeInfo.TaskManagerQueryScopeInfo("tmid");
- assertEquals(QueryScopeInfo.INFO_CATEGORY_TM, info.getCategory());
- assertEquals("", info.scope);
- assertEquals("tmid", info.taskManagerID);
-
- info = info.copy("world");
- assertEquals(QueryScopeInfo.INFO_CATEGORY_TM, info.getCategory());
- assertEquals("world", info.scope);
- assertEquals("tmid", info.taskManagerID);
-
- info = new QueryScopeInfo.TaskManagerQueryScopeInfo("tmid", "hello");
- assertEquals(QueryScopeInfo.INFO_CATEGORY_TM, info.getCategory());
- assertEquals("hello", info.scope);
- assertEquals("tmid", info.taskManagerID);
-
- info = info.copy("world");
- assertEquals(QueryScopeInfo.INFO_CATEGORY_TM, info.getCategory());
- assertEquals("hello.world", info.scope);
- assertEquals("tmid", info.taskManagerID);
- }
-
- @Test
- public void testJobQueryScopeInfo() {
- QueryScopeInfo.JobQueryScopeInfo info = new QueryScopeInfo.JobQueryScopeInfo("jobid");
- assertEquals(QueryScopeInfo.INFO_CATEGORY_JOB, info.getCategory());
- assertEquals("", info.scope);
- assertEquals("jobid", info.jobID);
-
- info = info.copy("world");
- assertEquals(QueryScopeInfo.INFO_CATEGORY_JOB, info.getCategory());
- assertEquals("world", info.scope);
- assertEquals("jobid", info.jobID);
-
- info = new QueryScopeInfo.JobQueryScopeInfo("jobid", "hello");
- assertEquals(QueryScopeInfo.INFO_CATEGORY_JOB, info.getCategory());
- assertEquals("hello", info.scope);
- assertEquals("jobid", info.jobID);
-
- info = info.copy("world");
- assertEquals(QueryScopeInfo.INFO_CATEGORY_JOB, info.getCategory());
- assertEquals("hello.world", info.scope);
- assertEquals("jobid", info.jobID);
- }
-
- @Test
- public void testTaskQueryScopeInfo() {
- QueryScopeInfo.TaskQueryScopeInfo info = new QueryScopeInfo.TaskQueryScopeInfo("jobid", "taskid", 2);
- assertEquals(QueryScopeInfo.INFO_CATEGORY_TASK, info.getCategory());
- assertEquals("", info.scope);
- assertEquals("jobid", info.jobID);
- assertEquals("taskid", info.vertexID);
- assertEquals(2, info.subtaskIndex);
-
- info = info.copy("world");
- assertEquals(QueryScopeInfo.INFO_CATEGORY_TASK, info.getCategory());
- assertEquals("world", info.scope);
- assertEquals("jobid", info.jobID);
- assertEquals("taskid", info.vertexID);
- assertEquals(2, info.subtaskIndex);
-
- info = new QueryScopeInfo.TaskQueryScopeInfo("jobid", "taskid", 2, "hello");
- assertEquals(QueryScopeInfo.INFO_CATEGORY_TASK, info.getCategory());
- assertEquals("hello", info.scope);
- assertEquals("jobid", info.jobID);
- assertEquals("taskid", info.vertexID);
- assertEquals(2, info.subtaskIndex);
-
- info = info.copy("world");
- assertEquals(QueryScopeInfo.INFO_CATEGORY_TASK, info.getCategory());
- assertEquals("hello.world", info.scope);
- assertEquals("jobid", info.jobID);
- assertEquals("taskid", info.vertexID);
- assertEquals(2, info.subtaskIndex);
- }
-
- @Test
- public void testOperatorQueryScopeInfo() {
- QueryScopeInfo.OperatorQueryScopeInfo info = new QueryScopeInfo.OperatorQueryScopeInfo("jobid", "taskid", 2, "opname");
- assertEquals(QueryScopeInfo.INFO_CATEGORY_OPERATOR, info.getCategory());
- assertEquals("", info.scope);
- assertEquals("jobid", info.jobID);
- assertEquals("taskid", info.vertexID);
- assertEquals("opname", info.operatorName);
- assertEquals(2, info.subtaskIndex);
-
- info = info.copy("world");
- assertEquals(QueryScopeInfo.INFO_CATEGORY_OPERATOR, info.getCategory());
- assertEquals("world", info.scope);
- assertEquals("jobid", info.jobID);
- assertEquals("taskid", info.vertexID);
- assertEquals("opname", info.operatorName);
- assertEquals(2, info.subtaskIndex);
-
- info = new QueryScopeInfo.OperatorQueryScopeInfo("jobid", "taskid", 2, "opname", "hello");
- assertEquals(QueryScopeInfo.INFO_CATEGORY_OPERATOR, info.getCategory());
- assertEquals("hello", info.scope);
- assertEquals("jobid", info.jobID);
- assertEquals("taskid", info.vertexID);
- assertEquals("opname", info.operatorName);
- assertEquals(2, info.subtaskIndex);
-
- info = info.copy("world");
- assertEquals(QueryScopeInfo.INFO_CATEGORY_OPERATOR, info.getCategory());
- assertEquals("hello.world", info.scope);
- assertEquals("jobid", info.jobID);
- assertEquals("taskid", info.vertexID);
- assertEquals("opname", info.operatorName);
- assertEquals(2, info.subtaskIndex);
- }
-}
[16/21] flink git commit: [FLINK-6488] [scripts] Deprecate
'start-local.sh' and 'stop-local.sh' scripts
Posted by ch...@apache.org.
[FLINK-6488] [scripts] Deprecate 'start-local.sh' and 'stop-local.sh' scripts
This closes #4074.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/3557ff86
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/3557ff86
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/3557ff86
Branch: refs/heads/master
Commit: 3557ff8636e899d56ba23534944669aeca2b5df4
Parents: c736ba2
Author: zhangminglei <zm...@163.com>
Authored: Tue Jun 20 18:28:33 2017 +0800
Committer: zentol <ch...@apache.org>
Committed: Fri Jun 23 14:14:30 2017 +0200
----------------------------------------------------------------------
flink-dist/src/main/flink-bin/bin/start-local.sh | 5 +++--
flink-dist/src/main/flink-bin/bin/stop-local.sh | 4 ++--
2 files changed, 5 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/3557ff86/flink-dist/src/main/flink-bin/bin/start-local.sh
----------------------------------------------------------------------
diff --git a/flink-dist/src/main/flink-bin/bin/start-local.sh b/flink-dist/src/main/flink-bin/bin/start-local.sh
index b3d41bf..2df3c22 100755
--- a/flink-dist/src/main/flink-bin/bin/start-local.sh
+++ b/flink-dist/src/main/flink-bin/bin/start-local.sh
@@ -19,10 +19,11 @@
STREAMING_MODE=$1
+echo "Warning: this file is deprecated and will be removed in 1.5."
+
bin=`dirname "$0"`
bin=`cd "$bin"; pwd`
. "$bin"/config.sh
-# local mode, only bring up job manager. The job manager will start an internal task manager
-"$FLINK_BIN_DIR"/jobmanager.sh start local
+"$FLINK_BIN_DIR"/start-cluster.sh
http://git-wip-us.apache.org/repos/asf/flink/blob/3557ff86/flink-dist/src/main/flink-bin/bin/stop-local.sh
----------------------------------------------------------------------
diff --git a/flink-dist/src/main/flink-bin/bin/stop-local.sh b/flink-dist/src/main/flink-bin/bin/stop-local.sh
index 79627fa..8d079e0 100755
--- a/flink-dist/src/main/flink-bin/bin/stop-local.sh
+++ b/flink-dist/src/main/flink-bin/bin/stop-local.sh
@@ -17,11 +17,11 @@
# limitations under the License.
################################################################################
+echo "Warning: this file is deprecated and will be removed in 1.5."
bin=`dirname "$0"`
bin=`cd "$bin"; pwd`
. "$bin"/config.sh
-# stop local job manager (has an internal task manager)
-"$FLINK_BIN_DIR"/jobmanager.sh stop
+"$FLINK_BIN_DIR"/stop-cluster.sh
[06/21] flink git commit: [FLINK-6920] Remove minor guava usages
Posted by ch...@apache.org.
[FLINK-6920] Remove minor guava usages
This closes #4124.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/3424924e
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/3424924e
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/3424924e
Branch: refs/heads/master
Commit: 3424924e49898cd2d581eafac338a306c9e0efc5
Parents: 72b0ae0
Author: zentol <ch...@apache.org>
Authored: Wed Jun 14 15:18:25 2017 +0200
Committer: zentol <ch...@apache.org>
Committed: Fri Jun 23 14:14:29 2017 +0200
----------------------------------------------------------------------
.../connectors/elasticsearch/ElasticsearchSinkITCase.java | 6 +++---
.../connectors/fs/RollingSinkFaultToleranceITCase.java | 3 +--
.../fs/bucketing/BucketingSinkFaultToleranceITCase.java | 3 +--
.../graph/test/examples/EuclideanGraphWeighingITCase.java | 7 +++----
.../flink/graph/test/examples/IncrementalSSSPITCase.java | 9 ++++-----
.../flink/graph/test/examples/MusicProfilesITCase.java | 7 +++----
.../test/examples/SingleSourceShortestPathsITCase.java | 5 ++---
.../scala/test/operations/GraphCreationWithCsvITCase.scala | 4 ++--
.../java/org/apache/flink/test/util/AbstractTestBase.java | 9 ++++++---
9 files changed, 25 insertions(+), 28 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/3424924e/flink-connectors/flink-connector-elasticsearch/src/test/java/org/apache/flink/streaming/connectors/elasticsearch/ElasticsearchSinkITCase.java
----------------------------------------------------------------------
diff --git a/flink-connectors/flink-connector-elasticsearch/src/test/java/org/apache/flink/streaming/connectors/elasticsearch/ElasticsearchSinkITCase.java b/flink-connectors/flink-connector-elasticsearch/src/test/java/org/apache/flink/streaming/connectors/elasticsearch/ElasticsearchSinkITCase.java
index ecbebd7..1fcfa2e 100644
--- a/flink-connectors/flink-connector-elasticsearch/src/test/java/org/apache/flink/streaming/connectors/elasticsearch/ElasticsearchSinkITCase.java
+++ b/flink-connectors/flink-connector-elasticsearch/src/test/java/org/apache/flink/streaming/connectors/elasticsearch/ElasticsearchSinkITCase.java
@@ -25,7 +25,6 @@ import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.elasticsearch.testutils.SourceSinkDataTestKit;
import org.apache.flink.streaming.connectors.elasticsearch.util.ElasticsearchUtils;
-import com.google.common.collect.Lists;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.Client;
import org.elasticsearch.client.Requests;
@@ -34,6 +33,7 @@ import org.elasticsearch.common.transport.TransportAddress;
import org.junit.Test;
import java.net.InetSocketAddress;
+import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -113,7 +113,7 @@ public class ElasticsearchSinkITCase extends ElasticsearchSinkTestBase {
userConfig.put("cluster.name", CLUSTER_NAME);
userConfig.put("node.local", "true");
- List<TransportAddress> transports = Lists.newArrayList();
+ List<TransportAddress> transports = new ArrayList<>();
transports.add(new LocalTransportAddress("1"));
source.addSink(new ElasticsearchSink<>(
@@ -146,7 +146,7 @@ public class ElasticsearchSinkITCase extends ElasticsearchSinkTestBase {
// LocalTransportAddress to connect to a local embedded node
userConfig.put("node.local", "true");
- List<TransportAddress> transports = Lists.newArrayList();
+ List<TransportAddress> transports = new ArrayList<>();
transports.add(new LocalTransportAddress("1"));
return new ElasticsearchSink<>(
http://git-wip-us.apache.org/repos/asf/flink/blob/3424924e/flink-connectors/flink-connector-filesystem/src/test/java/org/apache/flink/streaming/connectors/fs/RollingSinkFaultToleranceITCase.java
----------------------------------------------------------------------
diff --git a/flink-connectors/flink-connector-filesystem/src/test/java/org/apache/flink/streaming/connectors/fs/RollingSinkFaultToleranceITCase.java b/flink-connectors/flink-connector-filesystem/src/test/java/org/apache/flink/streaming/connectors/fs/RollingSinkFaultToleranceITCase.java
index b096db4..7473685 100644
--- a/flink-connectors/flink-connector-filesystem/src/test/java/org/apache/flink/streaming/connectors/fs/RollingSinkFaultToleranceITCase.java
+++ b/flink-connectors/flink-connector-filesystem/src/test/java/org/apache/flink/streaming/connectors/fs/RollingSinkFaultToleranceITCase.java
@@ -26,7 +26,6 @@ import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunctio
import org.apache.flink.test.checkpointing.StreamFaultToleranceTestBase;
import org.apache.flink.util.NetUtils;
-import com.google.common.collect.Sets;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.LocatedFileStatus;
@@ -139,7 +138,7 @@ public class RollingSinkFaultToleranceITCase extends StreamFaultToleranceTestBas
// Keep a set of the message IDs that we read. The size must equal the read count and
// the NUM_STRINGS. If numRead is bigger than the size of the set we have seen some
// elements twice.
- Set<Integer> readNumbers = Sets.newHashSet();
+ Set<Integer> readNumbers = new HashSet<>();
HashSet<String> uniqMessagesRead = new HashSet<>();
HashSet<String> messagesInCommittedFiles = new HashSet<>();
http://git-wip-us.apache.org/repos/asf/flink/blob/3424924e/flink-connectors/flink-connector-filesystem/src/test/java/org/apache/flink/streaming/connectors/fs/bucketing/BucketingSinkFaultToleranceITCase.java
----------------------------------------------------------------------
diff --git a/flink-connectors/flink-connector-filesystem/src/test/java/org/apache/flink/streaming/connectors/fs/bucketing/BucketingSinkFaultToleranceITCase.java b/flink-connectors/flink-connector-filesystem/src/test/java/org/apache/flink/streaming/connectors/fs/bucketing/BucketingSinkFaultToleranceITCase.java
index 1ed4a7f..5e417d0 100644
--- a/flink-connectors/flink-connector-filesystem/src/test/java/org/apache/flink/streaming/connectors/fs/bucketing/BucketingSinkFaultToleranceITCase.java
+++ b/flink-connectors/flink-connector-filesystem/src/test/java/org/apache/flink/streaming/connectors/fs/bucketing/BucketingSinkFaultToleranceITCase.java
@@ -26,7 +26,6 @@ import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunctio
import org.apache.flink.test.checkpointing.StreamFaultToleranceTestBase;
import org.apache.flink.util.NetUtils;
-import com.google.common.collect.Sets;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.LocatedFileStatus;
@@ -136,7 +135,7 @@ public class BucketingSinkFaultToleranceITCase extends StreamFaultToleranceTestB
// Keep a set of the message IDs that we read. The size must equal the read count and
// the NUM_STRINGS. If numRead is bigger than the size of the set we have seen some
// elements twice.
- Set<Integer> readNumbers = Sets.newHashSet();
+ Set<Integer> readNumbers = new HashSet<>();
HashSet<String> uniqMessagesRead = new HashSet<>();
HashSet<String> messagesInCommittedFiles = new HashSet<>();
http://git-wip-us.apache.org/repos/asf/flink/blob/3424924e/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/test/examples/EuclideanGraphWeighingITCase.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/test/examples/EuclideanGraphWeighingITCase.java b/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/test/examples/EuclideanGraphWeighingITCase.java
index deac519..d75684b 100644
--- a/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/test/examples/EuclideanGraphWeighingITCase.java
+++ b/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/test/examples/EuclideanGraphWeighingITCase.java
@@ -22,9 +22,8 @@ import org.apache.flink.graph.examples.EuclideanGraphWeighing;
import org.apache.flink.graph.examples.data.EuclideanGraphData;
import org.apache.flink.test.util.MultipleProgramsTestBase;
import org.apache.flink.test.util.TestBaseUtils;
+import org.apache.flink.util.FileUtils;
-import com.google.common.base.Charsets;
-import com.google.common.io.Files;
import org.junit.After;
import org.junit.Before;
import org.junit.Rule;
@@ -60,10 +59,10 @@ public class EuclideanGraphWeighingITCase extends MultipleProgramsTestBase {
public void before() throws Exception {
resultPath = tempFolder.newFile().toURI().toString();
File verticesFile = tempFolder.newFile();
- Files.write(EuclideanGraphData.VERTICES, verticesFile, Charsets.UTF_8);
+ FileUtils.writeFileUtf8(verticesFile, EuclideanGraphData.VERTICES);
File edgesFile = tempFolder.newFile();
- Files.write(EuclideanGraphData.EDGES, edgesFile, Charsets.UTF_8);
+ FileUtils.writeFileUtf8(edgesFile, EuclideanGraphData.EDGES);
verticesPath = verticesFile.toURI().toString();
edgesPath = edgesFile.toURI().toString();
http://git-wip-us.apache.org/repos/asf/flink/blob/3424924e/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/test/examples/IncrementalSSSPITCase.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/test/examples/IncrementalSSSPITCase.java b/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/test/examples/IncrementalSSSPITCase.java
index 9147a6d..8c0a032 100644
--- a/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/test/examples/IncrementalSSSPITCase.java
+++ b/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/test/examples/IncrementalSSSPITCase.java
@@ -29,9 +29,8 @@ import org.apache.flink.graph.examples.data.IncrementalSSSPData;
import org.apache.flink.graph.spargel.ScatterGatherConfiguration;
import org.apache.flink.test.util.MultipleProgramsTestBase;
import org.apache.flink.test.util.TestBaseUtils;
+import org.apache.flink.util.FileUtils;
-import com.google.common.base.Charsets;
-import com.google.common.io.Files;
import org.junit.After;
import org.junit.Before;
import org.junit.Rule;
@@ -69,13 +68,13 @@ public class IncrementalSSSPITCase extends MultipleProgramsTestBase {
public void before() throws Exception {
resultPath = tempFolder.newFile().toURI().toString();
File verticesFile = tempFolder.newFile();
- Files.write(IncrementalSSSPData.VERTICES, verticesFile, Charsets.UTF_8);
+ FileUtils.writeFileUtf8(verticesFile, IncrementalSSSPData.VERTICES);
File edgesFile = tempFolder.newFile();
- Files.write(IncrementalSSSPData.EDGES, edgesFile, Charsets.UTF_8);
+ FileUtils.writeFileUtf8(edgesFile, IncrementalSSSPData.EDGES);
File edgesInSSSPFile = tempFolder.newFile();
- Files.write(IncrementalSSSPData.EDGES_IN_SSSP, edgesInSSSPFile, Charsets.UTF_8);
+ FileUtils.writeFileUtf8(edgesInSSSPFile, IncrementalSSSPData.EDGES_IN_SSSP);
verticesPath = verticesFile.toURI().toString();
edgesPath = edgesFile.toURI().toString();
http://git-wip-us.apache.org/repos/asf/flink/blob/3424924e/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/test/examples/MusicProfilesITCase.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/test/examples/MusicProfilesITCase.java b/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/test/examples/MusicProfilesITCase.java
index be412cf..98eb166 100644
--- a/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/test/examples/MusicProfilesITCase.java
+++ b/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/test/examples/MusicProfilesITCase.java
@@ -22,9 +22,8 @@ import org.apache.flink.graph.examples.MusicProfiles;
import org.apache.flink.graph.examples.data.MusicProfilesData;
import org.apache.flink.test.util.MultipleProgramsTestBase;
import org.apache.flink.test.util.TestBaseUtils;
+import org.apache.flink.util.FileUtils;
-import com.google.common.base.Charsets;
-import com.google.common.io.Files;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
@@ -67,11 +66,11 @@ public class MusicProfilesITCase extends MultipleProgramsTestBase {
communitiesResultPath = tempFolder.newFile().toURI().toString();
File tripletsFile = tempFolder.newFile();
- Files.write(MusicProfilesData.USER_SONG_TRIPLETS, tripletsFile, Charsets.UTF_8);
+ FileUtils.writeFileUtf8(tripletsFile, MusicProfilesData.USER_SONG_TRIPLETS);
tripletsPath = tripletsFile.toURI().toString();
File mismatchesFile = tempFolder.newFile();
- Files.write(MusicProfilesData.MISMATCHES, mismatchesFile, Charsets.UTF_8);
+ FileUtils.writeFileUtf8(mismatchesFile, MusicProfilesData.MISMATCHES);
mismatchesPath = mismatchesFile.toURI().toString();
}
http://git-wip-us.apache.org/repos/asf/flink/blob/3424924e/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/test/examples/SingleSourceShortestPathsITCase.java
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/test/examples/SingleSourceShortestPathsITCase.java b/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/test/examples/SingleSourceShortestPathsITCase.java
index 60bd161..24a2f3d 100644
--- a/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/test/examples/SingleSourceShortestPathsITCase.java
+++ b/flink-libraries/flink-gelly-examples/src/test/java/org/apache/flink/graph/test/examples/SingleSourceShortestPathsITCase.java
@@ -24,9 +24,8 @@ import org.apache.flink.graph.examples.SingleSourceShortestPaths;
import org.apache.flink.graph.examples.data.SingleSourceShortestPathsData;
import org.apache.flink.test.util.MultipleProgramsTestBase;
import org.apache.flink.test.util.TestBaseUtils;
+import org.apache.flink.util.FileUtils;
-import com.google.common.base.Charsets;
-import com.google.common.io.Files;
import org.junit.After;
import org.junit.Before;
import org.junit.Rule;
@@ -61,7 +60,7 @@ public class SingleSourceShortestPathsITCase extends MultipleProgramsTestBase {
resultPath = tempFolder.newFile().toURI().toString();
File edgesFile = tempFolder.newFile();
- Files.write(SingleSourceShortestPathsData.EDGES, edgesFile, Charsets.UTF_8);
+ FileUtils.writeFileUtf8(edgesFile, SingleSourceShortestPathsData.EDGES);
edgesPath = edgesFile.toURI().toString();
}
http://git-wip-us.apache.org/repos/asf/flink/blob/3424924e/flink-libraries/flink-gelly-scala/src/test/scala/org/apache/flink/graph/scala/test/operations/GraphCreationWithCsvITCase.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-gelly-scala/src/test/scala/org/apache/flink/graph/scala/test/operations/GraphCreationWithCsvITCase.scala b/flink-libraries/flink-gelly-scala/src/test/scala/org/apache/flink/graph/scala/test/operations/GraphCreationWithCsvITCase.scala
index 3d36af7..bce65af 100644
--- a/flink-libraries/flink-gelly-scala/src/test/scala/org/apache/flink/graph/scala/test/operations/GraphCreationWithCsvITCase.scala
+++ b/flink-libraries/flink-gelly-scala/src/test/scala/org/apache/flink/graph/scala/test/operations/GraphCreationWithCsvITCase.scala
@@ -19,8 +19,8 @@
package org.apache.flink.graph.scala.test.operations
import java.io.{File, FileOutputStream, IOException, OutputStreamWriter}
+import java.nio.charset.StandardCharsets
-import com.google.common.base.Charsets
import org.apache.flink.api.common.functions.MapFunction
import org.apache.flink.api.scala._
import org.apache.flink.core.fs.{FileInputSplit, Path}
@@ -195,7 +195,7 @@ MultipleProgramsTestBase(mode) {
val tempFile = File.createTempFile("test_contents", "tmp")
tempFile.deleteOnExit()
- val wrt = new OutputStreamWriter(new FileOutputStream(tempFile), Charsets.UTF_8)
+ val wrt = new OutputStreamWriter(new FileOutputStream(tempFile), StandardCharsets.UTF_8)
wrt.write(content)
wrt.close()
http://git-wip-us.apache.org/repos/asf/flink/blob/3424924e/flink-test-utils-parent/flink-test-utils/src/main/java/org/apache/flink/test/util/AbstractTestBase.java
----------------------------------------------------------------------
diff --git a/flink-test-utils-parent/flink-test-utils/src/main/java/org/apache/flink/test/util/AbstractTestBase.java b/flink-test-utils-parent/flink-test-utils/src/main/java/org/apache/flink/test/util/AbstractTestBase.java
index e426fc9..dd0005b 100644
--- a/flink-test-utils-parent/flink-test-utils/src/main/java/org/apache/flink/test/util/AbstractTestBase.java
+++ b/flink-test-utils-parent/flink-test-utils/src/main/java/org/apache/flink/test/util/AbstractTestBase.java
@@ -21,9 +21,8 @@ package org.apache.flink.test.util;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.akka.AkkaUtils;
import org.apache.flink.runtime.minicluster.LocalFlinkMiniCluster;
+import org.apache.flink.util.FileUtils;
-import com.google.common.base.Charsets;
-import com.google.common.io.Files;
import org.junit.ClassRule;
import org.junit.rules.TemporaryFolder;
@@ -112,7 +111,11 @@ public abstract class AbstractTestBase extends TestBaseUtils {
public String createTempFile(String fileName, String contents) throws IOException {
File f = createAndRegisterTempFile(fileName);
- Files.write(contents, f, Charsets.UTF_8);
+ if (!f.getParentFile().exists()) {
+ f.getParentFile().mkdirs();
+ }
+ f.createNewFile();
+ FileUtils.writeFileUtf8(f, contents);
return f.toURI().toString();
}
[15/21] flink git commit: [hotfix] Fix typo in JobManager
Posted by ch...@apache.org.
[hotfix] Fix typo in JobManager
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/a1017e41
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/a1017e41
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/a1017e41
Branch: refs/heads/master
Commit: a1017e4117a27a75da45d787d72f480b7676cb4b
Parents: 3557ff8
Author: zhangminglei <zm...@163.com>
Authored: Tue Jun 20 18:30:07 2017 +0800
Committer: zentol <ch...@apache.org>
Committed: Fri Jun 23 14:14:30 2017 +0200
----------------------------------------------------------------------
.../scala/org/apache/flink/runtime/jobmanager/JobManager.scala | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/a1017e41/flink-runtime/src/main/scala/org/apache/flink/runtime/jobmanager/JobManager.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/scala/org/apache/flink/runtime/jobmanager/JobManager.scala b/flink-runtime/src/main/scala/org/apache/flink/runtime/jobmanager/JobManager.scala
index e488bef..f2934b5 100644
--- a/flink-runtime/src/main/scala/org/apache/flink/runtime/jobmanager/JobManager.scala
+++ b/flink-runtime/src/main/scala/org/apache/flink/runtime/jobmanager/JobManager.scala
@@ -472,7 +472,7 @@ class JobManager(
taskManagerGateway match {
case x: ActorTaskManagerGateway =>
handleTaskManagerTerminated(x.getActorGateway().actor(), instance.getId)
- case _ => log.debug(s"Cannot remove reosurce ${resourceID}, because there is " +
+ case _ => log.debug(s"Cannot remove resource ${resourceID}, because there is " +
s"no ActorRef registered.")
}
[13/21] flink git commit: [FLINK-6943] Improve exceptions within
TypeExtractionUtils#getSingleAbstractMethod
Posted by ch...@apache.org.
[FLINK-6943] Improve exceptions within TypeExtractionUtils#getSingleAbstractMethod
This closes #4140.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/2274bf74
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/2274bf74
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/2274bf74
Branch: refs/heads/master
Commit: 2274bf74f0ac84fdcf6a6109426ada3c61b4454e
Parents: a1017e4
Author: Dawid Wysakowicz <dw...@apache.org>
Authored: Mon Jun 19 09:38:25 2017 +0200
Committer: zentol <ch...@apache.org>
Committed: Fri Jun 23 14:14:30 2017 +0200
----------------------------------------------------------------------
.../api/java/typeutils/TypeExtractionUtils.java | 21 ++++++--
.../java/type/lambdas/LambdaExtractionTest.java | 51 ++++++++++++++++++++
2 files changed, 67 insertions(+), 5 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/2274bf74/flink-core/src/main/java/org/apache/flink/api/java/typeutils/TypeExtractionUtils.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/TypeExtractionUtils.java b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/TypeExtractionUtils.java
index c6ffd55..c2a01c3 100644
--- a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/TypeExtractionUtils.java
+++ b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/TypeExtractionUtils.java
@@ -215,22 +215,33 @@ public class TypeExtractionUtils {
* Extracts a Single Abstract Method (SAM) as defined in Java Specification (4.3.2. The Class Object,
* 9.8 Functional Interfaces, 9.4.3 Interface Method Body) from given class.
*
- * @param baseClass
- * @throws InvalidTypesException if the given class does not implement
- * @return
+ * @param baseClass a class that is a FunctionalInterface to retrieve a SAM from
+ * @throws InvalidTypesException if the given class does not implement FunctionalInterface
+ * @return single abstract method of the given class
*/
public static Method getSingleAbstractMethod(Class<?> baseClass) {
+
+ if (!baseClass.isInterface()) {
+ throw new InvalidTypesException("Given class: " + baseClass + "is not a FunctionalInterface.");
+ }
+
Method sam = null;
for (Method method : baseClass.getMethods()) {
if (Modifier.isAbstract(method.getModifiers())) {
if (sam == null) {
sam = method;
} else {
- throw new InvalidTypesException(
- "Given class: " + baseClass + " is not a FunctionalInterface. It does not have a SAM.");
+ throw new InvalidTypesException("Given class: " + baseClass +
+ " is not a FunctionalInterface. It has more than one abstract method.");
}
}
}
+
+ if (sam == null) {
+ throw new InvalidTypesException(
+ "Given class: " + baseClass + " is not a FunctionalInterface. It does not have any abstract methods.");
+ }
+
return sam;
}
http://git-wip-us.apache.org/repos/asf/flink/blob/2274bf74/flink-java8/src/test/java/org/apache/flink/api/java/type/lambdas/LambdaExtractionTest.java
----------------------------------------------------------------------
diff --git a/flink-java8/src/test/java/org/apache/flink/api/java/type/lambdas/LambdaExtractionTest.java b/flink-java8/src/test/java/org/apache/flink/api/java/type/lambdas/LambdaExtractionTest.java
index 7500d73..175c6fe 100644
--- a/flink-java8/src/test/java/org/apache/flink/api/java/type/lambdas/LambdaExtractionTest.java
+++ b/flink-java8/src/test/java/org/apache/flink/api/java/type/lambdas/LambdaExtractionTest.java
@@ -23,6 +23,7 @@ import org.apache.flink.api.common.functions.CrossFunction;
import org.apache.flink.api.common.functions.FlatJoinFunction;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.GroupReduceFunction;
+import org.apache.flink.api.common.functions.InvalidTypesException;
import org.apache.flink.api.common.functions.JoinFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.MapPartitionFunction;
@@ -35,13 +36,17 @@ import org.apache.flink.api.java.tuple.Tuple1;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.typeutils.MissingTypeInfo;
import org.apache.flink.api.java.typeutils.TupleTypeInfo;
+import org.apache.flink.api.java.typeutils.TypeExtractionUtils;
import org.apache.flink.api.java.typeutils.TypeExtractor;
import org.apache.flink.api.java.typeutils.TypeInfoParser;
import org.junit.Assert;
import org.junit.Test;
+import java.lang.reflect.Method;
+
import static org.apache.flink.api.java.typeutils.TypeExtractionUtils.checkAndExtractLambda;
+import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.fail;
@@ -322,4 +327,50 @@ public class LambdaExtractionTest {
Assert.assertEquals(BasicTypeInfo.INT_TYPE_INFO, ti);
}
+ public interface InterfaceWithDefaultMethod {
+ void samMethod();
+
+ default void defaultMethod() {
+
+ }
+ }
+
+ @Test
+ public void testSamMethodExtractionInterfaceWithDefaultMethod() {
+ final Method sam = TypeExtractionUtils.getSingleAbstractMethod(InterfaceWithDefaultMethod.class);
+ assertNotNull(sam);
+ assertEquals("samMethod", sam.getName());
+ }
+
+ public interface InterfaceWithMultipleMethods {
+ void firstMethod();
+
+ void secondMethod();
+ }
+
+ @Test(expected = InvalidTypesException.class)
+ public void getSingleAbstractMethodMultipleMethods() throws Exception {
+ TypeExtractionUtils.getSingleAbstractMethod(InterfaceWithMultipleMethods.class);
+ }
+
+ public interface InterfaceWithoutAbstractMethod {
+ default void defaultMethod() {
+
+ };
+ }
+
+ @Test(expected = InvalidTypesException.class)
+ public void getSingleAbstractMethodNoAbstractMethods() throws Exception {
+ TypeExtractionUtils.getSingleAbstractMethod(InterfaceWithoutAbstractMethod.class);
+ }
+
+ public abstract class AbstractClassWithSingleAbstractMethod {
+ public abstract void defaultMethod();
+ }
+
+ @Test(expected = InvalidTypesException.class)
+ public void getSingleAbstractMethodNotAnInterface() throws Exception {
+ TypeExtractionUtils.getSingleAbstractMethod(AbstractClassWithSingleAbstractMethod.class);
+ }
+
}
[12/21] flink git commit: [FLINK-6863] Remove batch dependency from
streaming-examples
Posted by ch...@apache.org.
[FLINK-6863] Remove batch dependency from streaming-examples
This closes #4115.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/387dc4b9
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/387dc4b9
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/387dc4b9
Branch: refs/heads/master
Commit: 387dc4b992fc0ea3b728ee343c3a052bc746e3a3
Parents: d7bf7ee
Author: zentol <ch...@apache.org>
Authored: Wed Jun 7 15:59:38 2017 +0200
Committer: zentol <ch...@apache.org>
Committed: Fri Jun 23 14:14:29 2017 +0200
----------------------------------------------------------------------
flink-examples/flink-examples-streaming/pom.xml | 6 --
.../examples/sideoutput/SideOutputExample.java | 4 +-
.../examples/windowing/WindowWordCount.java | 4 +-
.../examples/wordcount/PojoExample.java | 2 +-
.../streaming/examples/wordcount/WordCount.java | 2 +-
.../examples/wordcount/util/WordCountData.java | 65 ++++++++++++++++++++
6 files changed, 71 insertions(+), 12 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/387dc4b9/flink-examples/flink-examples-streaming/pom.xml
----------------------------------------------------------------------
diff --git a/flink-examples/flink-examples-streaming/pom.xml b/flink-examples/flink-examples-streaming/pom.xml
index b52116b..3c5c971 100644
--- a/flink-examples/flink-examples-streaming/pom.xml
+++ b/flink-examples/flink-examples-streaming/pom.xml
@@ -52,12 +52,6 @@ under the License.
<dependency>
<groupId>org.apache.flink</groupId>
- <artifactId>flink-examples-batch_${scala.binary.version}</artifactId>
- <version>${project.version}</version>
- </dependency>
-
- <dependency>
- <groupId>org.apache.flink</groupId>
<artifactId>flink-connector-twitter_${scala.binary.version}</artifactId>
<version>${project.version}</version>
</dependency>
http://git-wip-us.apache.org/repos/asf/flink/blob/387dc4b9/flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/sideoutput/SideOutputExample.java
----------------------------------------------------------------------
diff --git a/flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/sideoutput/SideOutputExample.java b/flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/sideoutput/SideOutputExample.java
index a506624..0cb8930 100644
--- a/flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/sideoutput/SideOutputExample.java
+++ b/flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/sideoutput/SideOutputExample.java
@@ -21,7 +21,6 @@ import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.utils.ParameterTool;
-import org.apache.flink.examples.java.wordcount.util.WordCountData;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
@@ -29,13 +28,14 @@ import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
+import org.apache.flink.streaming.examples.wordcount.util.WordCountData;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
/**
* An example that illustrates the use of side outputs.
*
- * <p>This is a modified version of {@link org.apache.flink.streaming.examples.windowing.WindowWordCount}
+ * <p>This is a modified version of {@link WindowWordCount}
* that has a filter in the tokenizer and only emits some words for counting
* while emitting the other words to a side output.
*/
http://git-wip-us.apache.org/repos/asf/flink/blob/387dc4b9/flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/windowing/WindowWordCount.java
----------------------------------------------------------------------
diff --git a/flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/windowing/WindowWordCount.java b/flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/windowing/WindowWordCount.java
index ab64575..9b747f0 100644
--- a/flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/windowing/WindowWordCount.java
+++ b/flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/windowing/WindowWordCount.java
@@ -19,10 +19,10 @@ package org.apache.flink.streaming.examples.windowing;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.utils.ParameterTool;
-import org.apache.flink.examples.java.wordcount.util.WordCountData;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.examples.wordcount.WordCount;
+import org.apache.flink.streaming.examples.wordcount.util.WordCountData;
/**
* Implements a windowed version of the streaming "WordCount" program.
@@ -31,7 +31,7 @@ import org.apache.flink.streaming.examples.wordcount.WordCount;
*
* <p>Usage: <code>WordCount --input <path> --output <path> --window <n> --slide <n></code><br>
* If no parameters are provided, the program is run with default data from
- * {@link org.apache.flink.examples.java.wordcount.util.WordCountData}.
+ * {@link WordCountData}.
*
* <p>This example shows how to:
* <ul>
http://git-wip-us.apache.org/repos/asf/flink/blob/387dc4b9/flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/wordcount/PojoExample.java
----------------------------------------------------------------------
diff --git a/flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/wordcount/PojoExample.java b/flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/wordcount/PojoExample.java
index 67f75fd..a8ece16 100644
--- a/flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/wordcount/PojoExample.java
+++ b/flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/wordcount/PojoExample.java
@@ -20,9 +20,9 @@ package org.apache.flink.streaming.examples.wordcount;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.utils.ParameterTool;
-import org.apache.flink.examples.java.wordcount.util.WordCountData;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.examples.wordcount.util.WordCountData;
import org.apache.flink.util.Collector;
/**
http://git-wip-us.apache.org/repos/asf/flink/blob/387dc4b9/flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/wordcount/WordCount.java
----------------------------------------------------------------------
diff --git a/flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/wordcount/WordCount.java b/flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/wordcount/WordCount.java
index b839bd7..9c7c435 100644
--- a/flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/wordcount/WordCount.java
+++ b/flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/wordcount/WordCount.java
@@ -20,9 +20,9 @@ package org.apache.flink.streaming.examples.wordcount;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.utils.ParameterTool;
-import org.apache.flink.examples.java.wordcount.util.WordCountData;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.examples.wordcount.util.WordCountData;
import org.apache.flink.util.Collector;
/**
http://git-wip-us.apache.org/repos/asf/flink/blob/387dc4b9/flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/wordcount/util/WordCountData.java
----------------------------------------------------------------------
diff --git a/flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/wordcount/util/WordCountData.java b/flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/wordcount/util/WordCountData.java
new file mode 100644
index 0000000..30b412e
--- /dev/null
+++ b/flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/wordcount/util/WordCountData.java
@@ -0,0 +1,65 @@
+/*
+ * 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.streaming.examples.wordcount.util;
+
+/**
+ * Provides the default data sets used for the WordCount example program.
+ * The default data sets are used, if no parameters are given to the program.
+ *
+ */
+public class WordCountData {
+
+ public static final String[] WORDS = new String[] {
+ "To be, or not to be,--that is the question:--",
+ "Whether 'tis nobler in the mind to suffer",
+ "The slings and arrows of outrageous fortune",
+ "Or to take arms against a sea of troubles,",
+ "And by opposing end them?--To die,--to sleep,--",
+ "No more; and by a sleep to say we end",
+ "The heartache, and the thousand natural shocks",
+ "That flesh is heir to,--'tis a consummation",
+ "Devoutly to be wish'd. To die,--to sleep;--",
+ "To sleep! perchance to dream:--ay, there's the rub;",
+ "For in that sleep of death what dreams may come,",
+ "When we have shuffled off this mortal coil,",
+ "Must give us pause: there's the respect",
+ "That makes calamity of so long life;",
+ "For who would bear the whips and scorns of time,",
+ "The oppressor's wrong, the proud man's contumely,",
+ "The pangs of despis'd love, the law's delay,",
+ "The insolence of office, and the spurns",
+ "That patient merit of the unworthy takes,",
+ "When he himself might his quietus make",
+ "With a bare bodkin? who would these fardels bear,",
+ "To grunt and sweat under a weary life,",
+ "But that the dread of something after death,--",
+ "The undiscover'd country, from whose bourn",
+ "No traveller returns,--puzzles the will,",
+ "And makes us rather bear those ills we have",
+ "Than fly to others that we know not of?",
+ "Thus conscience does make cowards of us all;",
+ "And thus the native hue of resolution",
+ "Is sicklied o'er with the pale cast of thought;",
+ "And enterprises of great pith and moment,",
+ "With this regard, their currents turn awry,",
+ "And lose the name of action.--Soft you now!",
+ "The fair Ophelia!--Nymph, in thy orisons",
+ "Be all my sins remember'd."
+ };
+}
[11/21] flink git commit: [FLINK-6769] Replace usage of deprecated
FileSystem#create(Path, boolean)
Posted by ch...@apache.org.
[FLINK-6769] Replace usage of deprecated FileSystem#create(Path, boolean)
This closes #4116.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/d7bf7eeb
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/d7bf7eeb
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/d7bf7eeb
Branch: refs/heads/master
Commit: d7bf7eebed29f35c797bd3e21bafd0eb071f9f34
Parents: 3424924
Author: zentol <ch...@apache.org>
Authored: Tue May 30 09:23:05 2017 +0200
Committer: zentol <ch...@apache.org>
Committed: Fri Jun 23 14:14:29 2017 +0200
----------------------------------------------------------------------
.../java/org/apache/flink/core/fs/InitOutputPathTest.java | 2 +-
.../apache/flink/core/fs/SafetyNetCloseableRegistryTest.java | 4 ++--
.../java/org/apache/flink/examples/java/distcp/DistCp.java | 2 +-
.../src/test/java/org/apache/flink/hdfstests/HDFSTest.java | 2 +-
.../org/apache/flink/runtime/blob/FileSystemBlobStore.java | 2 +-
.../java/org/apache/flink/runtime/filecache/FileCache.java | 2 +-
.../runtime/state/filesystem/FsCheckpointStreamFactory.java | 2 +-
.../zookeeper/filesystem/FileSystemStateStorageHelper.java | 2 +-
.../runtime/checkpoint/savepoint/MigrationV0ToV1Test.java | 2 +-
.../runtime/checkpoint/savepoint/SavepointStoreTest.java | 2 +-
.../state/filesystem/FsCheckpointStateOutputStreamTest.java | 8 ++++----
11 files changed, 15 insertions(+), 15 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/d7bf7eeb/flink-core/src/test/java/org/apache/flink/core/fs/InitOutputPathTest.java
----------------------------------------------------------------------
diff --git a/flink-core/src/test/java/org/apache/flink/core/fs/InitOutputPathTest.java b/flink-core/src/test/java/org/apache/flink/core/fs/InitOutputPathTest.java
index c332324..9b67388 100644
--- a/flink-core/src/test/java/org/apache/flink/core/fs/InitOutputPathTest.java
+++ b/flink-core/src/test/java/org/apache/flink/core/fs/InitOutputPathTest.java
@@ -191,7 +191,7 @@ public class InitOutputPathTest {
@Override
public void go() throws Exception {
fs.initOutPathLocalFS(path.getParent(), WriteMode.OVERWRITE, true);
- try (FSDataOutputStream out = fs.create(path, true)) {
+ try (FSDataOutputStream out = fs.create(path, WriteMode.OVERWRITE)) {
out.write(11);
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/d7bf7eeb/flink-core/src/test/java/org/apache/flink/core/fs/SafetyNetCloseableRegistryTest.java
----------------------------------------------------------------------
diff --git a/flink-core/src/test/java/org/apache/flink/core/fs/SafetyNetCloseableRegistryTest.java b/flink-core/src/test/java/org/apache/flink/core/fs/SafetyNetCloseableRegistryTest.java
index 7973c69..05ee894 100644
--- a/flink-core/src/test/java/org/apache/flink/core/fs/SafetyNetCloseableRegistryTest.java
+++ b/flink-core/src/test/java/org/apache/flink/core/fs/SafetyNetCloseableRegistryTest.java
@@ -84,7 +84,7 @@ public class SafetyNetCloseableRegistryTest {
Path tmp = new Path(tmpFolder.newFolder().toURI().toString(), "test_file");
- try (FSDataOutputStream stream = fs1.create(tmp, false)) {
+ try (FSDataOutputStream stream = fs1.create(tmp, FileSystem.WriteMode.NO_OVERWRITE)) {
CheckedThread t2 = new CheckedThread() {
@Override
public void go() {
@@ -285,4 +285,4 @@ public class SafetyNetCloseableRegistryTest {
}
}
}
-}
\ No newline at end of file
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/d7bf7eeb/flink-examples/flink-examples-batch/src/main/java/org/apache/flink/examples/java/distcp/DistCp.java
----------------------------------------------------------------------
diff --git a/flink-examples/flink-examples-batch/src/main/java/org/apache/flink/examples/java/distcp/DistCp.java b/flink-examples/flink-examples-batch/src/main/java/org/apache/flink/examples/java/distcp/DistCp.java
index a358490..3e66241 100644
--- a/flink-examples/flink-examples-batch/src/main/java/org/apache/flink/examples/java/distcp/DistCp.java
+++ b/flink-examples/flink-examples-batch/src/main/java/org/apache/flink/examples/java/distcp/DistCp.java
@@ -129,7 +129,7 @@ public class DistCp {
FSDataOutputStream outputStream = null;
FSDataInputStream inputStream = null;
try {
- outputStream = targetFs.create(outPath, true);
+ outputStream = targetFs.create(outPath, FileSystem.WriteMode.OVERWRITE);
inputStream = task.getPath().getFileSystem().open(task.getPath());
int bytes = IOUtils.copy(inputStream, outputStream);
bytesCounter.add(bytes);
http://git-wip-us.apache.org/repos/asf/flink/blob/d7bf7eeb/flink-fs-tests/src/test/java/org/apache/flink/hdfstests/HDFSTest.java
----------------------------------------------------------------------
diff --git a/flink-fs-tests/src/test/java/org/apache/flink/hdfstests/HDFSTest.java b/flink-fs-tests/src/test/java/org/apache/flink/hdfstests/HDFSTest.java
index 5f778f2..950f20f 100644
--- a/flink-fs-tests/src/test/java/org/apache/flink/hdfstests/HDFSTest.java
+++ b/flink-fs-tests/src/test/java/org/apache/flink/hdfstests/HDFSTest.java
@@ -199,7 +199,7 @@ public class HDFSTest {
byte[] data = "HDFSTest#testDeletePathIfEmpty".getBytes(ConfigConstants.DEFAULT_CHARSET);
for (Path file: Arrays.asList(singleFile, directoryFile)) {
- org.apache.flink.core.fs.FSDataOutputStream outputStream = fs.create(file, true);
+ org.apache.flink.core.fs.FSDataOutputStream outputStream = fs.create(file, FileSystem.WriteMode.OVERWRITE);
outputStream.write(data);
outputStream.close();
}
http://git-wip-us.apache.org/repos/asf/flink/blob/d7bf7eeb/flink-runtime/src/main/java/org/apache/flink/runtime/blob/FileSystemBlobStore.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/blob/FileSystemBlobStore.java b/flink-runtime/src/main/java/org/apache/flink/runtime/blob/FileSystemBlobStore.java
index b54756c..f3607d8 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/blob/FileSystemBlobStore.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/blob/FileSystemBlobStore.java
@@ -74,7 +74,7 @@ public class FileSystemBlobStore implements BlobStoreService {
}
private void put(File fromFile, String toBlobPath) throws IOException {
- try (OutputStream os = fileSystem.create(new Path(toBlobPath), true)) {
+ try (OutputStream os = fileSystem.create(new Path(toBlobPath), FileSystem.WriteMode.OVERWRITE)) {
LOG.debug("Copying from {} to {}.", fromFile, toBlobPath);
Files.copy(fromFile, os);
}
http://git-wip-us.apache.org/repos/asf/flink/blob/d7bf7eeb/flink-runtime/src/main/java/org/apache/flink/runtime/filecache/FileCache.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/filecache/FileCache.java b/flink-runtime/src/main/java/org/apache/flink/runtime/filecache/FileCache.java
index 84b8feb..d78801d 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/filecache/FileCache.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/filecache/FileCache.java
@@ -252,7 +252,7 @@ public class FileCache {
copy(content.getPath(), new Path(localPath), executable);
}
} else {
- try (FSDataOutputStream lfsOutput = tFS.create(targetPath, false); FSDataInputStream fsInput = sFS.open(sourcePath)) {
+ try (FSDataOutputStream lfsOutput = tFS.create(targetPath, FileSystem.WriteMode.NO_OVERWRITE); FSDataInputStream fsInput = sFS.open(sourcePath)) {
IOUtils.copyBytes(fsInput, lfsOutput);
//noinspection ResultOfMethodCallIgnored
new File(targetPath.toString()).setExecutable(executable);
http://git-wip-us.apache.org/repos/asf/flink/blob/d7bf7eeb/flink-runtime/src/main/java/org/apache/flink/runtime/state/filesystem/FsCheckpointStreamFactory.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/filesystem/FsCheckpointStreamFactory.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/filesystem/FsCheckpointStreamFactory.java
index 8455d84..a824651 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/filesystem/FsCheckpointStreamFactory.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/filesystem/FsCheckpointStreamFactory.java
@@ -359,7 +359,7 @@ public class FsCheckpointStreamFactory implements CheckpointStreamFactory {
for (int attempt = 0; attempt < 10; attempt++) {
try {
statePath = createStatePath();
- outStream = fs.create(statePath, false);
+ outStream = fs.create(statePath, FileSystem.WriteMode.NO_OVERWRITE);
break;
}
catch (Exception e) {
http://git-wip-us.apache.org/repos/asf/flink/blob/d7bf7eeb/flink-runtime/src/main/java/org/apache/flink/runtime/zookeeper/filesystem/FileSystemStateStorageHelper.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/zookeeper/filesystem/FileSystemStateStorageHelper.java b/flink-runtime/src/main/java/org/apache/flink/runtime/zookeeper/filesystem/FileSystemStateStorageHelper.java
index 7658afb..fcad617 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/zookeeper/filesystem/FileSystemStateStorageHelper.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/zookeeper/filesystem/FileSystemStateStorageHelper.java
@@ -62,7 +62,7 @@ public class FileSystemStateStorageHelper<T extends Serializable> implements Ret
for (int attempt = 0; attempt < 10; attempt++) {
Path filePath = getNewFilePath();
- try (FSDataOutputStream outStream = fs.create(filePath, false)) {
+ try (FSDataOutputStream outStream = fs.create(filePath, FileSystem.WriteMode.NO_OVERWRITE)) {
InstantiationUtil.serializeObject(outStream, state);
return new RetrievableStreamStateHandle<T>(filePath, outStream.getPos());
}
http://git-wip-us.apache.org/repos/asf/flink/blob/d7bf7eeb/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/savepoint/MigrationV0ToV1Test.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/savepoint/MigrationV0ToV1Test.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/savepoint/MigrationV0ToV1Test.java
index 1ecb2e3..16f3769 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/savepoint/MigrationV0ToV1Test.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/savepoint/MigrationV0ToV1Test.java
@@ -109,7 +109,7 @@ public class MigrationV0ToV1Test {
}
try {
- fdos = fs.create(path, false);
+ fdos = fs.create(path, FileSystem.WriteMode.NO_OVERWRITE);
break;
} catch (Exception e) {
latestException = e;
http://git-wip-us.apache.org/repos/asf/flink/blob/d7bf7eeb/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/savepoint/SavepointStoreTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/savepoint/SavepointStoreTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/savepoint/SavepointStoreTest.java
index 391102c..0444936 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/savepoint/SavepointStoreTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/savepoint/SavepointStoreTest.java
@@ -104,7 +104,7 @@ public class SavepointStoreTest {
public void testUnexpectedSavepoint() throws Exception {
// Random file
Path filePath = new Path(tmp.getRoot().getPath(), UUID.randomUUID().toString());
- FSDataOutputStream fdos = FileSystem.get(filePath.toUri()).create(filePath, false);
+ FSDataOutputStream fdos = FileSystem.get(filePath.toUri()).create(filePath, FileSystem.WriteMode.NO_OVERWRITE);
DataOutputStream dos = new DataOutputStream(fdos);
for (int i = 0; i < 10; i++) {
dos.writeLong(ThreadLocalRandom.current().nextLong());
http://git-wip-us.apache.org/repos/asf/flink/blob/d7bf7eeb/flink-runtime/src/test/java/org/apache/flink/runtime/state/filesystem/FsCheckpointStateOutputStreamTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/state/filesystem/FsCheckpointStateOutputStreamTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/state/filesystem/FsCheckpointStateOutputStreamTest.java
index 980075d..95d00a9 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/state/filesystem/FsCheckpointStateOutputStreamTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/state/filesystem/FsCheckpointStateOutputStreamTest.java
@@ -130,7 +130,7 @@ public class FsCheckpointStateOutputStreamTest {
final ArgumentCaptor<Path> pathCaptor = ArgumentCaptor.forClass(Path.class);
- when(fs.create(pathCaptor.capture(), anyBoolean())).thenReturn(outputStream);
+ when(fs.create(pathCaptor.capture(), any(FileSystem.WriteMode.class))).thenReturn(outputStream);
CheckpointStreamFactory.CheckpointStateOutputStream stream = new FsCheckpointStreamFactory.FsCheckpointStateOutputStream(
TEMP_DIR_PATH,
@@ -141,7 +141,7 @@ public class FsCheckpointStateOutputStreamTest {
// this should create the underlying file stream
stream.write(new byte[]{1,2,3,4,5});
- verify(fs).create(any(Path.class), anyBoolean());
+ verify(fs).create(any(Path.class), any(FileSystem.WriteMode.class));
stream.close();
@@ -158,7 +158,7 @@ public class FsCheckpointStateOutputStreamTest {
final ArgumentCaptor<Path> pathCaptor = ArgumentCaptor.forClass(Path.class);
- when(fs.create(pathCaptor.capture(), anyBoolean())).thenReturn(outputStream);
+ when(fs.create(pathCaptor.capture(), any(FileSystem.WriteMode.class))).thenReturn(outputStream);
doThrow(new IOException("Test IOException.")).when(outputStream).close();
CheckpointStreamFactory.CheckpointStateOutputStream stream = new FsCheckpointStreamFactory.FsCheckpointStateOutputStream(
@@ -170,7 +170,7 @@ public class FsCheckpointStateOutputStreamTest {
// this should create the underlying file stream
stream.write(new byte[]{1,2,3,4,5});
- verify(fs).create(any(Path.class), anyBoolean());
+ verify(fs).create(any(Path.class), any(FileSystem.WriteMode.class));
try {
stream.closeAndGetHandle();
[10/21] flink git commit: [FLINK-6868][build] Using
`scala.binary.version` for `flink-streaming-scala` in `Cassandra Connector`
Posted by ch...@apache.org.
[FLINK-6868][build] Using `scala.binary.version` for `flink-streaming-scala` in `Cassandra Connector`
This closes #4087.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/b2062eee
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/b2062eee
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/b2062eee
Branch: refs/heads/master
Commit: b2062eeeaf68b37aa91e71b65b5201df04dab992
Parents: 387dc4b
Author: asdf2014 <15...@qq.com>
Authored: Thu Jun 8 14:55:25 2017 +0800
Committer: zentol <ch...@apache.org>
Committed: Fri Jun 23 14:14:29 2017 +0200
----------------------------------------------------------------------
flink-connectors/flink-connector-cassandra/pom.xml | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/b2062eee/flink-connectors/flink-connector-cassandra/pom.xml
----------------------------------------------------------------------
diff --git a/flink-connectors/flink-connector-cassandra/pom.xml b/flink-connectors/flink-connector-cassandra/pom.xml
index 4a720c4..f006f3a 100644
--- a/flink-connectors/flink-connector-cassandra/pom.xml
+++ b/flink-connectors/flink-connector-cassandra/pom.xml
@@ -100,7 +100,7 @@ under the License.
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
- <artifactId>flink-streaming-scala_2.10</artifactId>
+ <artifactId>flink-streaming-scala_${scala.binary.version}</artifactId>
<version>${project.version}</version>
<scope>provided</scope>
</dependency>