Apache
/
nifi
mirror of https://github.com/apache/nifi.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

NIFI-7683: Fixed bug that resulted in a byte[] being allocated and held onto by a member variable that was unprotected. This caused the byte array to be modified by two different threads concurrently, which can potentially cause corruption of FlowFile's data 

Signed-off-by: Matthew Burgess <mattyb149@apache.org>

This closes #4439
This commit is contained in:
Mark Payne 2020-07-28 13:44:32 -04:00 committed by Matthew Burgess
parent 90c9db8208
commit e926a17968
No known key found for this signature in database
GPG Key ID: 05D3DEB8126DAD24
1 changed files with 22 additions and 51 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

73
nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ReplaceText.java
View File

@ -46,8 +46,6 @@ import org.apache.nifi.processor.ProcessSession;
import org.apache.nifi.processor.ProcessorInitializationContext; import org.apache.nifi.processor.ProcessorInitializationContext;
import org.apache.nifi.processor.Relationship; import org.apache.nifi.processor.Relationship;
import org.apache.nifi.processor.exception.ProcessException; import org.apache.nifi.processor.exception.ProcessException;
import org.apache.nifi.processor.io.InputStreamCallback;
import org.apache.nifi.processor.io.OutputStreamCallback;
import org.apache.nifi.processor.io.StreamCallback; import org.apache.nifi.processor.io.StreamCallback;
import org.apache.nifi.processor.util.StandardValidators; import org.apache.nifi.processor.util.StandardValidators;
import org.apache.nifi.stream.io.StreamUtils; import org.apache.nifi.stream.io.StreamUtils;
@ -260,14 +258,6 @@ public class ReplaceText extends AbstractProcessor {
public void setup(ProcessContext context) { public void setup(ProcessContext context) {
final String replacementStrategy = context.getProperty(REPLACEMENT_STRATEGY).getValue(); final String replacementStrategy = context.getProperty(REPLACEMENT_STRATEGY).getValue();
final String evaluateMode = context.getProperty(EVALUATION_MODE).getValue(); final String evaluateMode = context.getProperty(EVALUATION_MODE).getValue();
final int maxBufferSize = context.getProperty(MAX_BUFFER_SIZE).asDataSize(DataUnit.B).intValue();
final byte[] buffer;
if (replacementStrategy.equalsIgnoreCase(regexReplaceValue) || replacementStrategy.equalsIgnoreCase(literalReplaceValue)) {
buffer = new byte[maxBufferSize];
} else {
buffer = null;
}
switch (replacementStrategy) { switch (replacementStrategy) {
case prependValue: case prependValue:
@ -285,12 +275,13 @@ public class ReplaceText extends AbstractProcessor {
&& context.getProperty(REPLACEMENT_VALUE).getValue().isEmpty()) { && context.getProperty(REPLACEMENT_VALUE).getValue().isEmpty()) {
replacementStrategyExecutor = new AlwaysReplace(); replacementStrategyExecutor = new AlwaysReplace();
} else { } else {
replacementStrategyExecutor = new RegexReplace(buffer, context); final String regex = context.getProperty(SEARCH_VALUE).evaluateAttributeExpressions().getValue();
replacementStrategyExecutor = new RegexReplace(regex);
} }
break; break;
case literalReplaceValue: case literalReplaceValue:
replacementStrategyExecutor = new LiteralReplace(buffer); replacementStrategyExecutor = new LiteralReplace();
break; break;
case alwaysReplace: case alwaysReplace:
replacementStrategyExecutor = new AlwaysReplace(); replacementStrategyExecutor = new AlwaysReplace();
@ -382,7 +373,7 @@ public class ReplaceText extends AbstractProcessor {
return value; return value;
} }
private class AlwaysReplace implements ReplacementStrategyExecutor { private static class AlwaysReplace implements ReplacementStrategyExecutor {
@Override @Override
public FlowFile replace(FlowFile flowFile, final ProcessSession session, final ProcessContext context, final String evaluateMode, final Charset charset, final int maxBufferSize) { public FlowFile replace(FlowFile flowFile, final ProcessSession session, final ProcessContext context, final String evaluateMode, final Charset charset, final int maxBufferSize) {
@ -390,12 +381,7 @@ public class ReplaceText extends AbstractProcessor {
final StringBuilder lineEndingBuilder = new StringBuilder(2); final StringBuilder lineEndingBuilder = new StringBuilder(2);
if (evaluateMode.equalsIgnoreCase(ENTIRE_TEXT)) { if (evaluateMode.equalsIgnoreCase(ENTIRE_TEXT)) {
flowFile = session.write(flowFile, new StreamCallback() { flowFile = session.write(flowFile, out -> out.write(replacementValue.getBytes(charset)));
@Override
public void process(final InputStream in, final OutputStream out) throws IOException {
out.write(replacementValue.getBytes(charset));
}
});
} else { } else {
flowFile = session.write(flowFile, new StreamReplaceCallback(charset, maxBufferSize, context.getProperty(LINE_BY_LINE_EVALUATION_MODE).getValue(), flowFile = session.write(flowFile, new StreamReplaceCallback(charset, maxBufferSize, context.getProperty(LINE_BY_LINE_EVALUATION_MODE).getValue(),
((bw, oneLine) -> { ((bw, oneLine) -> {
@ -427,7 +413,7 @@ public class ReplaceText extends AbstractProcessor {
} }
} }
private class PrependReplace implements ReplacementStrategyExecutor { private static class PrependReplace implements ReplacementStrategyExecutor {
@Override @Override
public FlowFile replace(FlowFile flowFile, final ProcessSession session, final ProcessContext context, final String evaluateMode, final Charset charset, final int maxBufferSize) { public FlowFile replace(FlowFile flowFile, final ProcessSession session, final ProcessContext context, final String evaluateMode, final Charset charset, final int maxBufferSize) {
final String replacementValue = context.getProperty(REPLACEMENT_VALUE).evaluateAttributeExpressions(flowFile).getValue(); final String replacementValue = context.getProperty(REPLACEMENT_VALUE).evaluateAttributeExpressions(flowFile).getValue();
@ -454,7 +440,7 @@ public class ReplaceText extends AbstractProcessor {
} }
private class AppendReplace implements ReplacementStrategyExecutor { private static class AppendReplace implements ReplacementStrategyExecutor {
@Override @Override
public FlowFile replace(FlowFile flowFile, final ProcessSession session, final ProcessContext context, final String evaluateMode, final Charset charset, final int maxBufferSize) { public FlowFile replace(FlowFile flowFile, final ProcessSession session, final ProcessContext context, final String evaluateMode, final Charset charset, final int maxBufferSize) {
@ -506,8 +492,7 @@ public class ReplaceText extends AbstractProcessor {
} }
private class RegexReplace implements ReplacementStrategyExecutor { private static class RegexReplace implements ReplacementStrategyExecutor {
private final byte[] buffer;
private final int numCapturingGroups; private final int numCapturingGroups;
private final Map<String, String> additionalAttrs; private final Map<String, String> additionalAttrs;
@ -520,11 +505,8 @@ public class ReplaceText extends AbstractProcessor {
} }
}; };
public RegexReplace(final byte[] buffer, final ProcessContext context) { public RegexReplace(final String regex) {
this.buffer = buffer; numCapturingGroups = Pattern.compile(regex).matcher("").groupCount();
final String regexValue = context.getProperty(SEARCH_VALUE).evaluateAttributeExpressions().getValue();
numCapturingGroups = Pattern.compile(regexValue).matcher("").groupCount();
additionalAttrs = new HashMap<>(numCapturingGroups); additionalAttrs = new HashMap<>(numCapturingGroups);
} }
@ -535,15 +517,13 @@ public class ReplaceText extends AbstractProcessor {
final String searchRegex = context.getProperty(SEARCH_VALUE).evaluateAttributeExpressions(flowFile, quotedAttributeDecorator).getValue(); final String searchRegex = context.getProperty(SEARCH_VALUE).evaluateAttributeExpressions(flowFile, quotedAttributeDecorator).getValue();
final Pattern searchPattern = Pattern.compile(searchRegex); final Pattern searchPattern = Pattern.compile(searchRegex);
final int flowFileSize = (int) flowFile.getSize();
FlowFile updatedFlowFile; FlowFile updatedFlowFile;
if (evaluateMode.equalsIgnoreCase(ENTIRE_TEXT)) { if (evaluateMode.equalsIgnoreCase(ENTIRE_TEXT)) {
session.read(flowFile, new InputStreamCallback() { final int flowFileSize = (int) flowFile.getSize();
@Override final int bufferSize = Math.min(maxBufferSize, flowFileSize);
public void process(final InputStream in) throws IOException { final byte[] buffer = new byte[bufferSize];
StreamUtils.fillBuffer(in, buffer, false);
} session.read(flowFile, in -> StreamUtils.fillBuffer(in, buffer, false));
});
final String contentString = new String(buffer, 0, flowFileSize, charset); final String contentString = new String(buffer, 0, flowFileSize, charset);
additionalAttrs.clear(); additionalAttrs.clear();
@ -571,12 +551,7 @@ public class ReplaceText extends AbstractProcessor {
matcher.appendTail(sb); matcher.appendTail(sb);
final String updatedValue = sb.toString(); final String updatedValue = sb.toString();
updatedFlowFile = session.write(flowFile, new OutputStreamCallback() { updatedFlowFile = session.write(flowFile, out -> out.write(updatedValue.getBytes(charset)));
@Override
public void process(final OutputStream out) throws IOException {
out.write(updatedValue.getBytes(charset));
}
});
} else { } else {
return flowFile; return flowFile;
} }
@ -626,22 +601,18 @@ public class ReplaceText extends AbstractProcessor {
} }
} }
private class LiteralReplace implements ReplacementStrategyExecutor { private static class LiteralReplace implements ReplacementStrategyExecutor {
private final byte[] buffer;
public LiteralReplace(final byte[] buffer) {
this.buffer = buffer;
}
@Override @Override
public FlowFile replace(FlowFile flowFile, final ProcessSession session, final ProcessContext context, final String evaluateMode, final Charset charset, final int maxBufferSize) { public FlowFile replace(FlowFile flowFile, final ProcessSession session, final ProcessContext context, final String evaluateMode, final Charset charset, final int maxBufferSize) {
final String replacementValue = context.getProperty(REPLACEMENT_VALUE).evaluateAttributeExpressions(flowFile).getValue(); final String replacementValue = context.getProperty(REPLACEMENT_VALUE).evaluateAttributeExpressions(flowFile).getValue();
final AttributeValueDecorator quotedAttributeDecorator = Pattern::quote; final AttributeValueDecorator quotedAttributeDecorator = Pattern::quote;
final String searchValue = context.getProperty(SEARCH_VALUE).evaluateAttributeExpressions(flowFile, quotedAttributeDecorator).getValue(); final String searchValue = context.getProperty(SEARCH_VALUE).evaluateAttributeExpressions(flowFile, quotedAttributeDecorator).getValue();
final int flowFileSize = (int) flowFile.getSize();
if (evaluateMode.equalsIgnoreCase(ENTIRE_TEXT)) { if (evaluateMode.equalsIgnoreCase(ENTIRE_TEXT)) {
final int flowFileSize = (int) flowFile.getSize();
final int bufferSize = Math.min(maxBufferSize, flowFileSize);
final byte[] buffer = new byte[bufferSize];
flowFile = session.write(flowFile, new StreamCallback() { flowFile = session.write(flowFile, new StreamCallback() {
@Override @Override
public void process(final InputStream in, final OutputStream out) throws IOException { public void process(final InputStream in, final OutputStream out) throws IOException {
@ -710,7 +681,7 @@ public class ReplaceText extends AbstractProcessor {
} }
private class StreamReplaceCallback implements StreamCallback { private static class StreamReplaceCallback implements StreamCallback {
private final Charset charset; private final Charset charset;
private final int maxBufferSize; private final int maxBufferSize;
private final String lineByLineEvaluationMode; private final String lineByLineEvaluationMode;