turcsanyip commented on code in PR #6350:
URL: https://github.com/apache/nifi/pull/6350#discussion_r959468701


##########
nifi-nar-bundles/nifi-airtable-bundle/nifi-airtable-processors/src/main/java/org/apache/nifi/processors/airtable/service/AirtableRestService.java:
##########
@@ -0,0 +1,108 @@
+/*
+ * 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.processors.airtable.service;
+
+import java.io.IOException;
+import java.net.URI;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.List;
+import javax.ws.rs.core.UriBuilder;
+import org.apache.commons.io.IOUtils;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.web.client.api.HttpResponseEntity;
+import org.apache.nifi.web.client.api.WebClientService;
+
+public class AirtableRestService {
+
+    public static final String API_V0_BASE_URL = "https://api.airtable.com/v0";;
+
+    private final WebClientService webClientService;
+    private final String apiUrl;
+    private final String apiToken;
+    private final String baseId;
+    private final String tableId;
+
+    public AirtableRestService(final WebClientService webClientService,
+            final String apiUrl,
+            final String apiToken,
+            final String baseId,
+            final String tableId) {
+        this.webClientService = webClientService;
+        this.apiUrl = apiUrl;
+        this.apiToken = apiToken;
+        this.baseId = baseId;
+        this.tableId = tableId;
+    }
+
+    public byte[] getRecords(final AirtableGetRecordsParameters filter) throws 
RateLimitExceeded {
+        final URI uri = buildUri(filter);
+        try (final HttpResponseEntity response = webClientService.get()
+                .uri(uri)
+                .header("Authorization", "Bearer " + apiToken)
+                .retrieve()) {
+
+            if (response.statusCode() >= 200 && response.statusCode() < 300) {
+                return IOUtils.toByteArray(response.body());
+            }
+            if (response.statusCode() == 429) {
+                throw new RateLimitExceeded();
+            }
+            final StringBuilder exceptionMessageBuilder = new 
StringBuilder("Invalid response. Code: " + response.statusCode());
+            final String bodyText = IOUtils.toString(response.body(), 
StandardCharsets.UTF_8);
+            if (bodyText != null) {
+                exceptionMessageBuilder.append(" Body: ").append(bodyText);
+            }
+
+            throw new ProcessException(exceptionMessageBuilder.toString());
+        } catch (IOException e) {
+            throw new ProcessException(String.format("Airtable HTTP request 
failed [%s]", uri), e);
+        }
+    }
+
+    public static class RateLimitExceeded extends Exception {

Review Comment:
   It is quite common to add -Exception suffix for exception classes: 
`RateLimitExceededException`



##########
nifi-nar-bundles/nifi-airtable-bundle/nifi-airtable-processors/src/main/java/org/apache/nifi/processors/airtable/service/AirtableRestService.java:
##########
@@ -0,0 +1,108 @@
+/*
+ * 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.processors.airtable.service;
+
+import java.io.IOException;
+import java.net.URI;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.List;
+import javax.ws.rs.core.UriBuilder;
+import org.apache.commons.io.IOUtils;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.web.client.api.HttpResponseEntity;
+import org.apache.nifi.web.client.api.WebClientService;
+
+public class AirtableRestService {
+
+    public static final String API_V0_BASE_URL = "https://api.airtable.com/v0";;
+
+    private final WebClientService webClientService;
+    private final String apiUrl;
+    private final String apiToken;
+    private final String baseId;
+    private final String tableId;
+
+    public AirtableRestService(final WebClientService webClientService,
+            final String apiUrl,
+            final String apiToken,
+            final String baseId,
+            final String tableId) {
+        this.webClientService = webClientService;
+        this.apiUrl = apiUrl;
+        this.apiToken = apiToken;
+        this.baseId = baseId;
+        this.tableId = tableId;
+    }
+
+    public byte[] getRecords(final AirtableGetRecordsParameters filter) throws 
RateLimitExceeded {
+        final URI uri = buildUri(filter);
+        try (final HttpResponseEntity response = webClientService.get()
+                .uri(uri)
+                .header("Authorization", "Bearer " + apiToken)
+                .retrieve()) {
+
+            if (response.statusCode() >= 200 && response.statusCode() < 300) {
+                return IOUtils.toByteArray(response.body());
+            }
+            if (response.statusCode() == 429) {
+                throw new RateLimitExceeded();
+            }
+            final StringBuilder exceptionMessageBuilder = new 
StringBuilder("Invalid response. Code: " + response.statusCode());
+            final String bodyText = IOUtils.toString(response.body(), 
StandardCharsets.UTF_8);
+            if (bodyText != null) {
+                exceptionMessageBuilder.append(" Body: ").append(bodyText);
+            }
+
+            throw new ProcessException(exceptionMessageBuilder.toString());
+        } catch (IOException e) {
+            throw new ProcessException(String.format("Airtable HTTP request 
failed [%s]", uri), e);
+        }
+    }
+
+    public static class RateLimitExceeded extends Exception {
+        public RateLimitExceeded() {
+            super("Airtable REST API rate limit exceeded");
+        }
+    }
+
+    private URI buildUri(AirtableGetRecordsParameters filter) {
+        final UriBuilder uriBuilder = 
UriBuilder.fromUri(apiUrl).path(baseId).path(tableId);
+
+        for (final String field : filter.getFields()) {
+            uriBuilder.queryParam("fields[]", field);
+        }
+
+        final List<String> filtersByFormula = new ArrayList<>();
+        filter.getFilterByFormula()
+                .ifPresent(filtersByFormula::add);
+        filter.getModifiedAfter()
+                .map(modifiedAfter -> 
"IS_AFTER(LAST_MODIFIED_TIME(),DATETIME_PARSE(\"" + modifiedAfter + "\"))")
+                .ifPresent(filtersByFormula::add);
+        filter.getModifiedBefore()
+                .map(modifiedBefore -> 
"IS_BEFORE(LAST_MODIFIED_TIME(),DATETIME_PARSE(\"" + modifiedBefore + "\"))")
+                .ifPresent(filtersByFormula::add);
+        if (!filtersByFormula.isEmpty()) {
+            uriBuilder.queryParam("filterByFormula", "AND(" + String.join(",", 
filtersByFormula) + ")");
+        }
+        filter.getOffset().ifPresent(offset -> uriBuilder.queryParam("offset", 
offset));
+        filter.getPageSize().ifPresent(pageSize -> 
uriBuilder.queryParam("pageSize", pageSize));
+
+        return uriBuilder.build();
+    }
+}

Review Comment:
   It is not easy to follow here what are `filter` / `filterByFormula` / 
`filtersByFormula`, so I suggest renaming these variables:
   - `filter` => `getRecordsParameters`
   - `AirtableGetRecordsParameters.filterByFormula` => 
`AirtableGetRecordsParameters.customFilter`
   - `filtersByFormula` => `filters`
   



##########
nifi-nar-bundles/nifi-airtable-bundle/nifi-airtable-processors/src/main/java/org/apache/nifi/processors/airtable/QueryAirtableTable.java:
##########
@@ -0,0 +1,422 @@
+/*
+ * 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.processors.airtable;
+
+import static 
org.apache.nifi.flowfile.attributes.FragmentAttributes.FRAGMENT_COUNT;
+import static 
org.apache.nifi.flowfile.attributes.FragmentAttributes.FRAGMENT_ID;
+import static 
org.apache.nifi.flowfile.attributes.FragmentAttributes.FRAGMENT_INDEX;
+import static 
org.apache.nifi.processors.airtable.service.AirtableRestService.API_V0_BASE_URL;
+
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.time.OffsetDateTime;
+import java.time.format.DateTimeFormatter;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Collectors;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
+import org.apache.nifi.annotation.behavior.PrimaryNodeOnly;
+import org.apache.nifi.annotation.behavior.Stateful;
+import org.apache.nifi.annotation.behavior.TriggerSerially;
+import org.apache.nifi.annotation.behavior.TriggerWhenEmpty;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.configuration.DefaultSettings;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+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.state.Scope;
+import org.apache.nifi.components.state.StateMap;
+import org.apache.nifi.expression.ExpressionLanguageScope;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.flowfile.attributes.CoreAttributes;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.processors.airtable.record.AirtableRecordSet;
+import 
org.apache.nifi.processors.airtable.record.AirtableJsonTreeRowRecordReaderFactory;
+import 
org.apache.nifi.processors.airtable.service.AirtableGetRecordsParameters;
+import org.apache.nifi.processors.airtable.service.AirtableRestService;
+import 
org.apache.nifi.processors.airtable.service.AirtableRestService.RateLimitExceeded;
+import org.apache.nifi.schema.access.SchemaNotFoundException;
+import org.apache.nifi.serialization.JsonRecordReaderFactory;
+import org.apache.nifi.serialization.MalformedRecordException;
+import org.apache.nifi.serialization.RecordSetWriter;
+import org.apache.nifi.serialization.RecordSetWriterFactory;
+import org.apache.nifi.serialization.WriteResult;
+import org.apache.nifi.serialization.record.RecordSchema;
+import org.apache.nifi.serialization.record.RecordSet;
+import org.apache.nifi.web.client.provider.api.WebClientServiceProvider;
+
+@PrimaryNodeOnly
+@InputRequirement(Requirement.INPUT_FORBIDDEN)
+@TriggerSerially
+@TriggerWhenEmpty
+@Tags({"airtable", "query", "database"})
+@CapabilityDescription("Query records from an Airtable table. Records are 
incrementally retrieved based on the last modified time of the records."
+        + " Records can also be further filtered by setting the 'Custom 
Filter' property which supports the formulas provided by the Airtable API."
+        + " Schema can be provided by setting up a JsonTreeReader controller 
service properly. This processor is intended to be run on the Primary Node 
only.")
+@Stateful(scopes = Scope.CLUSTER, description = "The last successful query's 
time is stored in order to enable incremental loading."
+        + " The initial query returns all the records in the table and each 
subsequent query filters the records by their last modified time."
+        + " In other words, if a record is updated after the last successful 
query only the updated records will be returned in the next query."
+        + " State is stored across the cluster so that this Processor can be 
run on Primary Node only and if a new Primary Node is selected,"
+        + " the new node can pick up where the previous node left off, without 
duplicating the data.")
+@WritesAttributes({
+        @WritesAttribute(attribute = "mime.type", description = "Sets the 
mime.type attribute to the MIME Type specified by the Record Writer."),
+        @WritesAttribute(attribute = "record.count", description = "Sets the 
number of records in the FlowFile.")
+})
+@DefaultSettings(yieldDuration = "35 sec")
+public class QueryAirtableTable extends AbstractProcessor {
+
+    static final PropertyDescriptor API_URL = new PropertyDescriptor.Builder()
+            .name("api-url")
+            .displayName("API URL")
+            .description("The URL for the Airtable REST API including the 
domain and the path to the API (e.g. https://api.airtable.com/v0).")
+            .defaultValue(API_V0_BASE_URL)
+            .addValidator(StandardValidators.NON_BLANK_VALIDATOR)
+            .addValidator(StandardValidators.URL_VALIDATOR)
+            
.expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY)
+            .required(true)
+            .build();
+
+    static final PropertyDescriptor BASE_ID = new PropertyDescriptor.Builder()
+            .name("base-id")
+            .displayName("Base ID")
+            .description("The ID of the Airtable base to be queried.")
+            .required(true)
+            
.expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY)
+            .addValidator(StandardValidators.NON_BLANK_VALIDATOR)
+            .build();
+
+    static final PropertyDescriptor TABLE_ID = new PropertyDescriptor.Builder()
+            .name("table-id")
+            .displayName("Table Name or ID")
+            .description("The name or the ID of the Airtable table to be 
queried.")
+            .required(true)
+            
.expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY)
+            .addValidator(StandardValidators.NON_BLANK_VALIDATOR)
+            .build();
+
+    static final PropertyDescriptor API_TOKEN = new 
PropertyDescriptor.Builder()
+            .name("api-token")
+            .displayName("API Token")
+            .description("The REST API token to use in queries. Should be 
generated on Airtable's account page.")
+            .required(true)
+            .sensitive(true)
+            .expressionLanguageSupported(ExpressionLanguageScope.NONE)
+            .addValidator(StandardValidators.NON_BLANK_VALIDATOR)
+            .build();
+
+    static final PropertyDescriptor FIELDS = new PropertyDescriptor.Builder()
+            .name("fields")
+            .displayName("Fields")
+            .description("Comma-separated list of fields to query from the 
table. Both the field's name and ID can be used.")
+            
.expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY)
+            .addValidator(StandardValidators.NON_BLANK_VALIDATOR)
+            .build();
+
+    static final PropertyDescriptor CUSTOM_FILTER = new 
PropertyDescriptor.Builder()
+            .name("custom-filter")
+            .displayName("Custom Filter")
+            .description("Filter records by Airtable's formulas.")
+            
.expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY)
+            .addValidator(StandardValidators.NON_BLANK_VALIDATOR)
+            .build();
+
+    static final PropertyDescriptor PAGE_SIZE = new 
PropertyDescriptor.Builder()
+            .name("page-size")
+            .displayName("Page Size")
+            .description("Number of records to be fetched in a page. Should be 
between 0 and 100 inclusively.")
+            .defaultValue("0")
+            .required(true)
+            
.expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY)
+            .addValidator(StandardValidators.createLongValidator(0, 100, true))
+            .build();
+
+    static final PropertyDescriptor MAX_RECORDS_PER_FLOW_FILE = new 
PropertyDescriptor.Builder()
+            .name("max-records-per-flow-file")
+            .displayName("Max Records Per Flow File")
+            .description("The maximum number of result records that will be 
included in a single FlowFile. This will allow you to break up very large"
+                    + " result sets into multiple FlowFiles. If the value 
specified is zero, then all records are returned in a single FlowFile.")
+            .defaultValue("0")
+            .required(true)
+            
.expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY)
+            .addValidator(StandardValidators.NON_NEGATIVE_INTEGER_VALIDATOR)
+            .build();
+
+    static final PropertyDescriptor OUTPUT_BATCH_SIZE = new 
PropertyDescriptor.Builder()
+            .name("output-batch-size")
+            .displayName("Output Batch Size")
+            .description("The number of output FlowFiles to queue before 
committing the process session. When set to zero, the session will be committed 
when all records"
+                    + " have been processed and the output FlowFiles are ready 
for transfer to the downstream relationship. For large result sets, this can 
cause a large burst of FlowFiles"
+                    + " to be transferred at the end of processor execution. 
If this property is set, then when the specified number of FlowFiles are ready 
for transfer, then the session will"
+                    + " be committed, thus releasing the FlowFiles to the 
downstream relationship.")
+            .defaultValue("0")
+            .required(true)
+            
.expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY)
+            .addValidator(StandardValidators.NON_NEGATIVE_INTEGER_VALIDATOR)
+            .build();
+
+    static final PropertyDescriptor METADATA_STRATEGY = new 
PropertyDescriptor.Builder()
+            .name("metadata-strategy")
+            .displayName("Metadata Strategy")
+            .description("Strategy to use for fetching record schema. 
Currently only 'Use JSON Record Reader' is supported."
+                    + " When Airtable Metadata API becomes more stable it will 
be possible to fetch the record schema through it.")
+            .required(true)
+            .defaultValue(MetadataStrategy.USE_JSON_RECORD_READER.name())
+            .allowableValues(MetadataStrategy.class)
+            .build();
+
+    static final PropertyDescriptor SCHEMA_READER = new 
PropertyDescriptor.Builder()
+            .name("schema-reader")
+            .displayName("Schema Reader")
+            .description("JsonTreeReader service to use for fetching the 
schema of records returned by the Airtable REST API")
+            .dependsOn(METADATA_STRATEGY, 
MetadataStrategy.USE_JSON_RECORD_READER.name())
+            .identifiesControllerService(JsonRecordReaderFactory.class)
+            .required(true)
+            .build();
+
+    static final PropertyDescriptor RECORD_WRITER = new 
PropertyDescriptor.Builder()
+            .name("record-writer")
+            .displayName("Record Writer")
+            .description("Service used for writing records returned by the 
Airtable REST API")
+            .identifiesControllerService(RecordSetWriterFactory.class)
+            .required(true)
+            .build();
+
+    static final PropertyDescriptor WEB_CLIENT_SERVICE_PROVIDER = new 
PropertyDescriptor.Builder()
+            .name("web-client-service-provider")
+            .displayName("Web Client Service Provider")
+            .description("Web Client Service Provider to use for Airtable REST 
API requests")
+            .identifiesControllerService(WebClientServiceProvider.class)
+            .required(true)
+            .build();
+
+    static final Relationship REL_SUCCESS = new Relationship.Builder()
+            .name("success")
+            .description("For FlowFiles created as a result of a successful 
query.")
+            .build();
+
+    private static final List<PropertyDescriptor> PROPERTIES = 
Collections.unmodifiableList(Arrays.asList(
+            API_URL,
+            BASE_ID,
+            TABLE_ID,
+            API_TOKEN,

Review Comment:
   Please move it up just after `API_URL`.



##########
nifi-nar-bundles/nifi-airtable-bundle/nifi-airtable-processors/src/main/java/org/apache/nifi/processors/airtable/QueryAirtableTable.java:
##########
@@ -0,0 +1,422 @@
+/*
+ * 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.processors.airtable;
+
+import static 
org.apache.nifi.flowfile.attributes.FragmentAttributes.FRAGMENT_COUNT;
+import static 
org.apache.nifi.flowfile.attributes.FragmentAttributes.FRAGMENT_ID;
+import static 
org.apache.nifi.flowfile.attributes.FragmentAttributes.FRAGMENT_INDEX;
+import static 
org.apache.nifi.processors.airtable.service.AirtableRestService.API_V0_BASE_URL;
+
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.time.OffsetDateTime;
+import java.time.format.DateTimeFormatter;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Collectors;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
+import org.apache.nifi.annotation.behavior.PrimaryNodeOnly;
+import org.apache.nifi.annotation.behavior.Stateful;
+import org.apache.nifi.annotation.behavior.TriggerSerially;
+import org.apache.nifi.annotation.behavior.TriggerWhenEmpty;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.configuration.DefaultSettings;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+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.state.Scope;
+import org.apache.nifi.components.state.StateMap;
+import org.apache.nifi.expression.ExpressionLanguageScope;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.flowfile.attributes.CoreAttributes;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.processors.airtable.record.AirtableRecordSet;
+import 
org.apache.nifi.processors.airtable.record.AirtableJsonTreeRowRecordReaderFactory;
+import 
org.apache.nifi.processors.airtable.service.AirtableGetRecordsParameters;
+import org.apache.nifi.processors.airtable.service.AirtableRestService;
+import 
org.apache.nifi.processors.airtable.service.AirtableRestService.RateLimitExceeded;
+import org.apache.nifi.schema.access.SchemaNotFoundException;
+import org.apache.nifi.serialization.JsonRecordReaderFactory;
+import org.apache.nifi.serialization.MalformedRecordException;
+import org.apache.nifi.serialization.RecordSetWriter;
+import org.apache.nifi.serialization.RecordSetWriterFactory;
+import org.apache.nifi.serialization.WriteResult;
+import org.apache.nifi.serialization.record.RecordSchema;
+import org.apache.nifi.serialization.record.RecordSet;
+import org.apache.nifi.web.client.provider.api.WebClientServiceProvider;
+
+@PrimaryNodeOnly
+@InputRequirement(Requirement.INPUT_FORBIDDEN)
+@TriggerSerially
+@TriggerWhenEmpty
+@Tags({"airtable", "query", "database"})
+@CapabilityDescription("Query records from an Airtable table. Records are 
incrementally retrieved based on the last modified time of the records."
+        + " Records can also be further filtered by setting the 'Custom 
Filter' property which supports the formulas provided by the Airtable API."
+        + " Schema can be provided by setting up a JsonTreeReader controller 
service properly. This processor is intended to be run on the Primary Node 
only.")
+@Stateful(scopes = Scope.CLUSTER, description = "The last successful query's 
time is stored in order to enable incremental loading."
+        + " The initial query returns all the records in the table and each 
subsequent query filters the records by their last modified time."
+        + " In other words, if a record is updated after the last successful 
query only the updated records will be returned in the next query."
+        + " State is stored across the cluster so that this Processor can be 
run on Primary Node only and if a new Primary Node is selected,"
+        + " the new node can pick up where the previous node left off, without 
duplicating the data.")
+@WritesAttributes({
+        @WritesAttribute(attribute = "mime.type", description = "Sets the 
mime.type attribute to the MIME Type specified by the Record Writer."),
+        @WritesAttribute(attribute = "record.count", description = "Sets the 
number of records in the FlowFile.")
+})
+@DefaultSettings(yieldDuration = "35 sec")
+public class QueryAirtableTable extends AbstractProcessor {
+
+    static final PropertyDescriptor API_URL = new PropertyDescriptor.Builder()
+            .name("api-url")
+            .displayName("API URL")
+            .description("The URL for the Airtable REST API including the 
domain and the path to the API (e.g. https://api.airtable.com/v0).")
+            .defaultValue(API_V0_BASE_URL)
+            .addValidator(StandardValidators.NON_BLANK_VALIDATOR)
+            .addValidator(StandardValidators.URL_VALIDATOR)
+            
.expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY)
+            .required(true)
+            .build();
+
+    static final PropertyDescriptor BASE_ID = new PropertyDescriptor.Builder()
+            .name("base-id")
+            .displayName("Base ID")
+            .description("The ID of the Airtable base to be queried.")
+            .required(true)
+            
.expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY)
+            .addValidator(StandardValidators.NON_BLANK_VALIDATOR)
+            .build();
+
+    static final PropertyDescriptor TABLE_ID = new PropertyDescriptor.Builder()
+            .name("table-id")
+            .displayName("Table Name or ID")
+            .description("The name or the ID of the Airtable table to be 
queried.")
+            .required(true)
+            
.expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY)
+            .addValidator(StandardValidators.NON_BLANK_VALIDATOR)
+            .build();
+
+    static final PropertyDescriptor API_TOKEN = new 
PropertyDescriptor.Builder()
+            .name("api-token")
+            .displayName("API Token")
+            .description("The REST API token to use in queries. Should be 
generated on Airtable's account page.")

Review Comment:
   The Airtable documentation refers it as "API Key" so I would suggest using 
the same terminology.



-- 
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.

To unsubscribe, e-mail: [email protected]

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

Reply via email to