You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by ij...@apache.org on 2018/11/09 08:24:02 UTC
nifi git commit: NIFI-5724 make database connection autocommit
configurable
Repository: nifi
Updated Branches:
refs/heads/master d5bce9197 -> 63f55d05b
NIFI-5724 make database connection autocommit configurable
making the database session autocommit value a configurable property
adding custom validation to PutSQL processor so as to disallow 'supports transaction' and 'rollback on failure' to be true when the autocommit value has been set to true
fixing some style issues to conform to standards
This closes #3113.
Signed-off-by: Koji Kawamura <ij...@apache.org>
Project: http://git-wip-us.apache.org/repos/asf/nifi/repo
Commit: http://git-wip-us.apache.org/repos/asf/nifi/commit/63f55d05
Tree: http://git-wip-us.apache.org/repos/asf/nifi/tree/63f55d05
Diff: http://git-wip-us.apache.org/repos/asf/nifi/diff/63f55d05
Branch: refs/heads/master
Commit: 63f55d05b4f3f83b9e9f2206f4129ae7a4ade569
Parents: d5bce91
Author: Vish Uma <vi...@gmail.com>
Authored: Fri Oct 26 15:32:46 2018 -0400
Committer: Koji Kawamura <ij...@apache.org>
Committed: Fri Nov 9 17:07:27 2018 +0900
----------------------------------------------------------------------
.../apache/nifi/processors/standard/PutSQL.java | 53 ++++++++++++++++++--
1 file changed, 49 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/nifi/blob/63f55d05/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutSQL.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutSQL.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutSQL.java
index 9957c2e..38134c2 100644
--- a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutSQL.java
+++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutSQL.java
@@ -28,6 +28,8 @@ import org.apache.nifi.annotation.documentation.SeeAlso;
import org.apache.nifi.annotation.documentation.Tags;
import org.apache.nifi.annotation.lifecycle.OnScheduled;
import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.ValidationContext;
+import org.apache.nifi.components.ValidationResult;
import org.apache.nifi.dbcp.DBCPService;
import org.apache.nifi.expression.ExpressionLanguageScope;
import org.apache.nifi.flowfile.FlowFile;
@@ -64,6 +66,7 @@ import java.sql.SQLNonTransientException;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.BitSet;
+import java.util.Collection;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
@@ -74,6 +77,7 @@ import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.function.BiFunction;
+import static java.lang.String.format;
import static org.apache.nifi.processor.util.pattern.ExceptionHandler.createOnError;
@SupportsBatching
@@ -134,6 +138,14 @@ public class PutSQL extends AbstractSessionFactoryProcessor {
.expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
.build();
+ static final PropertyDescriptor AUTO_COMMIT = new PropertyDescriptor.Builder()
+ .name("database-session-autocommit")
+ .displayName("Database Session AutoCommit")
+ .description("The autocommit mode to set on the database connection being used.")
+ .allowableValues("true", "false")
+ .defaultValue("false")
+ .build();
+
static final PropertyDescriptor SUPPORT_TRANSACTIONS = new PropertyDescriptor.Builder()
.name("Support Fragmented Transactions")
.description("If true, when a FlowFile is consumed by this Processor, the Processor will first check the fragment.identifier and fragment.count attributes of that FlowFile. "
@@ -189,6 +201,7 @@ public class PutSQL extends AbstractSessionFactoryProcessor {
properties.add(CONNECTION_POOL);
properties.add(SQL_STATEMENT);
properties.add(SUPPORT_TRANSACTIONS);
+ properties.add(AUTO_COMMIT);
properties.add(TRANSACTION_TIMEOUT);
properties.add(BATCH_SIZE);
properties.add(OBTAIN_GENERATED_KEYS);
@@ -197,6 +210,34 @@ public class PutSQL extends AbstractSessionFactoryProcessor {
}
@Override
+ protected final Collection<ValidationResult> customValidate(ValidationContext context) {
+ final Collection<ValidationResult> results = new ArrayList<>();
+ final String support_transactions = context.getProperty(SUPPORT_TRANSACTIONS).getValue();
+ final String rollback_on_failure = context.getProperty(RollbackOnFailure.ROLLBACK_ON_FAILURE).getValue();
+ final String auto_commit = context.getProperty(AUTO_COMMIT).getValue();
+
+ if(auto_commit.equalsIgnoreCase("true")) {
+ if(support_transactions.equalsIgnoreCase("true")) {
+ results.add(new ValidationResult.Builder()
+ .subject(SUPPORT_TRANSACTIONS.getDisplayName())
+ .explanation(format("'%s' cannot be set to 'true' when '%s' is also set to 'true'."
+ + "Transactions for batch updates cannot be supported when auto commit is set to 'true'",
+ SUPPORT_TRANSACTIONS.getDisplayName(), AUTO_COMMIT.getDisplayName()))
+ .build());
+ }
+ if(rollback_on_failure.equalsIgnoreCase("true")) {
+ results.add(new ValidationResult.Builder()
+ .subject(RollbackOnFailure.ROLLBACK_ON_FAILURE.getDisplayName())
+ .explanation(format("'%s' cannot be set to 'true' when '%s' is also set to 'true'."
+ + "Transaction rollbacks for batch updates cannot be supported when auto commit is set to 'true'",
+ RollbackOnFailure.ROLLBACK_ON_FAILURE.getDisplayName(), AUTO_COMMIT.getDisplayName()))
+ .build());
+ }
+ }
+ return results;
+ }
+
+ @Override
public Set<Relationship> getRelationships() {
final Set<Relationship> rels = new HashSet<>();
rels.add(REL_SUCCESS);
@@ -239,7 +280,10 @@ public class PutSQL extends AbstractSessionFactoryProcessor {
.getConnection(ff == null ? Collections.emptyMap() : ff.getAttributes());
try {
fc.originalAutoCommit = connection.getAutoCommit();
- connection.setAutoCommit(false);
+ final boolean autocommit = c.getProperty(AUTO_COMMIT).asBoolean();
+ if(fc.originalAutoCommit != autocommit) {
+ connection.setAutoCommit(autocommit);
+ }
} catch (SQLException e) {
throw new ProcessException("Failed to disable auto commit due to " + e, e);
}
@@ -521,9 +565,10 @@ public class PutSQL extends AbstractSessionFactoryProcessor {
process.cleanup((c, s, fc, conn) -> {
// make sure that we try to set the auto commit back to whatever it was.
- if (fc.originalAutoCommit) {
+ final boolean autocommit = c.getProperty(AUTO_COMMIT).asBoolean();
+ if (fc.originalAutoCommit != autocommit) {
try {
- conn.setAutoCommit(true);
+ conn.setAutoCommit(fc.originalAutoCommit);
} catch (final SQLException se) {
getLogger().warn("Failed to reset autocommit due to {}", new Object[]{se});
}
@@ -670,7 +715,7 @@ public class PutSQL extends AbstractSessionFactoryProcessor {
int selectedNumFragments = 0;
final BitSet bitSet = new BitSet();
- BiFunction<String, Object[], IllegalArgumentException> illegal = (s, objects) -> new IllegalArgumentException(String.format(s, objects));
+ BiFunction<String, Object[], IllegalArgumentException> illegal = (s, objects) -> new IllegalArgumentException(format(s, objects));
for (final FlowFile flowFile : flowFiles) {
final String fragmentCount = flowFile.getAttribute(FRAGMENT_COUNT_ATTR);