This closes If an Exception is thrown when merging FlowFiles, ensure that we remove the 'bundle' flowfile before exiting the method
@@ -578,6 +578,7 @@ public class MergeContent extends BinFiles {
final ProcessSession session = bin.getSession();
FlowFile bundle = session.create(bin.getContents());
final AtomicReference<String> bundleMimeTypeRef = new AtomicReference<>(null);
+ try {
bundle = session.write(bundle, new OutputStreamCallback() {
@Override
public void process(final OutputStream out) throws IOException {
@@ -621,6 +622,10 @@ public class MergeContent extends BinFiles {
}
}
});
+ } catch (final Exception e) {
+ session.remove(bundle);
+ throw e;
+ }
session.getProvenanceReporter().join(contents, bundle);
bundle = session.putAttribute(bundle, CoreAttributes.FILENAME.key(), createFilename(contents));
@@ -719,6 +724,7 @@ public class MergeContent extends BinFiles {
final boolean keepPath = context.getProperty(KEEP_PATH).asBoolean();
FlowFile bundle = session.create(); // we don't pass the parents to the #create method because the parents belong to different sessions
+ try {
bundle = session.putAttribute(bundle, CoreAttributes.FILENAME.key(), createFilename(contents) + ".tar");
bundle = session.write(bundle, new OutputStreamCallback() {
@Override
@@ -761,6 +767,10 @@ public class MergeContent extends BinFiles {
}
}
});
+ } catch (final Exception e) {
+ session.remove(bundle);
+ throw e;
+ }
bin.getSession().getProvenanceReporter().join(contents, bundle);
return bundle;
@@ -794,6 +804,7 @@ public class MergeContent extends BinFiles {
FlowFile bundle = session.create(contents);
+ try {
bundle = session.write(bundle, new OutputStreamCallback() {
@Override
public void process(final OutputStream rawOut) throws IOException {
@@ -823,6 +834,10 @@ public class MergeContent extends BinFiles {
}
}
});
+ } catch (final Exception e) {
+ session.remove(bundle);
+ throw e;
+ }
bundle = session.putAttribute(bundle, CoreAttributes.FILENAME.key(), createFilename(contents) + ".pkg");
session.getProvenanceReporter().join(contents, bundle);
@@ -860,6 +875,7 @@ public class MergeContent extends BinFiles {
FlowFile bundle = session.create(contents);
+ try {
bundle = session.putAttribute(bundle, CoreAttributes.FILENAME.key(), createFilename(contents) + ".zip");
bundle = session.write(bundle, new OutputStreamCallback() {
@Override
@@ -888,6 +904,10 @@ public class MergeContent extends BinFiles {
}
}
});
+ } catch (final Exception e) {
+ session.remove(bundle);
+ throw e;
+ }
session.getProvenanceReporter().join(contents, bundle);
return bundle;
@@ -921,6 +941,7 @@ public class MergeContent extends BinFiles {
// we don't pass the parents to the #create method because the parents belong to different sessions
FlowFile bundle = session.create(contents);
+ try {
bundle = session.write(bundle, new OutputStreamCallback() {
@Override
public void process(final OutputStream rawOut) throws IOException {
@@ -1007,6 +1028,10 @@ public class MergeContent extends BinFiles {
}
}
});
+ } catch (final Exception e) {
+ session.remove(bundle);
+ throw e;
+ }
final Collection<FlowFile> parents;
if (unmerged.isEmpty()) {
