turcsanyip commented on a change in pull request #4839:
URL: https://github.com/apache/nifi/pull/4839#discussion_r581216003



##########
File path: 
nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/status/history/storage/QuestDbComponentStatusStorage.java
##########
@@ -0,0 +1,115 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.controller.status.history.storage;
+
+import io.questdb.cairo.sql.Record;
+import org.apache.commons.lang3.time.DateUtils;
+import org.apache.commons.lang3.time.FastDateFormat;
+import org.apache.commons.math3.util.Pair;
+import org.apache.nifi.controller.status.history.ComponentDetailsStorage;
+import org.apache.nifi.controller.status.history.MetricDescriptor;
+import org.apache.nifi.controller.status.history.StandardStatusHistory;
+import org.apache.nifi.controller.status.history.StatusHistory;
+import org.apache.nifi.controller.status.history.StatusSnapshot;
+import org.apache.nifi.controller.status.history.questdb.QuestDbContext;
+import 
org.apache.nifi.controller.status.history.questdb.QuestDbEntityReadingTemplate;
+import 
org.apache.nifi.controller.status.history.questdb.QuestDbEntityWritingTemplate;
+import 
org.apache.nifi.controller.status.history.questdb.QuestDbStatusSnapshotMapper;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Date;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.function.Function;
+
+/**
+ * Component specific implementation of the {@link ComponentStatusStorage}.
+ *
+ * @param <T> Component status entry type.
+ */
+abstract class QuestDbComponentStatusStorage<T> implements 
ComponentStatusStorage<T> {

Review comment:
       It might be worth separating these `QuestDb***StatusStorage` 
implementation classes to their own subpackage.

##########
File path: 
nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/status/history/storage/StorageStatusReadingTemplate.java
##########
@@ -0,0 +1,136 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.controller.status.history.storage;
+
+import io.questdb.cairo.sql.Record;
+import io.questdb.cairo.sql.RecordCursor;
+import org.apache.nifi.controller.status.NodeStatus;
+import org.apache.nifi.controller.status.history.MetricDescriptor;
+import org.apache.nifi.controller.status.history.StandardMetricDescriptor;
+import 
org.apache.nifi.controller.status.history.questdb.QuestDbReadingTemplate;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+public class StorageStatusReadingTemplate extends 
QuestDbReadingTemplate<Map<Long, Map<StandardMetricDescriptor<NodeStatus>, 
Long>>> {
+    private static final String STORAGE_FREE_DESCRIPTION = "The usable space 
available for use by the underlying storage mechanism.";
+    private static final String STORAGE_USED_DESCRIPTION = "The space in use 
on the underlying storage mechanism";
+
+    private static final String STORAGE_READING_QUERY =
+            "SELECT * FROM storageStatus " +
+                    "WHERE capturedAt > to_timestamp('%s', '" + 
StatusStorage.CAPTURE_DATE_FORMAT + "') " +
+                    "AND capturedAt < to_timestamp('%s', '" + 
StatusStorage.CAPTURE_DATE_FORMAT + "') " +
+                    "ORDER BY capturedAt ASC";
+
+    public StorageStatusReadingTemplate() {
+        super(STORAGE_READING_QUERY, e -> Collections.emptyMap());
+    }
+
+    @Override
+    protected Map<Long, Map<StandardMetricDescriptor<NodeStatus>, Long>> 
processResult(final RecordCursor cursor) {
+        final Map<Long, Map<StandardMetricDescriptor<NodeStatus>, Long>> 
result = new HashMap<>();
+
+        int storageNumber = 1;
+
+        while (cursor.hasNext()) {
+            final Record record = cursor.getRecord();
+            final long createdAt = 
TimeUnit.MICROSECONDS.toMillis(record.getTimestamp(0));
+            final short type = record.getShort(2);
+            final CharSequence name = record.getSym(1);
+
+            if (!result.containsKey(createdAt)) {
+                result.put(createdAt, new HashMap<>());
+            }
+
+            result.get(createdAt).put(getDescriptor(
+                    QuestDbNodeStatusStorage.getMetrics().size() + 
result.get(createdAt).size(),
+                    getField(type, storageNumber, StorageMetric.FREE),
+                    getLabel(type, name, StorageMetric.FREE),
+                    STORAGE_FREE_DESCRIPTION
+            ), record.getLong(3));
+            result.get(createdAt).put(getDescriptor(
+                    QuestDbNodeStatusStorage.getMetrics().size() + 
result.get(createdAt).size(),
+                    getField(type, storageNumber, StorageMetric.USED),
+                    getLabel(type, name, StorageMetric.USED),
+                    STORAGE_USED_DESCRIPTION
+            ), record.getLong(4));
+            storageNumber++;
+        }
+
+        return result;
+    }
+
+    private StandardMetricDescriptor<NodeStatus> getDescriptor(final int 
ordinal, final String field, final String label, final String description) {
+        return new StandardMetricDescriptor<>(() -> ordinal, field, label, 
STORAGE_FREE_DESCRIPTION, MetricDescriptor.Formatter.DATA_SIZE, v -> 0L
+        );

Review comment:
       Minor formatting: `);` should be in the same line.

##########
File path: 
nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/status/history/EmbeddedQuestDbStatusHistoryRepository.java
##########
@@ -0,0 +1,263 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.controller.status.history;
+
+import io.questdb.MessageBusImpl;
+import io.questdb.cairo.CairoConfiguration;
+import io.questdb.cairo.CairoEngine;
+import io.questdb.cairo.DefaultCairoConfiguration;
+import org.apache.commons.lang3.concurrent.BasicThreadFactory;
+import org.apache.commons.math3.util.Pair;
+import org.apache.nifi.controller.status.ConnectionStatus;
+import org.apache.nifi.controller.status.NodeStatus;
+import org.apache.nifi.controller.status.ProcessGroupStatus;
+import org.apache.nifi.controller.status.ProcessorStatus;
+import org.apache.nifi.controller.status.RemoteProcessGroupStatus;
+import org.apache.nifi.controller.status.history.questdb.QuestDbContext;
+import 
org.apache.nifi.controller.status.history.questdb.QuestDbDatabaseManager;
+import 
org.apache.nifi.controller.status.history.storage.BufferedWriterFlushWorker;
+import 
org.apache.nifi.controller.status.history.storage.BufferedWriterForStatusStorage;
+import 
org.apache.nifi.controller.status.history.storage.ComponentStatusStorage;
+import 
org.apache.nifi.controller.status.history.storage.GarbageCollectionStatusStorage;
+import org.apache.nifi.controller.status.history.storage.NodeStatusStorage;
+import 
org.apache.nifi.controller.status.history.storage.ProcessorStatusStorage;
+import 
org.apache.nifi.controller.status.history.storage.QuestDbConnectionStatusStorage;
+import 
org.apache.nifi.controller.status.history.storage.QuestDbGarbageCollectionStatusStorage;
+import 
org.apache.nifi.controller.status.history.storage.QuestDbNodeStatusStorage;
+import 
org.apache.nifi.controller.status.history.storage.QuestDbProcessGroupStatusStorage;
+import 
org.apache.nifi.controller.status.history.storage.QuestDbProcessorStatusStorage;
+import 
org.apache.nifi.controller.status.history.storage.QuestDbRemoteProcessGroupStatusStorage;
+import org.apache.nifi.util.NiFiProperties;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.nio.file.Path;
+import java.util.Arrays;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+public class EmbeddedQuestDbStatusHistoryRepository implements 
StatusHistoryRepository {
+    private static final Logger LOGGER = 
LoggerFactory.getLogger(EmbeddedQuestDbStatusHistoryRepository.class);
+    private static final int PERSIST_BATCH_SIZE = 1000;
+    private static final long PERSIST_FREQUENCY = 
TimeUnit.MILLISECONDS.convert(5, TimeUnit.SECONDS);
+    private static final long ROLL_FREQUENCY = 
TimeUnit.MILLISECONDS.convert(10, TimeUnit.MINUTES);
+
+
+    private final InMemoryComponentDetailsStorage componentDetailsProvider = 
new InMemoryComponentDetailsStorage();
+    private final ScheduledExecutorService scheduledExecutorService = Executors
+            .newScheduledThreadPool(3, new 
BasicThreadFactory.Builder().namingPattern("EmbeddedQuestDbStatusHistoryRepositoryWorker-%d").build());
+
+    private final QuestDbContext dbContext;
+    private final long persistFrequency;
+    private final int daysToKeepNodeData;
+    private final int daysToKeepComponentData;
+
+    private final ProcessorStatusStorage processorStatusStorage;
+    private final ComponentStatusStorage<ConnectionStatus> 
connectionStatusStorage;
+    private final ComponentStatusStorage<ProcessGroupStatus> 
processGroupStatusStorage;
+    private final ComponentStatusStorage<RemoteProcessGroupStatus> 
remoteProcessGroupStatusStorage;
+    private final NodeStatusStorage nodeStatusStorage;
+    private final GarbageCollectionStatusStorage 
garbageCollectionStatusStorage;
+
+    private final BufferedWriterForStatusStorage<ProcessorStatus> 
processorStatusWriter;
+    private final BufferedWriterForStatusStorage<ConnectionStatus> 
connectionStatusWriter;
+    private final BufferedWriterForStatusStorage<ProcessGroupStatus> 
processGroupStatusWriter;
+    private final BufferedWriterForStatusStorage<RemoteProcessGroupStatus> 
remoteProcessGroupStatusWriter;
+    private final BufferedWriterForStatusStorage<NodeStatus> nodeStatusWriter;
+    private final BufferedWriterForStatusStorage<GarbageCollectionStatus> 
garbageCollectionStatusWriter;
+
+    /**
+     * Default no args constructor for service loading only
+     */
+    public EmbeddedQuestDbStatusHistoryRepository() {
+        dbContext = null;
+        persistFrequency = PERSIST_FREQUENCY;
+        daysToKeepNodeData = -1;
+        daysToKeepComponentData = -1;
+
+        processorStatusStorage = null;
+        connectionStatusStorage = null;
+        processGroupStatusStorage = null;
+        remoteProcessGroupStatusStorage = null;
+        nodeStatusStorage = null;
+        garbageCollectionStatusStorage = null;
+
+        processorStatusWriter = null;
+        connectionStatusWriter = null;
+        processGroupStatusWriter = null;
+        remoteProcessGroupStatusWriter = null;
+        nodeStatusWriter = null;
+        garbageCollectionStatusWriter = null;
+    }
+
+    public EmbeddedQuestDbStatusHistoryRepository(final NiFiProperties 
niFiProperties) {
+        this(niFiProperties, PERSIST_FREQUENCY);
+    }
+
+    EmbeddedQuestDbStatusHistoryRepository(final NiFiProperties 
niFiProperties, final long persistFrequency) {
+        final Path persistLocation = 
niFiProperties.getQuestDbStatusRepositoryPath();
+        final CairoConfiguration configuration = new 
DefaultCairoConfiguration(persistLocation.toString());
+        QuestDbDatabaseManager.checkDatabaseStatus(persistLocation);
+
+        this.persistFrequency = persistFrequency;
+        daysToKeepNodeData = getDaysToKeepNodeData(niFiProperties);
+        daysToKeepComponentData = getDaysToKeepComponentData(niFiProperties);
+        dbContext = new QuestDbContext(new CairoEngine(configuration), new 
MessageBusImpl());
+
+        nodeStatusStorage = new QuestDbNodeStatusStorage(dbContext);
+        garbageCollectionStatusStorage = new 
QuestDbGarbageCollectionStatusStorage(dbContext);
+        processorStatusStorage = new QuestDbProcessorStatusStorage(dbContext, 
componentDetailsProvider);
+        connectionStatusStorage = new 
QuestDbConnectionStatusStorage(dbContext, componentDetailsProvider);
+        processGroupStatusStorage = new 
QuestDbProcessGroupStatusStorage(dbContext, componentDetailsProvider);
+        remoteProcessGroupStatusStorage = new 
QuestDbRemoteProcessGroupStatusStorage(dbContext, componentDetailsProvider);
+
+        nodeStatusWriter = new 
BufferedWriterForStatusStorage<>(nodeStatusStorage, PERSIST_BATCH_SIZE);
+        garbageCollectionStatusWriter = new 
BufferedWriterForStatusStorage<>(garbageCollectionStatusStorage, 
PERSIST_BATCH_SIZE);
+        processorStatusWriter = new 
BufferedWriterForStatusStorage<>(processorStatusStorage, PERSIST_BATCH_SIZE);
+        connectionStatusWriter = new 
BufferedWriterForStatusStorage<>(connectionStatusStorage, PERSIST_BATCH_SIZE);
+        processGroupStatusWriter = new 
BufferedWriterForStatusStorage<>(processGroupStatusStorage, PERSIST_BATCH_SIZE);
+        remoteProcessGroupStatusWriter = new 
BufferedWriterForStatusStorage<>(remoteProcessGroupStatusStorage, 
PERSIST_BATCH_SIZE);
+    }
+
+    @Override
+    public void start() {
+        LOGGER.debug("Starting status history repository");
+
+        final EmbeddedQuestDbRolloverHandler nodeRolloverHandler = new 
EmbeddedQuestDbRolloverHandler(QuestDbDatabaseManager.getNodeTableNames(), 
daysToKeepNodeData, dbContext);
+        final EmbeddedQuestDbRolloverHandler componentRolloverHandler = new 
EmbeddedQuestDbRolloverHandler(QuestDbDatabaseManager.getComponentTableNames(), 
daysToKeepComponentData, dbContext);
+        final BufferedWriterFlushWorker writer = new 
BufferedWriterFlushWorker(Arrays.asList(
+            nodeStatusWriter,
+            garbageCollectionStatusWriter,
+            processorStatusWriter,
+            connectionStatusWriter,
+            processGroupStatusWriter,
+            remoteProcessGroupStatusWriter
+        ));
+
+        scheduledExecutorService.scheduleWithFixedDelay(nodeRolloverHandler, 
0, ROLL_FREQUENCY, TimeUnit.MILLISECONDS);
+        
scheduledExecutorService.scheduleWithFixedDelay(componentRolloverHandler, 0, 
ROLL_FREQUENCY, TimeUnit.MILLISECONDS);
+        scheduledExecutorService.scheduleWithFixedDelay(writer, 0, 
persistFrequency, TimeUnit.MILLISECONDS);
+
+        LOGGER.debug("Status history repository is started");
+    }
+
+    @Override
+    public void shutdown() {
+        LOGGER.debug("Component status repository is shutting down");

Review comment:
       Minor: not component specific.

##########
File path: 
nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/status/history/EmbeddedQuestDbRolloverHandler.java
##########
@@ -0,0 +1,121 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.controller.status.history;
+
+import io.questdb.cairo.sql.Record;
+import io.questdb.cairo.sql.RecordCursor;
+import io.questdb.cairo.sql.RecordCursorFactory;
+import io.questdb.griffin.SqlCompiler;
+import io.questdb.griffin.SqlExecutionContext;
+import org.apache.commons.lang3.time.FastDateFormat;
+import org.apache.nifi.controller.status.history.questdb.QuestDbContext;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * QuestDB does not provide the possibility for deleting individual lines. 
Instead there is the option to drop older
+ * partitions. In order to clean up older status information, the partitions 
are outside of the scope of data we intend
+ * to keep will be deleted.
+ */
+public class EmbeddedQuestDbRolloverHandler implements Runnable {

Review comment:
       Not sure but it is rather "purge" than "rollover" what this class is 
doing.

##########
File path: 
nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/status/history/storage/BufferedWriterFlushWorker.java
##########
@@ -0,0 +1,42 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.controller.status.history.storage;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class BufferedWriterFlushWorker implements Runnable {
+    private static final Logger LOGGER = 
LoggerFactory.getLogger(BufferedWriterFlushWorker.class);
+
+    private final List<BufferedWriter<?>> bufferedWriterList = new 
ArrayList<>();
+
+    public BufferedWriterFlushWorker(final List<BufferedWriter<?>> 
bufferedWriterList) {
+        this.bufferedWriterList.addAll(bufferedWriterList);
+    }
+
+    @Override
+    public void run() {
+        try {
+            bufferedWriterList.forEach(statusRepository -> 
statusRepository.flush());

Review comment:
       Minor: it is not a `statusRepository` specific function.

##########
File path: 
nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-resources/src/main/resources/conf/nifi.properties
##########
@@ -120,11 +120,18 @@ 
nifi.provenance.repository.concurrent.merge.threads=${nifi.provenance.repository
 # Volatile Provenance Respository Properties
 
nifi.provenance.repository.buffer.size=${nifi.provenance.repository.buffer.size}
 
-# Component Status Repository
+# Component and Node Status History Repository
 
nifi.components.status.repository.implementation=${nifi.components.status.repository.implementation}
+
+# Volatile Status History Repository Properties
 
nifi.components.status.repository.buffer.size=${nifi.components.status.repository.buffer.size}
 
nifi.components.status.snapshot.frequency=${nifi.components.status.snapshot.frequency}
 
+# QuestDB Status History Repository Properties
+# 
nifi.status.repository.questdb.persist.node.days=${nifi.status.repository.questdb.persist.node.days}
+# 
nifi.status.repository.questdb.persist.component.days=${nifi.status.repository.questdb.persist.component.days}
+# 
nifi.status.repository.questdb.persist.location=${nifi.status.repository.questdb.persist.location}

Review comment:
       I would consider to add these properties uncommented. In that way it 
would be clear which values are in effect. 

##########
File path: 
nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/status/history/questdb/QuestDbDatabaseManager.java
##########
@@ -0,0 +1,180 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.controller.status.history.questdb;
+
+import io.questdb.MessageBusImpl;
+import io.questdb.cairo.CairoConfiguration;
+import io.questdb.cairo.CairoEngine;
+import io.questdb.cairo.DefaultCairoConfiguration;
+import io.questdb.griffin.SqlCompiler;
+import io.questdb.griffin.SqlExecutionContext;
+import io.questdb.griffin.SqlExecutionContextImpl;
+import org.apache.nifi.util.FileUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.nio.file.Path;
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * The database manager is responsible for checking and maintaining the health 
of the database during startup.
+ */
+public final class QuestDbDatabaseManager {
+    private enum DatabaseStatus {
+        HEALTHY, NON_EXISTING, CORRUPTED;
+    }
+
+    private static final Logger LOGGER = 
LoggerFactory.getLogger(QuestDbQueries.class);

Review comment:
       Wrong class for logger.

##########
File path: 
nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/status/history/storage/ComponentCounterWritingTemplate.java
##########
@@ -0,0 +1,52 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.controller.status.history.storage;
+
+import io.questdb.cairo.TableWriter;
+import org.apache.commons.math3.util.Pair;
+import org.apache.nifi.controller.status.ProcessorStatus;
+import 
org.apache.nifi.controller.status.history.questdb.QuestDbWritingTemplate;
+
+import java.util.Collection;
+import java.util.Date;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+class ComponentCounterWritingTemplate extends 
QuestDbWritingTemplate<Pair<Date, ProcessorStatus>> {
+
+    public ComponentCounterWritingTemplate() {
+        super("componentCounter");
+    }
+
+    @Override
+    protected void addRows(final TableWriter tableWriter, final 
Collection<Pair<Date, ProcessorStatus>> entries) {
+        for (final Pair<Date, ProcessorStatus> entry : entries) {
+            final Map<String, Long> counters = entry.getSecond().getCounters();
+
+            if (counters != null && counters.size() > 0) {
+                for (final Map.Entry<String, Long> counter : 
counters.entrySet()) {
+                    final long measuredAt = 
TimeUnit.MILLISECONDS.toMicros(entry.getFirst().getTime());
+                    final TableWriter.Row counterRow = 
tableWriter.newRow(measuredAt);
+                    counterRow.putSym(1,  entry.getSecond().getId());
+                    counterRow.putSym(2,  counter.getKey());
+                    counterRow.putLong(3,  counter.getValue());

Review comment:
       Minor formatting: double spaces.




----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
[email protected]


Reply via email to