// Copyright 2011 Google Inc.
//
// 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 com.google.appengine.tools.pipeline;
import com.google.appengine.tools.pipeline.JobInfo.State;
import com.google.appengine.tools.pipeline.JobSetting.StatusConsoleUrl;
import com.google.appengine.tools.pipeline.impl.PipelineManager;
import com.google.appengine.tools.pipeline.impl.model.JobRecord;
import com.google.appengine.tools.pipeline.impl.model.PipelineObjects;
import com.google.common.base.Function;
import com.google.common.collect.ImmutableList;
import java.io.Serializable;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Random;
import java.util.concurrent.CancellationException;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicBoolean;
/**
* Misc tests including:
* Passing large values.
* Delay used in a slow job
* JobSetting inheritance
* PromisedValue (and filling it more than once)
* Cancel pipeline
* WaitFor passed to a new pipeline
* FutureValue passed to a new pipeline
*
* @author rudominer@google.com (Mitch Rudominer)
*/
public class MiscPipelineTest extends PipelineTest {
private static long[] largeValue;
@Override
public void setUp() throws Exception {
super.setUp();
largeValue = new long[2000000];
Random random = new Random();
for (int i = 0; i < largeValue.length; i++) {
largeValue[i] = random.nextLong();
}
}
@Override
protected boolean isHrdSafe() {
return false;
}
@SuppressWarnings("serial")
private static class ReturnFutureListJob extends Job0<List<String>> {
@Override
public Value<List<String>> run() throws Exception {
FutureValue<String> child1 = futureCall(new StrJob<>(), immediate(Long.valueOf(123)));
FutureValue<String> child2 = futureCall(new StrJob<>(), immediate(Long.valueOf(456)));
return new FutureList<>(ImmutableList.of(child1, child2));
}
}
public void testReturnFutureList() throws Exception {
PipelineService service = PipelineServiceFactory.newPipelineService();
String pipelineId = service.startNewPipeline(new ReturnFutureListJob());
List<String> value = waitForJobToComplete(pipelineId);
assertEquals(ImmutableList.of("123", "456"), value);
}
private static class StringToLong implements Function<String, Long>, Serializable {
private static final long serialVersionUID = -913828405842203610L;
@Override
public Long apply(String input) {
return Long.valueOf(input);
}
}
@SuppressWarnings("serial")
private static class TestTransformJob extends Job0<Long> {
@Override
public Value<Long> run() {
FutureValue<String> child1 = futureCall(new StrJob<>(), immediate(Long.valueOf(123)));
return futureCall(new Jobs.Transform<>(new StringToLong()), child1);
}
}
public void testTransform() throws Exception {
PipelineService service = PipelineServiceFactory.newPipelineService();
String pipelineId = service.startNewPipeline(new TestTransformJob());
Long value = waitForJobToComplete(pipelineId);
assertEquals(Long.valueOf(123), value);
}
@SuppressWarnings("serial")
private static class RootJob extends Job0<String> {
private final boolean delete;
RootJob(boolean delete) {
this.delete = delete;
}
@Override
public Value<String> run() throws Exception {
FutureValue<String> child1 = futureCall(new StrJob<>(), immediate(Long.valueOf(123)));
FutureValue<String> child2 = futureCall(new StrJob<>(), immediate(Long.valueOf(456)));
if (delete) {
return Jobs.waitForAllAndDelete(this, child1, child2);
} else {
return Jobs.waitForAll(this, child1, child2);
}
}
}
public void testWaitForAll() throws Exception {
PipelineService service = PipelineServiceFactory.newPipelineService();
String pipelineId = service.startNewPipeline(new RootJob(false));
JobInfo jobInfo = waitUntilJobComplete(pipelineId);
assertEquals(JobInfo.State.COMPLETED_SUCCESSFULLY, jobInfo.getJobState());
assertEquals("123", jobInfo.getOutput());
assertNotNull(service.getJobInfo(pipelineId));
}
public void testWaitForAllAndDelete() throws Exception {
PipelineService service = PipelineServiceFactory.newPipelineService();
String pipelineId = service.startNewPipeline(new RootJob(true));
JobInfo jobInfo = waitUntilJobComplete(pipelineId);
assertEquals(JobInfo.State.COMPLETED_SUCCESSFULLY, jobInfo.getJobState());
assertEquals("123", jobInfo.getOutput());
waitUntilTaskQueueIsEmpty();
try {
service.getJobInfo(pipelineId);
fail("Was expecting a NoSuchObjectException exception");
} catch (NoSuchObjectException expected) {
// expected;
}
}
@SuppressWarnings("serial")
private static class CallerJob extends Job0<String> {
static AtomicBoolean flag = new AtomicBoolean();
@Override
public Value<String> run() throws Exception {
FutureValue<Void> child1 = futureCall(new ChildJob());
FutureValue<String> child2 = futureCall(new StrJob<>(), immediate(Long.valueOf(123)));
PipelineService service = PipelineServiceFactory.newPipelineService();
String str1 = service.startNewPipeline(new EchoJob<>(), child2);
String str2 = service.startNewPipeline(new CalledJob(), waitFor(child1));
return immediate(str1 + "," + str2);
}
}
@SuppressWarnings("serial")
private static class EchoJob<T> extends Job1<T, T> {
@Override
public Value<T> run(T t) throws Exception {
return immediate(t);
}
}
@SuppressWarnings("serial")
private static class ChildJob extends Job0<Void> {
@Override
public Value<Void> run() throws Exception {
Thread.sleep(5000);
CallerJob.flag.set(true);
return null;
}
}
@SuppressWarnings("serial")
private static class CalledJob extends Job0<Boolean> {
@Override
public Value<Boolean> run() throws Exception {
return immediate(CallerJob.flag.get());
}
}
public void testWaitForUsedByNewPipeline() throws Exception {
PipelineService service = PipelineServiceFactory.newPipelineService();
String pipelineId = service.startNewPipeline(new CallerJob());
JobInfo jobInfo = waitUntilJobComplete(pipelineId);
assertEquals(JobInfo.State.COMPLETED_SUCCESSFULLY, jobInfo.getJobState());
String[] calledPipelines = ((String) jobInfo.getOutput()).split(",");
jobInfo = waitUntilJobComplete(calledPipelines[0]);
assertEquals(JobInfo.State.COMPLETED_SUCCESSFULLY, jobInfo.getJobState());
assertEquals("123", jobInfo.getOutput());
jobInfo = waitUntilJobComplete(calledPipelines[1]);
assertEquals(JobInfo.State.COMPLETED_SUCCESSFULLY, jobInfo.getJobState());
assertTrue((Boolean) jobInfo.getOutput());
}
@SuppressWarnings("serial")
private abstract static class AbstractJob extends Job0<String> {
@Override
public Value<String> run() throws Exception {
return immediate(getValue());
}
protected abstract String getValue();
}
@SuppressWarnings("serial")
private static class ConcreteJob extends AbstractJob {
@Override
protected String getValue() {
return "Shalom";
}
@Override
public String getJobDisplayName() {
return "ConcreteJob: " + getValue();
}
@SuppressWarnings("unused")
public Value<String> handleException(Throwable t) {
return immediate("Got exception!");
}
}
public void testGetJobDisplayName() throws Exception {
PipelineService service = PipelineServiceFactory.newPipelineService();
ConcreteJob job = new ConcreteJob();
String pipelineId = service.startNewPipeline(job);
JobRecord jobRecord = PipelineManager.getJob(pipelineId);
assertEquals(job.getJobDisplayName(), jobRecord.getRootJobDisplayName());
JobInfo jobInfo = waitUntilJobComplete(pipelineId);
assertEquals("Shalom", jobInfo.getOutput());
jobRecord = PipelineManager.getJob(pipelineId);
assertEquals(job.getJobDisplayName(), jobRecord.getRootJobDisplayName());
PipelineObjects pobjects = PipelineManager.queryFullPipeline(pipelineId);
assertEquals(job.getJobDisplayName(), pobjects.rootJob.getRootJobDisplayName());
}
public void testJobInheritence() throws Exception {
PipelineService service = PipelineServiceFactory.newPipelineService();
String pipelineId = service.startNewPipeline(new ConcreteJob());
JobInfo jobInfo = waitUntilJobComplete(pipelineId);
assertEquals("Shalom", jobInfo.getOutput());
}
@SuppressWarnings("serial")
private static class FailedJob extends Job0<String> {
@Override
public Value<String> run() throws Exception {
throw new RuntimeException("koko");
}
}
public void testJobFailure() throws Exception {
PipelineService service = PipelineServiceFactory.newPipelineService();
String pipelineId = service.startNewPipeline(new FailedJob());
JobInfo jobInfo = waitUntilJobComplete(pipelineId);
assertEquals(JobInfo.State.STOPPED_BY_ERROR, jobInfo.getJobState());
assertEquals("koko", jobInfo.getException().getMessage());
assertNull(jobInfo.getOutput());
}
public void testReturnValue() throws Exception {
// Testing that return value from parent is always after all children complete
// which is not the case right now. This this *SHOULD* change after we fix
// it, as fixing it should cause a dead-lock.
// see b/12249138
PipelineService service = PipelineServiceFactory.newPipelineService();
String pipelineId = service.startNewPipeline(new ReturnValueParentJob());
String value = waitForJobToComplete(pipelineId);
assertEquals("bla", value);
ReturnValueParentJob.latch1.countDown();
waitUntilTaskQueueIsEmpty();
ReturnValueParentJob.latch2.await();
}
@SuppressWarnings("serial")
private static class ReturnValueParentJob extends Job0<String> {
static CountDownLatch latch1 = new CountDownLatch(1);
static CountDownLatch latch2 = new CountDownLatch(1);
@Override
public Value<String> run() throws Exception {
futureCall(new ReturnedValueChildJob());
return immediate("bla");
}
}
@SuppressWarnings("serial")
private static class ReturnedValueChildJob extends Job0<Void> {
@Override
public Value<Void> run() throws Exception {
ReturnValueParentJob.latch1.await();
ReturnValueParentJob.latch2.countDown();
return null;
}
}
public void testSubmittingPromisedValueMoreThanOnce() throws Exception {
PipelineService service = PipelineServiceFactory.newPipelineService();
String pipelineId = service.startNewPipeline(new SubmitPromisedParentJob());
String value = waitForJobToComplete(pipelineId);
assertEquals("2", value);
}
@SuppressWarnings("serial")
private static class SubmitPromisedParentJob extends Job0<String> {
@Override
public Value<String> run() throws Exception {
PromisedValue<String> promise = newPromise();
FutureValue<Void> child1 = futureCall(
new FillPromiseJob(), immediate("1"), immediate(promise.getHandle()));
FutureValue<Void> child2 = futureCall(
new FillPromiseJob(), immediate("2"), immediate(promise.getHandle()), waitFor(child1));
FutureValue<String> child3 = futureCall(new ReadPromiseJob(), promise, waitFor(child2));
// If we return promise directly then the value would be "1" rather than "2", see b/12216307
//return promise;
return child3;
}
}
@SuppressWarnings("serial")
private static class ReadPromiseJob extends Job1<String, String> {
@Override
public Value<String> run(String value) {
return immediate(value);
}
}
@SuppressWarnings("serial")
private static class FillPromiseJob extends Job2<Void, String, String> {
@Override
public Value<Void> run(String value, String handle) throws Exception {
PipelineServiceFactory.newPipelineService().submitPromisedValue(handle, value);
return null;
}
}
public void testCancelPipeline() throws Exception {
PipelineService service = PipelineServiceFactory.newPipelineService();
String pipelineId = service.startNewPipeline(new HandleExceptionParentJob(),
new JobSetting.BackoffSeconds(1), new JobSetting.BackoffFactor(1),
new JobSetting.MaxAttempts(2));
JobInfo jobInfo = service.getJobInfo(pipelineId);
assertEquals(State.RUNNING, jobInfo.getJobState());
HandleExceptionChild2Job.childLatch1.await();
service.cancelPipeline(pipelineId);
HandleExceptionChild2Job.childLatch2.countDown();
waitUntilTaskQueueIsEmpty();
jobInfo = service.getJobInfo(pipelineId);
assertEquals(State.CANCELED_BY_REQUEST, jobInfo.getJobState());
assertNull(jobInfo.getOutput());
assertTrue(HandleExceptionParentJob.child0.get());
assertTrue(HandleExceptionParentJob.child1.get());
assertTrue(HandleExceptionParentJob.child2.get());
assertFalse(HandleExceptionParentJob.child3.get());
assertFalse(HandleExceptionParentJob.child4.get());
// Unexpected callbacks (should be fixed after b/12250957)
assertTrue(HandleExceptionParentJob.child3Cancel.get()); // job not started
assertTrue(HandleExceptionParentJob.child4Cancel.get()); // job not started
// expected callbacks
assertFalse(HandleExceptionParentJob.child0Cancel.get()); // job already finalized
assertTrue(HandleExceptionParentJob.parentCancel.get());
assertTrue(HandleExceptionParentJob.child1Cancel.get());
assertTrue(HandleExceptionParentJob.child2Cancel.get()); // after job run, but not finalized
}
@SuppressWarnings("serial")
private static class HandleExceptionParentJob extends Job0<String> {
private static AtomicBoolean parentCancel = new AtomicBoolean();
private static AtomicBoolean child0 = new AtomicBoolean();
private static AtomicBoolean child0Cancel = new AtomicBoolean();
private static AtomicBoolean child1 = new AtomicBoolean();
private static AtomicBoolean child1Cancel = new AtomicBoolean();
private static AtomicBoolean child2 = new AtomicBoolean();
private static AtomicBoolean child2Cancel = new AtomicBoolean();
private static AtomicBoolean child3 = new AtomicBoolean();
private static AtomicBoolean child3Cancel = new AtomicBoolean();
private static AtomicBoolean child4 = new AtomicBoolean();
private static AtomicBoolean child4Cancel = new AtomicBoolean();
@Override
public Value<String> run() {
FutureValue<String> child0 = futureCall(new HandleExceptionChild0Job());
FutureValue<String> child1 = futureCall(new HandleExceptionChild1Job(), waitFor(child0));
FutureValue<String> child2 = futureCall(new HandleExceptionChild3Job(), waitFor(child1));
return futureCall(new HandleExceptionChild4Job(), child1, child2);
}
@SuppressWarnings("unused")
public Value<String> handleException(CancellationException ex) throws Exception {
HandleExceptionParentJob.parentCancel.set(true);
return immediate("should not be used");
}
}
@SuppressWarnings("serial")
private static class HandleExceptionChild0Job extends Job0<String> {
@Override
public Value<String> run() {
HandleExceptionParentJob.child0.set(true);
return immediate("1");
}
@SuppressWarnings("unused")
public Value<String> handleException(CancellationException ex) {
HandleExceptionParentJob.child0Cancel.set(true);
return null;
}
}
@SuppressWarnings("serial")
private static class HandleExceptionChild1Job extends Job0<String> {
@Override
public Value<String> run() {
HandleExceptionParentJob.child1.set(true);
return futureCall(new HandleExceptionChild2Job());
}
@SuppressWarnings("unused")
public Value<String> handleException(CancellationException ex) {
HandleExceptionParentJob.child1Cancel.set(true);
return null;
}
}
@SuppressWarnings("serial")
private static class HandleExceptionChild2Job extends Job0<String> {
private static CountDownLatch childLatch1 = new CountDownLatch(1);
private static CountDownLatch childLatch2 = new CountDownLatch(1);
@Override
public Value<String> run() throws InterruptedException {
HandleExceptionParentJob.child2.set(true);
childLatch1.countDown();
childLatch2.await();
Thread.sleep(1000);
return immediate("1");
}
@SuppressWarnings("unused")
public Value<String> handleException(CancellationException ex) {
HandleExceptionParentJob.child2Cancel.set(true);
return null;
}
}
@SuppressWarnings("serial")
private static class HandleExceptionChild3Job extends Job0<String> {
@Override
public Value<String> run() {
HandleExceptionParentJob.child3.set(true);
return immediate("2");
}
@SuppressWarnings("unused")
public Value<String> handleException(CancellationException ex) {
HandleExceptionParentJob.child3Cancel.set(true);
return null;
}
}
@SuppressWarnings("serial")
private static class HandleExceptionChild4Job extends Job2<String, String, String> {
@Override
public Value<String> run(String str1, String str2) {
HandleExceptionParentJob.child4.set(true);
return immediate(str1 + str2);
}
@SuppressWarnings("unused")
public Value<String> handleException(CancellationException ex) {
HandleExceptionParentJob.child4Cancel.set(true);
return immediate("not going to be used");
}
}
public void testImmediateChild() throws Exception {
// This is also testing inheritance of statusConsoleUrl.
PipelineService service = PipelineServiceFactory.newPipelineService();
String pipelineId = service.startNewPipeline(new Returns5FromChildJob(), largeValue);
Integer five = waitForJobToComplete(pipelineId);
assertEquals(5, five.intValue());
}
public void testPromisedValue() throws Exception {
PipelineService service = PipelineServiceFactory.newPipelineService();
String pipelineId = service.startNewPipeline(new FillPromisedValueJob());
String helloWorld = (String) waitForJobToComplete(pipelineId);
assertEquals("hello world", helloWorld);
}
public void testDelayedValueInSlowJob() throws Exception {
PipelineService service = PipelineServiceFactory.newPipelineService();
String pipelineId = service.startNewPipeline(new UsingDelayedValueInSlowJob());
String hello = (String) waitForJobToComplete(pipelineId);
assertEquals("I am delayed", hello);
}
@SuppressWarnings("serial")
static class Temp<T extends Serializable> implements Serializable {
private final T value;
Temp(T value) {
this.value = value;
}
T getValue() {
return value;
}
}
@SuppressWarnings("serial")
private static class FillPromisedValueJob extends Job0<String> {
@Override
public Value<String> run() {
PromisedValue<List<Temp<String>>> ps = newPromise();
futureCall(new PopulatePromisedValueJob(), immediate(ps.getHandle()));
return futureCall(new ConsumePromisedValueJob(), ps);
}
}
@SuppressWarnings("serial")
private static class PopulatePromisedValueJob extends Job1<Void, String> {
@Override
public Value<Void> run(String handle) throws NoSuchObjectException, OrphanedObjectException {
List<Temp<String>> list = new ArrayList<>();
list.add(new Temp<>("hello"));
list.add(new Temp<>(" "));
list.add(new Temp<>("world"));
PipelineManager.acceptPromisedValue(handle, list);
return null;
}
}
@SuppressWarnings("serial")
private static class ConsumePromisedValueJob extends Job1<String, List<Temp<String>>> {
@Override
public Value<String> run(List<Temp<String>> values) {
String value = "";
for (Temp<String> temp : values) {
value += temp.getValue();
}
return immediate(value);
}
}
@SuppressWarnings("serial")
private static class StrJob<T extends Serializable> extends Job1<String, T> {
@Override
public Value<String> run(T obj) {
return immediate(obj == null ? "null" : obj.toString());
}
}
@SuppressWarnings("serial")
private static class UsingDelayedValueInSlowJob extends Job0<String> {
@Override
public Value<String> run() throws InterruptedException {
Value<Void> delayedValue = newDelayedValue(1);
Thread.sleep(3000);
// We would like to validate that the delay will work even when used after
// its delayed value. It used to fail before, b/12081152, but now it should
// pass (and semantic of the delay was changed, so delay value starts only
// after this run method completes.
return futureCall(new StrJob<>(), immediate("I am delayed"), waitFor(delayedValue));
}
}
@SuppressWarnings("serial")
private static class Returns5FromChildJob extends Job1<Integer, long[]> {
@Override
public Value<Integer> run(long[] bytes) {
setStatusConsoleUrl("my-console-url");
assertTrue(Arrays.equals(largeValue, bytes));
FutureValue<Integer> lengthJob = futureCall(new LengthJob(), immediate(bytes));
return futureCall(
new IdentityJob(bytes), immediate(5), lengthJob, new StatusConsoleUrl(null));
}
}
@SuppressWarnings("serial")
private static class IdentityJob extends Job2<Integer, Integer, Integer> {
private final long[] bytes;
public IdentityJob(long[] bytes) {
this.bytes = bytes;
}
@Override
public Value<Integer> run(Integer param1, Integer length) {
assertNull(getStatusConsoleUrl());
assertEquals(largeValue.length, length.intValue());
assertTrue(Arrays.equals(largeValue, bytes));
return immediate(param1);
}
}
@SuppressWarnings("serial")
private static class LengthJob extends Job1<Integer, long[]> {
@Override
public Value<Integer> run(long[] bytes) {
assertEquals("my-console-url", getStatusConsoleUrl());
assertTrue(Arrays.equals(largeValue, bytes));
return immediate(bytes.length);
}
}
}