diff --git a/nephele/nephele-common/src/main/java/eu/stratosphere/nephele/io/channels/bytebuffered/AbstractByteBufferedInputChannel.java b/nephele/nephele-common/src/main/java/eu/stratosphere/nephele/io/channels/bytebuffered/AbstractByteBufferedInputChannel.java index 786dc5dcea816..9f3fc6db9cad7 100644 --- a/nephele/nephele-common/src/main/java/eu/stratosphere/nephele/io/channels/bytebuffered/AbstractByteBufferedInputChannel.java +++ b/nephele/nephele-common/src/main/java/eu/stratosphere/nephele/io/channels/bytebuffered/AbstractByteBufferedInputChannel.java @@ -209,6 +209,23 @@ public void close() throws IOException, InterruptedException { releasedConsumedReadBuffer(); } + // This code fragment makes sure the isClosed method works in case the channel input has not been fully consumed + if (this.getType() == ChannelType.NETWORK) { + synchronized (this.synchronisationObject) { + if (!this.brokerAggreedToCloseChannel) { + while (!this.brokerAggreedToCloseChannel) { + + requestReadBuffersFromBroker(); + if (this.uncompressedDataBuffer != null || this.compressedDataBuffer != null) { + releasedConsumedReadBuffer(); + } + this.synchronisationObject.wait(500); + } + this.bufferedRecord = null; + } + } + } + /* * Send close event to indicate the input channel has successfully * processed all data it is interested in. diff --git a/nephele/nephele-common/src/main/java/eu/stratosphere/nephele/jobgraph/JobGenericInputVertex.java b/nephele/nephele-common/src/main/java/eu/stratosphere/nephele/jobgraph/JobGenericInputVertex.java index d7f2db2b0a4dc..e39d90d8a2f39 100644 --- a/nephele/nephele-common/src/main/java/eu/stratosphere/nephele/jobgraph/JobGenericInputVertex.java +++ b/nephele/nephele-common/src/main/java/eu/stratosphere/nephele/jobgraph/JobGenericInputVertex.java @@ -26,10 +26,6 @@ import eu.stratosphere.nephele.types.StringRecord; import eu.stratosphere.nephele.util.StringUtils; - -/** - * - */ public class JobGenericInputVertex extends JobInputVertex { /** diff --git a/nephele/nephele-ec2cloudmanager/src/main/java/eu/stratosphere/nephele/instance/cloud/CloudInstanceNotifier.java b/nephele/nephele-ec2cloudmanager/src/main/java/eu/stratosphere/nephele/instance/cloud/CloudInstanceNotifier.java index f2290b1ae5c20..f0fa691c5001f 100644 --- a/nephele/nephele-ec2cloudmanager/src/main/java/eu/stratosphere/nephele/instance/cloud/CloudInstanceNotifier.java +++ b/nephele/nephele-ec2cloudmanager/src/main/java/eu/stratosphere/nephele/instance/cloud/CloudInstanceNotifier.java @@ -18,7 +18,6 @@ import java.util.List; import eu.stratosphere.nephele.instance.AbstractInstance; -import eu.stratosphere.nephele.instance.AllocatedResource; import eu.stratosphere.nephele.instance.InstanceListener; import eu.stratosphere.nephele.jobgraph.JobID; diff --git a/nephele/nephele-examples/src/main/java/eu/stratosphere/nephele/example/grep/Grep.java b/nephele/nephele-examples/src/main/java/eu/stratosphere/nephele/example/grep/Grep.java index a49860d114e62..7702ca8999612 100644 --- a/nephele/nephele-examples/src/main/java/eu/stratosphere/nephele/example/grep/Grep.java +++ b/nephele/nephele-examples/src/main/java/eu/stratosphere/nephele/example/grep/Grep.java @@ -20,8 +20,6 @@ import java.net.InetSocketAddress; import eu.stratosphere.nephele.client.JobClient; -import eu.stratosphere.nephele.client.JobSubmissionResult; -import eu.stratosphere.nephele.configuration.ConfigConstants; import eu.stratosphere.nephele.configuration.Configuration; import eu.stratosphere.nephele.fs.Path; import eu.stratosphere.nephele.io.channels.ChannelType; @@ -66,7 +64,7 @@ public static void main(String[] args) { } // Create jar file and attach it - final File jarFile = new File("/tmp/broadcastJob.jar"); + final File jarFile = new File("/tmp/grepJob.jar"); final JarFileCreator jarFileCreator = new JarFileCreator(jarFile); jarFileCreator.addClass(GrepTask.class); diff --git a/nephele/nephele-examples/src/main/java/eu/stratosphere/nephele/example/grep/GrepJob.java b/nephele/nephele-examples/src/main/java/eu/stratosphere/nephele/example/grep/GrepJob.java deleted file mode 100644 index c330cea2c7df2..0000000000000 --- a/nephele/nephele-examples/src/main/java/eu/stratosphere/nephele/example/grep/GrepJob.java +++ /dev/null @@ -1,68 +0,0 @@ -/*********************************************************************************************************************** - * - * Copyright (C) 2010 by the Stratosphere project (http://stratosphere.eu) - * - * Licensed 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 eu.stratosphere.nephele.example.grep; - -import eu.stratosphere.nephele.fs.Path; -import eu.stratosphere.nephele.io.channels.ChannelType; -import eu.stratosphere.nephele.io.compression.CompressionLevel; -import eu.stratosphere.nephele.io.library.FileLineReader; -import eu.stratosphere.nephele.io.library.FileLineWriter; -import eu.stratosphere.nephele.jobgraph.JobFileInputVertex; -import eu.stratosphere.nephele.jobgraph.JobFileOutputVertex; -import eu.stratosphere.nephele.jobgraph.JobGraph; -import eu.stratosphere.nephele.jobgraph.JobGraphDefinitionException; -import eu.stratosphere.nephele.jobgraph.JobTaskVertex; - -public class GrepJob implements Job { - - JobGraph jobGraph = null; - - public GrepJob() { - - jobGraph = new JobGraph("Grep Example Job"); - - JobFileInputVertex input = new JobFileInputVertex("Input 1", jobGraph); - input.setFileInputClass(FileLineReader.class); - input.setFilePath(new Path("file:///home/ec2-user/text.txt")); - - JobTaskVertex task1 = new JobTaskVertex("Task 1", jobGraph); - task1.setTaskClass(GrepTask.class); - - JobTaskVertex task2 = new JobTaskVertex("Task 2", jobGraph); - task2.setTaskClass(GrepTask.class); - - JobFileOutputVertex output = new JobFileOutputVertex("Output 1", jobGraph); - output.setFileOutputClass(FileLineWriter.class); - output.setFilePath(new Path("file:///tmp/")); - - task1.setNumberOfSubtasks(2); - - try { - - input.connectTo(task1, ChannelType.NETWORK, CompressionLevel.NO_COMPRESSION); - task1.connectTo(task2, ChannelType.NETWORK, CompressionLevel.NO_COMPRESSION); - task2.connectTo(output, ChannelType.NETWORK, CompressionLevel.NO_COMPRESSION); - - } catch (JobGraphDefinitionException e) { - e.printStackTrace(); - } - - } - - public JobGraph getJobGraph() { - return jobGraph; - } -} diff --git a/nephele/nephele-examples/src/main/java/eu/stratosphere/nephele/example/grep/Job.java b/nephele/nephele-examples/src/main/java/eu/stratosphere/nephele/example/grep/Job.java deleted file mode 100644 index 4f5fcfd19da95..0000000000000 --- a/nephele/nephele-examples/src/main/java/eu/stratosphere/nephele/example/grep/Job.java +++ /dev/null @@ -1,23 +0,0 @@ -/*********************************************************************************************************************** - * - * Copyright (C) 2010 by the Stratosphere project (http://stratosphere.eu) - * - * Licensed 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 eu.stratosphere.nephele.example.grep; - -import eu.stratosphere.nephele.jobgraph.JobGraph; - -public interface Job { - - public JobGraph getJobGraph(); -} diff --git a/nephele/nephele-examples/src/main/java/eu/stratosphere/nephele/example/grep/JobLauncher.java b/nephele/nephele-examples/src/main/java/eu/stratosphere/nephele/example/grep/JobLauncher.java deleted file mode 100644 index 0f71731c6ea5b..0000000000000 --- a/nephele/nephele-examples/src/main/java/eu/stratosphere/nephele/example/grep/JobLauncher.java +++ /dev/null @@ -1,144 +0,0 @@ -/*********************************************************************************************************************** - * - * Copyright (C) 2010 by the Stratosphere project (http://stratosphere.eu) - * - * Licensed 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 eu.stratosphere.nephele.example.grep; - -import java.io.IOException; -import java.util.ArrayList; -import java.util.Arrays; -import java.util.List; - -import javax.xml.parsers.DocumentBuilder; -import javax.xml.parsers.DocumentBuilderFactory; -import javax.xml.parsers.ParserConfigurationException; - -import org.w3c.dom.Document; -import org.w3c.dom.Element; -import org.w3c.dom.Node; -import org.w3c.dom.NodeList; -import org.xml.sax.SAXException; - -import eu.stratosphere.nephele.client.JobClient; -import eu.stratosphere.nephele.client.JobExecutionException; -import eu.stratosphere.nephele.configuration.Configuration; -import eu.stratosphere.nephele.fs.Path; -import eu.stratosphere.nephele.jobgraph.JobGraph; - -public class JobLauncher { - - public static void main(String[] args) { - - List pars = Arrays.asList(args); - - int lengthPars = pars.size(); - int posJar = pars.indexOf("-jar"); - int posConf = pars.indexOf("-conf"); - - String jobName = ""; - List jars = new ArrayList(); - List confs = new ArrayList(); - - if (posJar == -1 && posConf == -1) { - if (lengthPars != 1) { - System.err.println("Usage: JobLauncher JobName -jar -conf "); - System.exit(1); - } else { - jobName = pars.get(0); - } - } else if (posJar != -1 && posConf == -1) { - if (posJar != 1 || lengthPars < 3) { - System.err.println("Usage: JobLauncher JobName -jar -conf "); - System.exit(1); - } else { - jobName = pars.get(0); - jars = pars.subList(2, pars.size()); - } - } else if (posJar == -1 && posConf != -1) { - if (posConf != 1 || lengthPars < 3) { - System.err.println("Usage: JobLauncher JobName -jar -conf "); - System.exit(1); - } else { - jobName = pars.get(0); - confs = pars.subList(posConf + 1, pars.size()); - } - } else { - if (pars.size() < 5 || posJar != 1 || posConf < 3 || posConf == pars.size() - 1) { - System.err.println("Usage: JobLauncher JobName -jar -conf "); - System.exit(1); - } else { - jobName = pars.get(0); - jars = pars.subList(2, posConf); - confs = pars.subList(posConf + 1, pars.size()); - } - } - - try { - - Job job = (Job) Class.forName(jobName).newInstance(); - JobGraph jobGraph = job.getJobGraph(); - - for (int i = 0; i < jars.size(); i++) { - jobGraph.addJar(new Path(jars.get(i))); - } - - Configuration jobConfiguration = jobGraph.getJobConfiguration(); - Configuration clientConfiguration = new Configuration(); - - DocumentBuilderFactory dbf = DocumentBuilderFactory.newInstance(); - - DocumentBuilder db = dbf.newDocumentBuilder(); - - for (int i = 0; i < confs.size(); i++) { - - Document doc = db.parse(confs.get(i)); - NodeList nl = doc.getElementsByTagName("property"); - - for (int j = 0; j < nl.getLength(); j++) { - - Element property = (Element) nl.item(j); - Node nodeKey = property.getElementsByTagName("key").item(0); - Node nodeValue = property.getElementsByTagName("value").item(0); - String key = nodeKey.getFirstChild().getNodeValue(); - String value = nodeValue.getFirstChild().getNodeValue(); - - if (key.startsWith("job.")) { - jobConfiguration.setString(key, value); - } else { - clientConfiguration.setString(key, value); - } - } - } - - JobClient jobClient = new JobClient(jobGraph, clientConfiguration); - jobClient.submitJobAndWait(); - - } catch (InstantiationException e) { - e.printStackTrace(); - } catch (IllegalAccessException e) { - e.printStackTrace(); - } catch (ClassNotFoundException e) { - e.printStackTrace(); - } catch (ParserConfigurationException e) { - e.printStackTrace(); - } catch (SAXException e) { - e.printStackTrace(); - } catch (IOException e) { - e.printStackTrace(); - } catch (JobExecutionException e) { - e.printStackTrace(); - } - - } -} diff --git a/nephele/nephele-queuescheduler/src/test/java/eu/stratosphere/nephele/jobmanager/scheduler/queue/TestDeploymentManager.java b/nephele/nephele-queuescheduler/src/test/java/eu/stratosphere/nephele/jobmanager/scheduler/queue/TestDeploymentManager.java index e9804ec0efecb..adc1db50518e0 100644 --- a/nephele/nephele-queuescheduler/src/test/java/eu/stratosphere/nephele/jobmanager/scheduler/queue/TestDeploymentManager.java +++ b/nephele/nephele-queuescheduler/src/test/java/eu/stratosphere/nephele/jobmanager/scheduler/queue/TestDeploymentManager.java @@ -1,12 +1,17 @@ -/** - * This abstract scheduler must be extended by a scheduler implementations for Nephele. The abstract class defines the - * fundamental methods for scheduling and removing jobs. While Nephele's - * {@link eu.stratosphere.nephele.jobmanager.JobManager} is responsible for requesting the required instances for the - * job at the {@link eu.stratosphere.nephele.instance.InstanceManager}, the scheduler is in charge of assigning the - * individual tasks to the instances. - * - * @author warneke - */ +/*********************************************************************************************************************** + * + * Copyright (C) 2010 by the Stratosphere project (http://stratosphere.eu) + * + * Licensed 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 eu.stratosphere.nephele.jobmanager.scheduler.queue; diff --git a/nephele/nephele-server/src/main/java/eu/stratosphere/nephele/jobmanager/splitassigner/InputSplitManager.java b/nephele/nephele-server/src/main/java/eu/stratosphere/nephele/jobmanager/splitassigner/InputSplitManager.java index 8ec79dd4e58e0..c7a9944838a0d 100644 --- a/nephele/nephele-server/src/main/java/eu/stratosphere/nephele/jobmanager/splitassigner/InputSplitManager.java +++ b/nephele/nephele-server/src/main/java/eu/stratosphere/nephele/jobmanager/splitassigner/InputSplitManager.java @@ -132,6 +132,15 @@ public void unregisterJob(final ExecutionGraph executionGraph) { while (it.hasNext()) { final ExecutionGroupVertex groupVertex = it.next(); + final InputSplit[] inputSplits = groupVertex.getInputSplits(); + + if (inputSplits == null) { + continue; + } + + if (inputSplits.length == 0) { + continue; + } final InputSplitAssigner assigner = this.assignerCache.remove(groupVertex); if (assigner == null) { diff --git a/pact/pact-common/src/main/java/eu/stratosphere/pact/common/contract/DataDistribution.java b/pact/pact-common/src/main/java/eu/stratosphere/pact/common/contract/DataDistribution.java index 06dbc318a98eb..d59b8c14d429c 100644 --- a/pact/pact-common/src/main/java/eu/stratosphere/pact/common/contract/DataDistribution.java +++ b/pact/pact-common/src/main/java/eu/stratosphere/pact/common/contract/DataDistribution.java @@ -1,3 +1,18 @@ +/*********************************************************************************************************************** + * + * Copyright (C) 2010 by the Stratosphere project (http://stratosphere.eu) + * + * Licensed 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 eu.stratosphere.pact.common.contract; import eu.stratosphere.pact.common.type.Key; diff --git a/pact/pact-common/src/main/java/eu/stratosphere/pact/common/plan/Plan.java b/pact/pact-common/src/main/java/eu/stratosphere/pact/common/plan/Plan.java index 3a18bab35abf4..67b0ffbdc487b 100644 --- a/pact/pact-common/src/main/java/eu/stratosphere/pact/common/plan/Plan.java +++ b/pact/pact-common/src/main/java/eu/stratosphere/pact/common/plan/Plan.java @@ -41,18 +41,25 @@ public class Plan implements Visitable { * The maximal number of machines to use in the job. */ protected int maxNumberMachines; + + /** + * The plan's configuration + */ + protected PlanConfiguration planConfiguration; // ------------------------------------------------------------------------ public Plan(Collection> sinks, String jobName) { this.sinks = sinks; this.jobName = jobName; + this.planConfiguration = new PlanConfiguration(); } public Plan(FileDataSinkContract sink, String jobName) { this.sinks = new ArrayList>(); this.sinks.add(sink); this.jobName = jobName; + this.planConfiguration = new PlanConfiguration(); } public Plan(Collection> sinks) { @@ -94,6 +101,13 @@ public void addDataSink(FileDataSinkContract sink) { public String getJobName() { return this.jobName; } + + /** + * Gets the plans configuration + */ + public PlanConfiguration getPlanConfiguration() { + return this.planConfiguration; + } /** * Gets the maximum number of machines to be used for this job. diff --git a/pact/pact-common/src/main/java/eu/stratosphere/pact/common/plan/PlanConfiguration.java b/pact/pact-common/src/main/java/eu/stratosphere/pact/common/plan/PlanConfiguration.java new file mode 100644 index 0000000000000..a5e37ce586b96 --- /dev/null +++ b/pact/pact-common/src/main/java/eu/stratosphere/pact/common/plan/PlanConfiguration.java @@ -0,0 +1,153 @@ +package eu.stratosphere.pact.common.plan; + +import eu.stratosphere.nephele.configuration.Configuration; + +public class PlanConfiguration extends Configuration { + + private static final String NEPHELE_PREFIX = "NEPHELE::"; + + /** + * Returns the Nephele value associated with the given key as a string. + * + * @param key + * the key pointing to the associated Nephele value + * @param defaultValue + * the default value which is returned in case there is no value associated with the given key + * @return the (default) value associated with the given key + */ + public String getNepheleString(String key, String defaultValue) { + return getString(NEPHELE_PREFIX+key, defaultValue); + } + + /** + * Adds the given key/value pair to the Nephele configuration. If either the key + * or the value is null the pair is not added. + * + * @param key + * the key of the key/value pair to be added + * @param value + * the value of the key/value pair to be added + */ + public void setNepheleString(String key, String value) { + setString(NEPHELE_PREFIX+key, value); + } + + /** + * Returns the Nephele value associated with the given key as an integer. + * + * @param key + * the key pointing to the associated value + * @param defaultValue + * the default value which is returned in case there is no value associated with the given key + * @return the (default) value associated with the given key + */ + public int getNepheleInteger(String key, int defaultValue) { + return getInteger(NEPHELE_PREFIX+key, defaultValue); + } + + /** + * Adds the given key/value pair to the Nephele configuration. If the + * key is null, the key/value pair is not added. + * + * @param key + * the key of the key/value pair to be added + * @param value + * the value of the key/value pair to be added + */ + public void setNepheleInteger(String key, int value) { + setInteger(NEPHELE_PREFIX+key, value); + } + + /** + * Returns the Nephele value associated with the given key as a long. + * + * @param key + * the key pointing to the associated Nephele value + * @param defaultValue + * the default value which is returned in case there is no value associated with the given key + * @return the (default) value associated with the given key + */ + public long getNepheleLong(String key, long defaultValue) { + return getLong(NEPHELE_PREFIX+key,defaultValue); + } + + /** + * Adds the given key/value pair to the Nephele configuration. If the + * key is null, the key/value pair is not added. + * + * @param key + * the key of the key/value pair to be added + * @param value + * the value of the key/value pair to be added + */ + public void setNepheleLong(String key, long value) { + setLong(NEPHELE_PREFIX+key, value); + } + + /** + * Returns the Nephele value associated with the given key as a boolean. + * + * @param key + * the key pointing to the associated Nephele value + * @param defaultValue + * the default value which is returned in case there is no value associated with the given key + * @return the (default) value associated with the given key + */ + public boolean getNepheleBoolean(String key, boolean defaultValue) { + return getBoolean(NEPHELE_PREFIX+key, defaultValue); + } + + /** + * Adds the given key/value pair to the Nephele configuration. If key is null the key/value pair is + * ignored and not added. + * + * @param key + * the key of the key/value pair to be added + * @param value + * the value of the key/value pair to be added + */ + public void setNepheleBoolean(String key, boolean value) { + setBoolean(NEPHELE_PREFIX+key, value); + } + + /** + * Returns the Nephele value associated with the given key as a float. + * + * @param key + * the key pointing to the associated Nephele value + * @param defaultValue + * the default value which is returned in case there is no value associated with the given key + * @return the (default) value associated with the given key + */ + public float getNepheleFloat(String key, float defaultValue) { + return getFloat(NEPHELE_PREFIX+key, defaultValue); + } + + /** + * Adds the given key/value pair to the Nephele configuration. If key is null the key/value pair is + * ignored and not added. + * + * @param key + * the key of the key/value pair to be added + * @param value + * the value of the key/value pair to be added + */ + public void setNepheleFloat(String key, float value) { + setFloat(NEPHELE_PREFIX+key, value); + } + + /** + * Copies all Nephele keys and values into the provided configuration object. + * + * @param nepheleConfiguration The configuration into which all Nephele keys and values are copied. + */ + public void extractNepheleConfiguration(Configuration nepheleConfiguration) { + + for(String key : this.keySet()) { + if(key.startsWith(NEPHELE_PREFIX)) { + nepheleConfiguration.setString(key.substring(NEPHELE_PREFIX.length()), this.getString(key, "")); + } + } + } + +} diff --git a/pact/pact-common/src/main/java/eu/stratosphere/pact/common/util/MutableObjectIterator.java b/pact/pact-common/src/main/java/eu/stratosphere/pact/common/util/MutableObjectIterator.java index ff236f072715a..104259979b399 100644 --- a/pact/pact-common/src/main/java/eu/stratosphere/pact/common/util/MutableObjectIterator.java +++ b/pact/pact-common/src/main/java/eu/stratosphere/pact/common/util/MutableObjectIterator.java @@ -1,8 +1,22 @@ +/*********************************************************************************************************************** + * + * Copyright (C) 2010 by the Stratosphere project (http://stratosphere.eu) + * + * Licensed 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 eu.stratosphere.pact.common.util; import java.io.IOException; - /** * A simple iterator interface. The key differences to the {@link java.util.Iterator} are that this * iterator accepts an object into which it places the content, and that is consolidates the logic diff --git a/pact/pact-compiler/src/main/java/eu/stratosphere/pact/compiler/PactCompiler.java b/pact/pact-compiler/src/main/java/eu/stratosphere/pact/compiler/PactCompiler.java index 3079a44657345..e2bffc5bb82d3 100644 --- a/pact/pact-compiler/src/main/java/eu/stratosphere/pact/compiler/PactCompiler.java +++ b/pact/pact-compiler/src/main/java/eu/stratosphere/pact/compiler/PactCompiler.java @@ -691,6 +691,7 @@ public OptimizedPlan compile(Plan pactPlan, InstanceTypeDescription type) throws // finalize the plan OptimizedPlan plan = new PlanFinalizer().createFinalPlan(bestPlanSinks, pactPlan.getJobName(), memoryMegabytes); plan.setInstanceTypeName(instanceName); + plan.setPlanConfiguration(pactPlan.getPlanConfiguration()); return plan; } @@ -706,8 +707,10 @@ public OptimizedPlan compile(Plan pactPlan, InstanceTypeDescription type) throws public static OptimizedPlan createPreOptimizedPlan(Plan pactPlan) { GraphCreatingVisitor graphCreator = new GraphCreatingVisitor(null, -1, -1, -1, false); pactPlan.accept(graphCreator); - return new OptimizedPlan(graphCreator.sources, graphCreator.sinks, graphCreator.con2node.values(), - pactPlan.getJobName()); + OptimizedPlan optPlan = new OptimizedPlan(graphCreator.sources, graphCreator.sinks, graphCreator.con2node.values(), + pactPlan.getJobName()); + optPlan.setPlanConfiguration(pactPlan.getPlanConfiguration()); + return optPlan; } /** diff --git a/pact/pact-compiler/src/main/java/eu/stratosphere/pact/compiler/jobgen/JobGraphGenerator.java b/pact/pact-compiler/src/main/java/eu/stratosphere/pact/compiler/jobgen/JobGraphGenerator.java index 4bc622e057b60..3741bbe89c47f 100644 --- a/pact/pact-compiler/src/main/java/eu/stratosphere/pact/compiler/jobgen/JobGraphGenerator.java +++ b/pact/pact-compiler/src/main/java/eu/stratosphere/pact/compiler/jobgen/JobGraphGenerator.java @@ -110,7 +110,10 @@ public JobGraph compileJobGraph(OptimizedPlan pactPlan) { this.vertices = new HashMap(); this.auxVertices = new ArrayList(); this.maxDegreeVertex = null; - + + // set Nephele JobGraph config + pactPlan.getPlanConfiguration().extractNepheleConfiguration(this.jobGraph.getJobConfiguration()); + // generate Nephele job graph pactPlan.accept(this); @@ -146,7 +149,7 @@ public JobGraph compileJobGraph(OptimizedPlan pactPlan) { // return job graph return graph; } - + /** * This methods implements the pre-visiting during a depth-first traversal. It create the job vertex and * sets local strategy. @@ -282,7 +285,7 @@ public void postVisit(OptimizerNode node) { "An error occurred while translating the optimized plan to a nephele JobGraph: " + e.getMessage(), e); } } - + // ------------------------------------------------------------------------ // Methods for creating individual vertices // ------------------------------------------------------------------------ @@ -290,7 +293,7 @@ public void postVisit(OptimizerNode node) { private boolean isDistributionGiven(PactConnection connection) { return (connection.getTargetPact().getPactContract().getCompilerHints().getInputDistributionClass() != null); } - + /** * @param mapNode * @return diff --git a/pact/pact-compiler/src/main/java/eu/stratosphere/pact/compiler/plan/OptimizedPlan.java b/pact/pact-compiler/src/main/java/eu/stratosphere/pact/compiler/plan/OptimizedPlan.java index e1a45d497841c..37d82e2e750b9 100644 --- a/pact/pact-compiler/src/main/java/eu/stratosphere/pact/compiler/plan/OptimizedPlan.java +++ b/pact/pact-compiler/src/main/java/eu/stratosphere/pact/compiler/plan/OptimizedPlan.java @@ -17,6 +17,7 @@ import java.util.Collection; +import eu.stratosphere.pact.common.plan.PlanConfiguration; import eu.stratosphere.pact.common.plan.Visitable; import eu.stratosphere.pact.common.plan.Visitor; @@ -48,12 +49,17 @@ public class OptimizedPlan implements Visitable { * Name of the PACT job */ private final String jobName; + + /** + * Configuration of the PACT job + */ + private PlanConfiguration planConfig; /** * The name of the instance type that is to be used. */ private String instanceTypeName; - + /** * Creates a new instance of this optimizer plan container. The plan is given and fully * described by the data sources, sinks and the collection of all nodes. @@ -110,7 +116,25 @@ public Collection getAllNodes() { public String getJobName() { return this.jobName; } - + + /** + * Returns the configuration of the PACT job. + * + * @return The configuration of the PACT job. + */ + public PlanConfiguration getPlanConfiguration() { + return this.planConfig; + } + + /** + * Sets the configuration of the PACT job. + * + * @param planConfig The configuration of the PACT job. + */ + public void setPlanConfiguration(PlanConfiguration planConfig) { + this.planConfig = planConfig; + } + /** * Gets the name of the instance type that should be used for this PACT job. * diff --git a/pact/pact-compiler/src/test/java/eu/stratosphere/pact/compiler/HardPlansCompilationTest.java b/pact/pact-compiler/src/test/java/eu/stratosphere/pact/compiler/HardPlansCompilationTest.java index feeacbe92e163..2728ac0c50437 100644 --- a/pact/pact-compiler/src/test/java/eu/stratosphere/pact/compiler/HardPlansCompilationTest.java +++ b/pact/pact-compiler/src/test/java/eu/stratosphere/pact/compiler/HardPlansCompilationTest.java @@ -1,3 +1,18 @@ +/*********************************************************************************************************************** + * + * Copyright (C) 2010 by the Stratosphere project (http://stratosphere.eu) + * + * Licensed 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 eu.stratosphere.pact.compiler; import java.net.InetAddress; diff --git a/pact/pact-compiler/src/test/java/eu/stratosphere/pact/compiler/PartitionLocalHashCompilerTest.java b/pact/pact-compiler/src/test/java/eu/stratosphere/pact/compiler/PartitionLocalHashCompilerTest.java index b0a3ec2479c84..f3e7a0b19f974 100644 --- a/pact/pact-compiler/src/test/java/eu/stratosphere/pact/compiler/PartitionLocalHashCompilerTest.java +++ b/pact/pact-compiler/src/test/java/eu/stratosphere/pact/compiler/PartitionLocalHashCompilerTest.java @@ -1,3 +1,18 @@ +/*********************************************************************************************************************** + * + * Copyright (C) 2010 by the Stratosphere project (http://stratosphere.eu) + * + * Licensed 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 eu.stratosphere.pact.compiler; import java.net.InetAddress; diff --git a/pact/pact-compiler/src/test/java/eu/stratosphere/pact/compiler/TempTaskSharingTest.java b/pact/pact-compiler/src/test/java/eu/stratosphere/pact/compiler/TempTaskSharingTest.java index 7f45c0d171f06..fb36e195c5c90 100644 --- a/pact/pact-compiler/src/test/java/eu/stratosphere/pact/compiler/TempTaskSharingTest.java +++ b/pact/pact-compiler/src/test/java/eu/stratosphere/pact/compiler/TempTaskSharingTest.java @@ -1,3 +1,18 @@ +/*********************************************************************************************************************** + * + * Copyright (C) 2010 by the Stratosphere project (http://stratosphere.eu) + * + * Licensed 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 eu.stratosphere.pact.compiler; import java.net.InetAddress; diff --git a/pact/pact-compiler/src/test/java/eu/stratosphere/pact/compiler/util/DummyCrossStub.java b/pact/pact-compiler/src/test/java/eu/stratosphere/pact/compiler/util/DummyCrossStub.java index 25bcc6f54eca6..f7748812154d9 100644 --- a/pact/pact-compiler/src/test/java/eu/stratosphere/pact/compiler/util/DummyCrossStub.java +++ b/pact/pact-compiler/src/test/java/eu/stratosphere/pact/compiler/util/DummyCrossStub.java @@ -1,3 +1,18 @@ +/*********************************************************************************************************************** + * + * Copyright (C) 2010 by the Stratosphere project (http://stratosphere.eu) + * + * Licensed 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 eu.stratosphere.pact.compiler.util; import eu.stratosphere.pact.common.stub.Collector; diff --git a/pact/pact-runtime/pom.xml b/pact/pact-runtime/pom.xml index 6ad1e38c3326a..81c20e2d6ad8f 100644 --- a/pact/pact-runtime/pom.xml +++ b/pact/pact-runtime/pom.xml @@ -48,6 +48,8 @@ **/TestData.java + + **/HashVsSortTest.java once -Xms512m -Xmx1024m diff --git a/pact/pact-runtime/src/main/java/eu/stratosphere/pact/runtime/task/AbstractPactTask.java b/pact/pact-runtime/src/main/java/eu/stratosphere/pact/runtime/task/AbstractPactTask.java index 2329322ec61f5..1dc179d3b7a43 100644 --- a/pact/pact-runtime/src/main/java/eu/stratosphere/pact/runtime/task/AbstractPactTask.java +++ b/pact/pact-runtime/src/main/java/eu/stratosphere/pact/runtime/task/AbstractPactTask.java @@ -1,5 +1,19 @@ -package eu.stratosphere.pact.runtime.task; +/*********************************************************************************************************************** + * + * Copyright (C) 2010 by the Stratosphere project (http://stratosphere.eu) + * + * Licensed 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 eu.stratosphere.pact.runtime.task; /** * diff --git a/pact/pact-runtime/src/main/java/eu/stratosphere/pact/runtime/task/CoGroupTask.java b/pact/pact-runtime/src/main/java/eu/stratosphere/pact/runtime/task/CoGroupTask.java index 5f41482e60dee..33ecc98ea2a66 100644 --- a/pact/pact-runtime/src/main/java/eu/stratosphere/pact/runtime/task/CoGroupTask.java +++ b/pact/pact-runtime/src/main/java/eu/stratosphere/pact/runtime/task/CoGroupTask.java @@ -499,7 +499,7 @@ private String getLogString(String message) bld.append(message); bld.append(':').append(' '); bld.append(this.getEnvironment().getTaskName()); - bld.append(' ').append('"'); + bld.append(' ').append('('); bld.append(this.getEnvironment().getIndexInSubtaskGroup() + 1); bld.append('/'); bld.append(this.getEnvironment().getCurrentNumberOfSubtasks()); diff --git a/pact/pact-runtime/src/main/java/eu/stratosphere/pact/runtime/task/CombineTask.java b/pact/pact-runtime/src/main/java/eu/stratosphere/pact/runtime/task/CombineTask.java index 604f0a63e7842..cb463eb5d204b 100644 --- a/pact/pact-runtime/src/main/java/eu/stratosphere/pact/runtime/task/CombineTask.java +++ b/pact/pact-runtime/src/main/java/eu/stratosphere/pact/runtime/task/CombineTask.java @@ -358,7 +358,7 @@ private String getLogString(String message) bld.append(message); bld.append(':').append(' '); bld.append(this.getEnvironment().getTaskName()); - bld.append(' ').append('"'); + bld.append(' ').append('('); bld.append(this.getEnvironment().getIndexInSubtaskGroup() + 1); bld.append('/'); bld.append(this.getEnvironment().getCurrentNumberOfSubtasks()); diff --git a/pact/pact-runtime/src/main/java/eu/stratosphere/pact/runtime/task/CrossTask.java b/pact/pact-runtime/src/main/java/eu/stratosphere/pact/runtime/task/CrossTask.java index fe7a245d32c5c..5d7540930b139 100644 --- a/pact/pact-runtime/src/main/java/eu/stratosphere/pact/runtime/task/CrossTask.java +++ b/pact/pact-runtime/src/main/java/eu/stratosphere/pact/runtime/task/CrossTask.java @@ -718,7 +718,7 @@ private String getLogString(String message) bld.append(message); bld.append(':').append(' '); bld.append(this.getEnvironment().getTaskName()); - bld.append(' ').append('"'); + bld.append(' ').append('('); bld.append(this.getEnvironment().getIndexInSubtaskGroup() + 1); bld.append('/'); bld.append(this.getEnvironment().getCurrentNumberOfSubtasks()); diff --git a/pact/pact-runtime/src/main/java/eu/stratosphere/pact/runtime/task/MapTask.java b/pact/pact-runtime/src/main/java/eu/stratosphere/pact/runtime/task/MapTask.java index c4f78e1cbedcc..fc41cf20a6a62 100644 --- a/pact/pact-runtime/src/main/java/eu/stratosphere/pact/runtime/task/MapTask.java +++ b/pact/pact-runtime/src/main/java/eu/stratosphere/pact/runtime/task/MapTask.java @@ -277,7 +277,7 @@ private String getLogString(String message) bld.append(message); bld.append(':').append(' '); bld.append(this.getEnvironment().getTaskName()); - bld.append(' ').append('"'); + bld.append(' ').append('('); bld.append(this.getEnvironment().getIndexInSubtaskGroup() + 1); bld.append('/'); bld.append(this.getEnvironment().getCurrentNumberOfSubtasks()); diff --git a/pact/pact-runtime/src/main/java/eu/stratosphere/pact/runtime/task/MatchTask.java b/pact/pact-runtime/src/main/java/eu/stratosphere/pact/runtime/task/MatchTask.java index 13cfd73c21332..77e5d9b50b54c 100644 --- a/pact/pact-runtime/src/main/java/eu/stratosphere/pact/runtime/task/MatchTask.java +++ b/pact/pact-runtime/src/main/java/eu/stratosphere/pact/runtime/task/MatchTask.java @@ -413,7 +413,7 @@ private String getLogString(String message) bld.append(message); bld.append(':').append(' '); bld.append(this.getEnvironment().getTaskName()); - bld.append(' ').append('"'); + bld.append(' ').append('('); bld.append(this.getEnvironment().getIndexInSubtaskGroup() + 1); bld.append('/'); bld.append(this.getEnvironment().getCurrentNumberOfSubtasks()); diff --git a/pact/pact-runtime/src/main/java/eu/stratosphere/pact/runtime/task/ReduceTask.java b/pact/pact-runtime/src/main/java/eu/stratosphere/pact/runtime/task/ReduceTask.java index a967e8bdbe6a9..23d8d50ecb949 100644 --- a/pact/pact-runtime/src/main/java/eu/stratosphere/pact/runtime/task/ReduceTask.java +++ b/pact/pact-runtime/src/main/java/eu/stratosphere/pact/runtime/task/ReduceTask.java @@ -475,7 +475,7 @@ private String getLogString(String message) bld.append(message); bld.append(':').append(' '); bld.append(this.getEnvironment().getTaskName()); - bld.append(' ').append('"'); + bld.append(' ').append('('); bld.append(this.getEnvironment().getIndexInSubtaskGroup() + 1); bld.append('/'); bld.append(this.getEnvironment().getCurrentNumberOfSubtasks()); diff --git a/pact/pact-runtime/src/main/java/eu/stratosphere/pact/runtime/task/SelfMatchTask.java b/pact/pact-runtime/src/main/java/eu/stratosphere/pact/runtime/task/SelfMatchTask.java index 9ee64edb88962..d6a8e33530584 100644 --- a/pact/pact-runtime/src/main/java/eu/stratosphere/pact/runtime/task/SelfMatchTask.java +++ b/pact/pact-runtime/src/main/java/eu/stratosphere/pact/runtime/task/SelfMatchTask.java @@ -656,7 +656,7 @@ private String getLogString(String message) bld.append(message); bld.append(':').append(' '); bld.append(this.getEnvironment().getTaskName()); - bld.append(' ').append('"'); + bld.append(' ').append('('); bld.append(this.getEnvironment().getIndexInSubtaskGroup() + 1); bld.append('/'); bld.append(this.getEnvironment().getCurrentNumberOfSubtasks()); diff --git a/pact/pact-runtime/src/main/java/eu/stratosphere/pact/runtime/task/TempTask.java b/pact/pact-runtime/src/main/java/eu/stratosphere/pact/runtime/task/TempTask.java index e828ea9201c20..8ca06b1e04a84 100644 --- a/pact/pact-runtime/src/main/java/eu/stratosphere/pact/runtime/task/TempTask.java +++ b/pact/pact-runtime/src/main/java/eu/stratosphere/pact/runtime/task/TempTask.java @@ -327,7 +327,7 @@ private String getLogString(String message) bld.append(message); bld.append(':').append(' '); bld.append(this.getEnvironment().getTaskName()); - bld.append(' ').append('"'); + bld.append(' ').append('('); bld.append(this.getEnvironment().getIndexInSubtaskGroup() + 1); bld.append('/'); bld.append(this.getEnvironment().getCurrentNumberOfSubtasks()); diff --git a/pact/pact-runtime/src/main/java/eu/stratosphere/pact/runtime/task/util/HistogramPartitionFunction.java b/pact/pact-runtime/src/main/java/eu/stratosphere/pact/runtime/task/util/HistogramPartitionFunction.java index 4d1ca6ca44184..e3a7e8939340f 100644 --- a/pact/pact-runtime/src/main/java/eu/stratosphere/pact/runtime/task/util/HistogramPartitionFunction.java +++ b/pact/pact-runtime/src/main/java/eu/stratosphere/pact/runtime/task/util/HistogramPartitionFunction.java @@ -1,3 +1,18 @@ +/*********************************************************************************************************************** + * + * Copyright (C) 2010 by the Stratosphere project (http://stratosphere.eu) + * + * Licensed 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 eu.stratosphere.pact.runtime.task.util; import java.util.Arrays; diff --git a/pact/pact-runtime/src/main/java/eu/stratosphere/pact/runtime/task/util/PartitionFunction.java b/pact/pact-runtime/src/main/java/eu/stratosphere/pact/runtime/task/util/PartitionFunction.java index a5c0797fe493f..09749a4a3d5e4 100644 --- a/pact/pact-runtime/src/main/java/eu/stratosphere/pact/runtime/task/util/PartitionFunction.java +++ b/pact/pact-runtime/src/main/java/eu/stratosphere/pact/runtime/task/util/PartitionFunction.java @@ -1,3 +1,18 @@ +/*********************************************************************************************************************** + * + * Copyright (C) 2010 by the Stratosphere project (http://stratosphere.eu) + * + * Licensed 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 eu.stratosphere.pact.runtime.task.util; import eu.stratosphere.pact.common.type.Key; diff --git a/pact/pact-runtime/src/test/java/eu/stratosphere/pact/runtime/hash/ProbeSideIteratorTest.java b/pact/pact-runtime/src/test/java/eu/stratosphere/pact/runtime/hash/ProbeSideIteratorTest.java index ba14a55a5cc83..aefec279ef921 100644 --- a/pact/pact-runtime/src/test/java/eu/stratosphere/pact/runtime/hash/ProbeSideIteratorTest.java +++ b/pact/pact-runtime/src/test/java/eu/stratosphere/pact/runtime/hash/ProbeSideIteratorTest.java @@ -1,5 +1,19 @@ -package eu.stratosphere.pact.runtime.hash; +/*********************************************************************************************************************** + * + * Copyright (C) 2010 by the Stratosphere project (http://stratosphere.eu) + * + * Licensed 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 eu.stratosphere.pact.runtime.hash; import java.util.ArrayList; import java.util.Iterator; @@ -14,8 +28,6 @@ import eu.stratosphere.pact.common.type.base.PactString; import eu.stratosphere.pact.runtime.hash.HashJoin.ProbeSideIterator; - - /** * @author Stephan Ewen */ diff --git a/pact/pact-tests/pom.xml b/pact/pact-tests/pom.xml index 3f9ded3bf0501..30a3e74c55f1a 100644 --- a/pact/pact-tests/pom.xml +++ b/pact/pact-tests/pom.xml @@ -123,9 +123,6 @@ always 1 false - - **/CrossITCase.class - diff --git a/stratosphere-dist/src/main/stratosphere-bin/bin/nephele-jobmanager.sh b/stratosphere-dist/src/main/stratosphere-bin/bin/nephele-jobmanager.sh index f157fc7ca2fe1..59b5a8d0a2078 100755 --- a/stratosphere-dist/src/main/stratosphere-bin/bin/nephele-jobmanager.sh +++ b/stratosphere-dist/src/main/stratosphere-bin/bin/nephele-jobmanager.sh @@ -61,6 +61,8 @@ constructJobManagerClassPath() { add=1 elif [[ "$jarfile" =~ 'nephele-clustermanager' ]]; then add=1 + elif [[ "$jarfile" =~ 'nephele-ec2cloudmanager' ]]; then + add=1 elif [[ "$jarfile" =~ 'commons-codec' ]]; then add=1 elif [[ "$jarfile" =~ 'commons-httpclient' ]]; then diff --git a/stratosphere-dist/src/main/stratosphere-bin/conf/nephele-user.xml b/stratosphere-dist/src/main/stratosphere-bin/conf/nephele-user.xml index 41af943a78a90..ae6e119b6ec68 100644 --- a/stratosphere-dist/src/main/stratosphere-bin/conf/nephele-user.xml +++ b/stratosphere-dist/src/main/stratosphere-bin/conf/nephele-user.xml @@ -102,7 +102,7 @@ jobmanager.instancemanager.cloud.classname - eu.stratosphere.nephele.instance.cloud.EC2CloudManager + eu.stratosphere.nephele.instance.cloud.CloudManager