More patches for NPE problems when not using input or output files
shlomo....@gmail.com
Here are some more patches for the current codebase to prevent
NullPointerExceptions that happen when operating a workflow with no
MetaFiles. The NPEs would occur upon ingesting a Properties or
transferring control to the next Service of a workflow lacking
MetaFiles.
.. Shlomo
David Kavanagh
way now and had to address this as well. I think I should just go back
through gmail to find all of your patches! (in fact, I'm look at that
now). I'm seeing a problem with instance status, after initially
reporting busy for a while, after they go idle, they never are able to
report busy with any kind of decent load value. I'm reviewing that
code now.
David
> Index: java/com/directthought/lifeguard/AbstractBaseService.java
> ===================================================================
> --- java/com/directthought/lifeguard/AbstractBaseService.java (revision 89)
> +++ java/com/directthought/lifeguard/AbstractBaseService.java (working copy)
> @@ -215,21 +215,23 @@
> }
> }
> // send results to S3
> - for (MetaFile file : results) {
> - if (file.key == null || file.key.trim().equals("")) {
> - file.key = MD5Util.md5Sum(new FileInputStream(file.file));
> + if (results != null) {
> + for (MetaFile file : results) {
> + if (file.key == null || file.key.trim().equals("")) {
> + file.key = MD5Util.md5Sum(new FileInputStream(file.file));
> + }
> + S3Bucket outBucket = new S3Bucket(request.getOutputBucket());
> + S3Object obj = new S3Object(outBucket, file.key);
> + obj.setDataInputFile(file.file);
> + obj.setContentLength(file.file.length());
> + obj.setContentType(file.mimeType);
> + obj = s3.putObject(outBucket, obj);
> + obj.closeDataInputStream();
> + }
> + // after all transferred, delete them locally
> + for (MetaFile file : results) {
> + file.file.delete();
> }
> - S3Bucket outBucket = new S3Bucket(request.getOutputBucket());
> - S3Object obj = new S3Object(outBucket, file.key);
> - obj.setDataInputFile(file.file);
> - obj.setContentLength(file.file.length());
> - obj.setContentType(file.mimeType);
> - obj = s3.putObject(outBucket, obj);
> - obj.closeDataInputStream();
> - }
> - // after all transferred, delete them locally
> - for (MetaFile file : results) {
> - file.file.delete();
> }
> long endTime = System.currentTimeMillis();
> // create status
> @@ -240,11 +242,13 @@
> if (next != null) {
> MessageQueue nextQueue = QueueUtil.getQueueOrElse(qs,
> queuePrefix+next.getWorkQueue());
> - String mimeType = next.getType();
> - for (MetaFile file : results) {
> - if (file.mimeType.equals(mimeType)) {
> - request.getInput().setType(mimeType);
> - request.getInput().setKey(file.key);
> + if (results != null) {
> + String mimeType = next.getType();
> + for (MetaFile file : results) {
> + if (file.mimeType.equals(mimeType)) {
> + request.getInput().setType(mimeType);
> + request.getInput().setKey(file.key);
> + }
> }
> }
> request.setNextStep(next.getNextStep());
> Index: java/com/directthought/lifeguard/IngestorBase.java
> ===================================================================
> --- java/com/directthought/lifeguard/IngestorBase.java (revision 89)
> +++ java/com/directthought/lifeguard/IngestorBase.java (working copy)
> @@ -153,31 +153,33 @@
>
> try {
> WorkRequest wr = constructBaseWorkRequest(outputKeys);
> - for (MetaFile file : files) {
> - long startTime = System.currentTimeMillis();
> - if (file.file != null) {
> - // put file in S3 input bucket
> - String s3Key = MD5Util.md5Sum(new FileInputStream(file.file));
> - RestS3Service s3 = new RestS3Service(new AWSCredentials(awsAccessId, awsSecretKey));
> - if (proxyHost != null && !proxyHost.equals("")) {
> - s3.initHttpProxy(proxyHost, proxyPort);
> + if (files != null) {
> + for (MetaFile file : files) {
> + long startTime = System.currentTimeMillis();
> + if (file.file != null) {
> + // put file in S3 input bucket
> + String s3Key = MD5Util.md5Sum(new FileInputStream(file.file));
> + RestS3Service s3 = new RestS3Service(new AWSCredentials(awsAccessId, awsSecretKey));
> + if (proxyHost != null && !proxyHost.equals("")) {
> + s3.initHttpProxy(proxyHost, proxyPort);
> + }
> + S3Object obj = new S3Object(new S3Bucket(inputBucket), s3Key);
> + obj.setDataInputFile(file.file);
> + obj.setContentLength(file.file.length());
> + obj.setContentType(file.mimeType);
> + obj = s3.putObject(inputBucket, obj);
> + file.key = s3Key;
> }
> - S3Object obj = new S3Object(new S3Bucket(inputBucket), s3Key);
> - obj.setDataInputFile(file.file);
> - obj.setContentLength(file.file.length());
> - obj.setContentType(file.mimeType);
> - obj = s3.putObject(inputBucket, obj);
> - file.key = s3Key;
> + // send work request message
> + FileRef ref = of.createFileRef();
> + ref.setKey(file.key);
> + ref.setType(file.mimeType);
> + ref.setLocation("");
> + wr.setInput(ref);
> + long endTime = System.currentTimeMillis();
> + sendMessages(wr, file, startTime, endTime, workQueue, statusQueue);
> + logger.debug("ingested : "+((file.file==null)?file.key:file.file.getName()));
> }
> - // send work request message
> - FileRef ref = of.createFileRef();
> - ref.setKey(file.key);
> - ref.setType(file.mimeType);
> - ref.setLocation("");
> - wr.setInput(ref);
> - long endTime = System.currentTimeMillis();
> - sendMessages(wr, file, startTime, endTime, workQueue, statusQueue);
> - logger.debug("ingested : "+((file.file==null)?file.key:file.file.getName()));
> }
> if (properties != null) {
> for (Properties props : properties) {
> Index: java/com/directthought/lifeguard/MessageHelper.java
> ===================================================================
> --- java/com/directthought/lifeguard/MessageHelper.java (revision 89)
> +++ java/com/directthought/lifeguard/MessageHelper.java (working copy)
> @@ -65,10 +65,14 @@
> ret.setBatch(wr.getBatch());
> ret.setServiceName(wr.getServiceName());
> ret.setInputBucket(wr.getInputBucket());
> - FileRef ref = of.createFileRef();
> - ref.setKey(wr.getInput().getKey());
> - ref.setType(wr.getInput().getType());
> - ref.setLocation(wr.getInput().getLocation());
> + FileRef ref = null;
> + FileRef input = wr.getInput();
> + if (input != null) {
> + ref = of.createFileRef();
> + ref.setKey(input.getKey());
> + ref.setType(input.getType());
> + ref.setLocation(input.getLocation());
> + }
> ret.setInput(ref);
> ret.setOutputBucket(wr.getOutputBucket());
> if (outFiles != null) {
>
>