Merge branch 'master' of https://github.com/apache/nifi into aws-web-api

Author: ottobackwards

nifi-commons/nifi-record/src/main/java/org/apache/nifi/serialization/record/ResultSetRecordSet.java

@@ -350,6 +350,8 @@ private static RecordFieldType getFieldType(final int sqlType) {
                 return RecordFieldType.TIME;
             case Types.TIMESTAMP:
             case Types.TIMESTAMP_WITH_TIMEZONE:
+            case -101: // Oracle's TIMESTAMP WITH TIME ZONE
+            case -102: // Oracle's TIMESTAMP WITH LOCAL TIME ZONE
                 return RecordFieldType.TIMESTAMP;
         }
 

nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/leader/election/CuratorLeaderElectionManager.java

@@ -16,8 +16,6 @@
  */
 package org.apache.nifi.controller.leader.election;
 
-import java.util.HashMap;
-import java.util.Map;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.curator.RetryPolicy;
 import org.apache.curator.framework.CuratorFramework;
@@ -36,6 +34,9 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.util.HashMap;
+import java.util.Map;
+
 public class CuratorLeaderElectionManager implements LeaderElectionManager {
 
     private static final Logger logger = LoggerFactory.getLogger(CuratorLeaderElectionManager.class);
@@ -112,7 +113,7 @@ public synchronized void register(final String roleName, final LeaderElectionSta
         final boolean isParticipant = participantId != null && !participantId.trim().isEmpty();
 
         if (!isStopped()) {
-            final ElectionListener electionListener = new ElectionListener(roleName, listener);
+            final ElectionListener electionListener = new ElectionListener(roleName, listener, participantId);
             final LeaderSelector leaderSelector = new LeaderSelector(curatorClient, leaderPath, leaderElectionMonitorEngine, electionListener);
             if (isParticipant) {
                 leaderSelector.autoRequeue();
@@ -358,12 +359,14 @@ private class ElectionListener extends LeaderSelectorListenerAdapter implements
 
         private final String roleName;
         private final LeaderElectionStateChangeListener listener;
+        private final String participantId;
 
         private volatile boolean leader;
 
-        public ElectionListener(final String roleName, final LeaderElectionStateChangeListener listener) {
+        public ElectionListener(final String roleName, final LeaderElectionStateChangeListener listener, final String participantId) {
             this.roleName = roleName;
             this.listener = listener;
+            this.participantId = participantId;
         }
 
         public boolean isLeader() {
@@ -373,9 +376,37 @@ public boolean isLeader() {
         @Override
         public void stateChanged(final CuratorFramework client, final ConnectionState newState) {
             logger.info("{} Connection State changed to {}", this, newState.name());
+
+            if (newState == ConnectionState.SUSPENDED || newState == ConnectionState.LOST) {
+                if (leader == true) {
+                    logger.info("Because Connection State was changed to {}, will relinquish leadership for role '{}'", newState, roleName);
+                }
+
+                leader = false;
+            }
+
             super.stateChanged(client, newState);
         }
 
+        /**
+         * Reach out to ZooKeeper to verify that this node still is the leader. We do this because at times, a node will lose
+         * its position as leader but the Curator client will fail to notify us, perhaps due to network failure, etc.
+         *
+         * @return <code>true</code> if this node is still the elected leader according to ZooKeeper, false otherwise
+         */
+        private boolean verifyLeader() {
+            final String leader = getLeader(roleName);
+            if (leader == null) {
+                logger.debug("Reached out to ZooKeeper to determine which node is the elected leader for Role '{}' but found that there is no leader.", roleName);
+                return false;
+            }
+
+            final boolean match = leader.equals(participantId);
+            logger.debug("Reached out to ZooKeeper to determine which node is the elected leader for Role '{}'. Elected Leader = '{}', Participant ID = '{}', This Node Elected = {}",
+                roleName, leader, participantId, match);
+            return match;
+        }
+
         @Override
         public void takeLeadership(final CuratorFramework client) throws Exception {
             leader = true;
@@ -396,14 +427,40 @@ public void takeLeadership(final CuratorFramework client) throws Exception {
             // Curator API states that we lose the leadership election when we return from this method,
             // so we will block as long as we are not interrupted or closed. Then, we will set leader to false.
             try {
-                while (!isStopped()) {
+                int failureCount = 0;
+                int loopCount = 0;
+                while (!isStopped() && leader) {
                     try {
                         Thread.sleep(100L);
                     } catch (final InterruptedException ie) {
                         logger.info("{} has been interrupted; no longer leader for role '{}'", this, roleName);
                         Thread.currentThread().interrupt();
                         return;
                     }
+
+                    if (leader && ++loopCount % 50 == 0) {
+                        // While Curator is supposed to interrupt this thread when we are no longer the leader, we have occasionally
+                        // seen occurrences where the thread does not get interrupted. As a result, we will reach out to ZooKeeper
+                        // periodically to determine whether or not this node is still the elected leader.
+                        try {
+                            final boolean stillLeader = verifyLeader();
+                            failureCount = 0; // we got a response, so we were successful in communicating with zookeeper. Set failureCount back to 0.
+
+                            if (!stillLeader) {
+                                logger.info("According to ZooKeeper, this node is no longer the leader for Role '{}'. Will relinquish leadership.", roleName);
+                                break;
+                            }
+                        } catch (final Exception e) {
+                            failureCount++;
+                            if (failureCount > 1) {
+                                logger.warn("Attempted to reach out to ZooKeeper to verify that this node still is the elected leader for Role '{}' "
+                                    + "but failed to communicate with ZooKeeper. This is the second failed attempt, so will relinquish leadership of this role.", roleName, e);
+                            } else {
+                                logger.warn("Attempted to reach out to ZooKeeper to verify that this node still is the elected leader for Role '{}' "
+                                    + "but failed to communicate with ZooKeeper. Will wait a bit and attempt to communicate with ZooKeeper again before relinquishing this role.", roleName, e);
+                            }
+                        }
+                    }
                 }
             } finally {
                 leader = false;

nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/dao/impl/StandardConnectionDAO.java

@@ -567,6 +567,7 @@ public Connection updateConnection(final ConnectionDTO connectionDTO) {
 
         // configure the connection
         configureConnection(connection, connectionDTO);
+        group.onComponentModified();
 
         // update the relationships if necessary
         if (!newProcessorRelationships.isEmpty()) {

nifi-nar-bundles/nifi-site-to-site-reporting-bundle/nifi-site-to-site-reporting-task/src/main/java/org/apache/nifi/reporting/SiteToSiteStatusReportingTask.java

@@ -339,6 +339,10 @@ private void serializeConnectionStatus(final JsonArrayBuilder arrayBuilder, fina
                     componentType, componentName);
 
             addField(builder, "componentId", status.getId());
+            addField(builder, "sourceId", status.getSourceId());
+            addField(builder, "sourceName", status.getSourceName());
+            addField(builder, "destinationId", status.getDestinationId());
+            addField(builder, "destinationName", status.getDestinationName());
             addField(builder, "maxQueuedBytes", status.getMaxQueuedBytes());
             addField(builder, "maxQueuedCount", status.getMaxQueuedCount());
             addField(builder, "queuedBytes", status.getQueuedBytes());

nifi-nar-bundles/nifi-site-to-site-reporting-bundle/nifi-site-to-site-reporting-task/src/test/java/org/apache/nifi/reporting/TestSiteToSiteStatusReportingTask.java

@@ -31,6 +31,7 @@
 import java.util.UUID;
 
 import javax.json.Json;
+import javax.json.JsonObject;
 import javax.json.JsonReader;
 import javax.json.JsonString;
 
@@ -144,8 +145,11 @@ public void testConnectionStatus() throws IOException, InitializationException {
 
         final String msg = new String(task.dataSent.get(0), StandardCharsets.UTF_8);
         JsonReader jsonReader = Json.createReader(new ByteArrayInputStream(msg.getBytes()));
-        JsonString backpressure = jsonReader.readArray().getJsonObject(0).getJsonString("isBackPressureEnabled");
+        JsonObject object = jsonReader.readArray().getJsonObject(0);
+        JsonString backpressure = object.getJsonString("isBackPressureEnabled");
+        JsonString source = object.getJsonString("sourceName");
         assertEquals("true", backpressure.getString());
+        assertEquals("source", source.getString());
     }
 
     @Test
@@ -318,6 +322,10 @@ public static ConnectionStatus generateConnectionStatus(String id, String namePr
         cStatus.setOutputCount(7);
         cStatus.setQueuedBytes(8l);
         cStatus.setQueuedCount(9);
+        cStatus.setSourceId(id);
+        cStatus.setSourceName("source");
+        cStatus.setDestinationId(id);
+        cStatus.setDestinationName("destination");
 
         return cStatus;
     }

nifi-nar-bundles/nifi-solr-bundle/nifi-solr-processors/src/main/java/org/apache/nifi/processors/solr/GetSolr.java

@@ -33,6 +33,7 @@
 import java.util.Map;
 import java.util.Set;
 import java.util.TimeZone;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 
 import org.apache.nifi.annotation.behavior.InputRequirement;
@@ -61,6 +62,7 @@
 import org.apache.nifi.serialization.RecordSetWriterFactory;
 import org.apache.nifi.serialization.record.RecordSchema;
 import org.apache.nifi.serialization.record.RecordSet;
+import org.apache.nifi.util.StopWatch;
 import org.apache.nifi.util.StringUtils;
 
 import org.apache.solr.client.solrj.SolrQuery;
@@ -336,6 +338,8 @@ public void onTrigger(final ProcessContext context, final ProcessSession session
             solrQuery.setParam("sort", sortClause.toString());
 
             while (continuePaging.get()) {
+                StopWatch timer = new StopWatch(true);
+
                 final QueryRequest req = new QueryRequest(solrQuery);
                 if (isBasicAuthEnabled()) {
                     req.setBasicAuthCredentials(getUsername(), getPassword());
@@ -385,8 +389,19 @@ public void process(final OutputStream out) throws IOException {
                                 }
                             }
                         });
+
                         flowFile = session.putAttribute(flowFile, CoreAttributes.MIME_TYPE.key(), mimeType.toString());
                     }
+
+                    timer.stop();
+                    StringBuilder transitUri = new StringBuilder("solr://");
+                    transitUri.append(getSolrLocation());
+                    if (getSolrLocation().equals(SolrUtils.SOLR_TYPE_CLOUD.getValue())) {
+                        transitUri.append(":").append(context.getProperty(COLLECTION).evaluateAttributeExpressions().getValue());
+                    }
+                    final long duration = timer.getDuration(TimeUnit.MILLISECONDS);
+                    session.getProvenanceReporter().receive(flowFile, transitUri.toString(), duration);
+
                     session.transfer(flowFile, REL_SUCCESS);
                 }
                 continuePaging.set(response.getResults().size() == Integer.parseInt(context.getProperty(BATCH_SIZE).getValue()));

nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/util/JdbcCommon.java

@@ -41,6 +41,7 @@
 import static java.sql.Types.SMALLINT;
 import static java.sql.Types.TIME;
 import static java.sql.Types.TIMESTAMP;
+import static java.sql.Types.TIMESTAMP_WITH_TIMEZONE;
 import static java.sql.Types.TINYINT;
 import static java.sql.Types.VARBINARY;
 import static java.sql.Types.VARCHAR;
@@ -356,7 +357,28 @@ public static long convertToAvroStream(final ResultSet rs, final OutputStream ou
                         continue;
                     }
 
-                    final Object value = rs.getObject(i);
+                    Object value;
+
+                    // If a Timestamp type, try getTimestamp() rather than getObject()
+                    if (javaSqlType == TIMESTAMP
+                            || javaSqlType == TIMESTAMP_WITH_TIMEZONE
+                            // The following are Oracle-specific codes for TIMESTAMP WITH TIME ZONE and TIMESTAMP WITH LOCAL TIME ZONE. This would be better
+                            // located in the DatabaseAdapter interfaces, but some processors (like ExecuteSQL) use this method but don't specify a DatabaseAdapter.
+                            || javaSqlType == -101
+                            || javaSqlType == -102) {
+                        try {
+                            value = rs.getTimestamp(i);
+                            // Some drivers (like Derby) return null for getTimestamp() but return a Timestamp object in getObject()
+                            if (value == null) {
+                                value = rs.getObject(i);
+                            }
+                        } catch (Exception e) {
+                            // The cause of the exception is not known, but we'll fall back to call getObject() and handle any "real" exception there
+                            value = rs.getObject(i);
+                        }
+                    } else {
+                        value = rs.getObject(i);
+                    }
 
                     if (value == null) {
                         rec.put(i - 1, null);
@@ -603,6 +625,9 @@ public static Schema createSchema(final ResultSet rs, AvroConversionOptions opti
                     break;
 
                 case TIMESTAMP:
+                case TIMESTAMP_WITH_TIMEZONE:
+                case -101: // Oracle's TIMESTAMP WITH TIME ZONE
+                case -102: // Oracle's TIMESTAMP WITH LOCAL TIME ZONE
                     addNullableField(builder, columnName,
                             u -> options.useLogicalTypes
                                     ? u.type(LogicalTypes.timestampMillis().addToSchema(SchemaBuilder.builder().longType()))