You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@nifi.apache.org by GitBox <gi...@apache.org> on 2022/08/02 07:16:05 UTC
[GitHub] [nifi] xuanronaldo opened a new pull request, #6265: implement PutIoTDB
xuanronaldo opened a new pull request, #6265:
URL: https://github.com/apache/nifi/pull/6265
## PutIoTDB
### Properties of PutIoTDB
| property | description | default value | necessary |
| ------------- | ------------------------------------------------------------ | ------------- | --------- |
| Host | The host of IoTDB. | null | true |
| Port | The port of IoTDB. | 6667 | true |
| Username | Username to access the IoTDB. | null | true |
| Password | Password to access the IoTDB. | null | true |
| Record Reader | Specifies the type of Record Reader controller service to use <br />for parsing the incoming data and determining the schema. | null | true |
| Time Field | The field name which represents time. It can be updated by expression language. | Time | false |
| Schema | The schema that IoTDB needs doesn't support good by NiFi.<br/>Therefore, you can define the schema here. <br />Besides, you can set encoding type and compression type by this method.<br />If you don't set this property, the inferred schema will be used.<br /> It can be updated by expression language. | null | false |
| Aligned | Whether using aligned interface? It can be updated by expression language. | false | false |
| MaxRowNumber | Specifies the max row number of each tablet. It can be updated by expression language. | 1024 | false |
### Inferred Schema of Flowfile
There are a couple of rules about flowfile:
1. The flowfile can be read by `Record Reader`.
2. The schema of flowfile must contains a field `Time`. Otherwise, the property `Time Field` should be set to represent it.
3. The data type of time must be `STRING` or `LONG`.
4. Fields excepted time must start with `root.`.
5. The supported data types are `INT`, `LONG`, `FLOAT`, `DOUBLE`, `BOOLEAN`, `TEXT`.
### Convert Schema by property
As mentioned above, converting schema by property which is more flexible and stronger than inferred schema.
The structure of property `Schema`:
```json
{
"timeType": "long",
"field": [{
"tsName": "root.sg.d1.s1",
"dataType": "INT32",
"encoding": "RLE",
"compressionType": "GZIP"
}, {
"tsName": "root.sg.d1.s2",
"dataType": "INT64",
"encoding": "RLE",
"compressionType": "GZIP"
}]
}
```
**Note**
1. The JSON of schema must contain `timeType` and `fields`.
2. There are only two options `long` and `string` for `timeType`.
3. The columns `tsName` and `dataType` must be set.
4. The tsName must start with `root.`.
5. The supported `dataTypes` are `INT32`, `INT64`, `FLOAT`, `DOUBLE`, `BOOLEAN`, `TEXT`.
6. The supported `encoding` are `PLAIN`, `DICTIONARY`, `RLE`, `DIFF`, `TS_2DIFF`, `BITMAP`, `GORILLA_V1`, `REGULAR`, `GORILLA`.
7. The supported `compressionType` are `UNCOMPRESSED`, `SNAPPY`, `GZIP`, `LZO`, `SDT`, `PAA`, `PLA`, `LZ4`.
## Relationships
| relationship | description |
| ------------ | ---------------------------------------------------- |
| success | Data can be written correctly or flow file is empty. |
| failure | The shema or flow file is abnormal. |
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] exceptionfactory commented on a diff in pull request #6265: NIFI-10234 Implement PutIoTDB
Posted by GitBox <gi...@apache.org>.
exceptionfactory commented on code in PR #6265:
URL: https://github.com/apache/nifi/pull/6265#discussion_r946167233
##########
nifi-nar-bundles/nifi-iotdb-bundle/nifi-iotdb-processors/src/test/java/org/apache/nifi/processors/PutIoTDBIT.java:
##########
@@ -0,0 +1,297 @@
+/*
+ * 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;
+
+import org.apache.iotdb.db.utils.EnvironmentUtils;
+import org.apache.iotdb.rpc.IoTDBConnectionException;
+import org.apache.iotdb.rpc.StatementExecutionException;
+import org.apache.iotdb.session.Session;
+import org.apache.iotdb.session.SessionDataSet;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.read.common.Field;
+import org.apache.iotdb.tsfile.read.common.RowRecord;
+import org.apache.nifi.reporting.InitializationException;
+import org.apache.nifi.serialization.record.MockRecordParser;
+import org.apache.nifi.serialization.record.RecordFieldType;
+import org.apache.nifi.util.TestRunner;
+import org.apache.nifi.util.TestRunners;
+import org.junit.Assert;
Review Comment:
@xuanronaldo Please review this comment and update the assertions to use JUnit 5 instead of JUnit 4. Thanks!
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] xuanronaldo commented on pull request #6265: NIFI-10234 Implement PutIoTDB
Posted by GitBox <gi...@apache.org>.
xuanronaldo commented on PR #6265:
URL: https://github.com/apache/nifi/pull/6265#issuecomment-1212674273
I have merged the main branch to resolve the problem which happened in workflow. Please approve running workflows agin. @exceptionfactory
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] exceptionfactory commented on pull request #6265: NIFI-10234 Implement PutIoTDB
Posted by GitBox <gi...@apache.org>.
exceptionfactory commented on PR #6265:
URL: https://github.com/apache/nifi/pull/6265#issuecomment-1206396816
Thanks for making the updates @xuanronaldo. Merging requires approval from at least one committer. I will try to take a look at the latest version soon.
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] xuanronaldo commented on pull request #6265: NIFI-10234 Implement PutIoTDB
Posted by GitBox <gi...@apache.org>.
xuanronaldo commented on PR #6265:
URL: https://github.com/apache/nifi/pull/6265#issuecomment-1207801918
> Thanks for making the initial changes @xuanronaldo.
>
> I took a closer look at the implementation and noted a number of areas for improvement. The division of the abstract class and concrete class is helpful to indicate the core functionality. The core logic in `onTrigger` was difficult to follow at points because of the layout and nesting, so some general refactoring would be helpful.
Thanks for your suggestions. I will update the code asap.
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] xuanronaldo commented on pull request #6265: NIFI-10234 Implement PutIoTDB
Posted by GitBox <gi...@apache.org>.
xuanronaldo commented on PR #6265:
URL: https://github.com/apache/nifi/pull/6265#issuecomment-1240341357
I have reverted these changes. However, several files were still marked for modification. I have no idea what to do. @exceptionfactory
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] xuanronaldo commented on pull request #6265: NIFI-10234 Implement PutIoTDB
Posted by GitBox <gi...@apache.org>.
xuanronaldo commented on PR #6265:
URL: https://github.com/apache/nifi/pull/6265#issuecomment-1218890319
>
Okay, I got it. I will update the code asap. Thanks for your suggestions.
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] xuanronaldo commented on pull request #6265: NIFI-10234 Implement PutIoTDB
Posted by GitBox <gi...@apache.org>.
xuanronaldo commented on PR #6265:
URL: https://github.com/apache/nifi/pull/6265#issuecomment-1203389409
>
Ok, I got it. Thanks for your suggestions. I'm working on it.
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] exceptionfactory commented on pull request #6265: NIFI-10234 Implement PutIoTDB
Posted by GitBox <gi...@apache.org>.
exceptionfactory commented on PR #6265:
URL: https://github.com/apache/nifi/pull/6265#issuecomment-1245316479
Thanks for following up @xuanronaldo. Sure, if it is easier for you to create a new branch with the current version, that's fine as well. It is easy to link to this PR for background, and we can work from the new PR.
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] xuanronaldo commented on pull request #6265: NIFI-10234 Implement PutIoTDB
Posted by GitBox <gi...@apache.org>.
xuanronaldo commented on PR #6265:
URL: https://github.com/apache/nifi/pull/6265#issuecomment-1205966477
I have pushed the code, and all checks have passed. What should I do next to merge the branch?
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] exceptionfactory commented on a diff in pull request #6265: NIFI-10234 Implement PutIoTDB
Posted by GitBox <gi...@apache.org>.
exceptionfactory commented on code in PR #6265:
URL: https://github.com/apache/nifi/pull/6265#discussion_r939172193
##########
nifi-nar-bundles/nifi-iotdb-bundle/nifi-iotdb-processors/pom.xml:
##########
@@ -0,0 +1,116 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ 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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <parent>
+ <artifactId>nifi-iotdb-bundle</artifactId>
+ <groupId>org.apache.nifi</groupId>
+ <version>1.18.0-SNAPSHOT</version>
+ </parent>
+ <modelVersion>4.0.0</modelVersion>
+
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-iotdb-processors</artifactId>
+ <packaging>jar</packaging>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.iotdb</groupId>
+ <artifactId>iotdb-session</artifactId>
+ <version>${iotdb.sdk.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.iotdb</groupId>
+ <artifactId>iotdb-server</artifactId>
+ <version>${iotdb.sdk.version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.iotdb</groupId>
+ <artifactId>iotdb-server</artifactId>
+ <version>${iotdb.sdk.version}</version>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-api</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-mock</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-record-serialization-service-api</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-record</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-mock-record-utils</artifactId>
+ <version>${project.version}</version>
Review Comment:
This dependency should be marked with a `test` scope.
##########
nifi-nar-bundles/nifi-iotdb-bundle/nifi-iotdb-processors/pom.xml:
##########
@@ -0,0 +1,116 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ 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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <parent>
+ <artifactId>nifi-iotdb-bundle</artifactId>
+ <groupId>org.apache.nifi</groupId>
+ <version>1.18.0-SNAPSHOT</version>
+ </parent>
+ <modelVersion>4.0.0</modelVersion>
+
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-iotdb-processors</artifactId>
+ <packaging>jar</packaging>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.iotdb</groupId>
+ <artifactId>iotdb-session</artifactId>
+ <version>${iotdb.sdk.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.iotdb</groupId>
+ <artifactId>iotdb-server</artifactId>
+ <version>${iotdb.sdk.version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.iotdb</groupId>
+ <artifactId>iotdb-server</artifactId>
+ <version>${iotdb.sdk.version}</version>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-api</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-mock</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-record-serialization-service-api</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-record</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-mock-record-utils</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-dbcp-service-api</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-standard-services-api-nar</artifactId>
+ <version>${project.version}</version>
+ <type>nar</type>
+ </dependency>
Review Comment:
This dependency must be removed to `nifi-iotdb-nar`, instead of this `jar` module.
##########
nifi-nar-bundles/nifi-iotdb-bundle/nifi-iotdb-nar/pom.xml:
##########
@@ -0,0 +1,37 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ 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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <parent>
+ <artifactId>nifi-iotdb-bundle</artifactId>
+ <groupId>org.apache.nifi</groupId>
+ <version>1.18.0-SNAPSHOT</version>
+ </parent>
+ <modelVersion>4.0.0</modelVersion>
+
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-iotdb-nar</artifactId>
+ <packaging>nar</packaging>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-iotdb-processors</artifactId>
+ </dependency>
Review Comment:
This module needs a dependency on `nifi-standard-services-api-nar` of type `nar` in order to access standard NiFi Controller Services at runtime.
##########
nifi-nar-bundles/nifi-iotdb-bundle/nifi-iotdb-processors/src/main/java/org/apache/nifi/processors/AbstractIoTDB.java:
##########
@@ -0,0 +1,442 @@
+/*
+ * 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;
+
+import java.text.SimpleDateFormat;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicReference;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.nifi.processors.model.Field;
+import org.apache.nifi.processors.model.Schema;
+import org.apache.iotdb.rpc.IoTDBConnectionException;
+import org.apache.iotdb.session.Session;
+import org.apache.iotdb.tsfile.file.metadata.enums.CompressionType;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSEncoding;
+import org.apache.iotdb.tsfile.utils.Binary;
+import org.apache.iotdb.tsfile.write.record.Tablet;
+import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.serialization.record.DataType;
+import org.apache.nifi.serialization.record.RecordFieldType;
+import org.apache.nifi.serialization.record.RecordSchema;
+import org.apache.nifi.util.Tuple;
+
+public abstract class AbstractIoTDB extends AbstractProcessor {
+ private static final int DEFAULT_IOTDB_PORT = 6667;
+
+ private static Map<RecordFieldType, TSDataType> typeMap =
+ new HashMap<>();
+
+ static final Set<RecordFieldType> supportedType =
+ new HashSet<>();
+
+ static final Set<RecordFieldType> supportedTimeType =
+ new HashSet<>();
+
+ private static final String[] STRING_TIME_FORMAT =
+ new String[]{
+ "yyyy-MM-dd HH:mm:ss.SSSX",
+ "yyyy/MM/dd HH:mm:ss.SSSX",
+ "yyyy.MM.dd HH:mm:ss.SSSX",
+ "yyyy-MM-dd HH:mm:ssX",
+ "yyyy/MM/dd HH:mm:ssX",
+ "yyyy.MM.dd HH:mm:ssX",
+ "yyyy-MM-dd HH:mm:ss.SSSz",
+ "yyyy/MM/dd HH:mm:ss.SSSz",
+ "yyyy.MM.dd HH:mm:ss.SSSz",
+ "yyyy-MM-dd HH:mm:ssz",
+ "yyyy/MM/dd HH:mm:ssz",
+ "yyyy.MM.dd HH:mm:ssz",
+ "yyyy-MM-dd HH:mm:ss.SSS",
+ "yyyy/MM/dd HH:mm:ss.SSS",
+ "yyyy.MM.dd HH:mm:ss.SSS",
+ "yyyy-MM-dd HH:mm:ss",
+ "yyyy/MM/dd HH:mm:ss",
+ "yyyy.MM.dd HH:mm:ss",
+ "yyyy-MM-dd'T'HH:mm:ss.SSSX",
+ "yyyy/MM/dd'T'HH:mm:ss.SSSX",
+ "yyyy.MM.dd'T'HH:mm:ss.SSSX",
+ "yyyy-MM-dd'T'HH:mm:ssX",
+ "yyyy/MM/dd'T'HH:mm:ssX",
+ "yyyy.MM.dd'T'HH:mm:ssX",
+ "yyyy-MM-dd'T'HH:mm:ss.SSSz",
+ "yyyy/MM/dd'T'HH:mm:ss.SSSz",
+ "yyyy.MM.dd'T'HH:mm:ss.SSSz",
+ "yyyy-MM-dd'T'HH:mm:ssz",
+ "yyyy/MM/dd'T'HH:mm:ssz",
+ "yyyy.MM.dd'T'HH:mm:ssz",
+ "yyyy-MM-dd'T'HH:mm:ss.SSS",
+ "yyyy/MM/dd'T'HH:mm:ss.SSS",
+ "yyyy.MM.dd'T'HH:mm:ss.SSS",
+ "yyyy-MM-dd'T'HH:mm:ss",
+ "yyyy/MM/dd'T'HH:mm:ss",
+ "yyyy.MM.dd'T'HH:mm:ss"
+ };
+
+ static final PropertyDescriptor IOTDB_HOST =
+ new PropertyDescriptor.Builder()
+ .name("Host")
+ .description("The host of IoTDB.")
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .required(true)
+ .build();
+
+ static final PropertyDescriptor IOTDB_PORT =
+ new PropertyDescriptor.Builder()
+ .name("Port")
+ .description("The port of IoTDB.")
+ .defaultValue(String.valueOf(DEFAULT_IOTDB_PORT))
+ .addValidator(StandardValidators.PORT_VALIDATOR)
+ .required(true)
+ .build();
+
+ static final PropertyDescriptor USERNAME =
+ new PropertyDescriptor.Builder()
+ .name("Username")
+ .description("Username to access the IoTDB.")
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .required(true)
+ .build();
+
+ static final PropertyDescriptor PASSWORD =
+ new PropertyDescriptor.Builder()
+ .name("Password")
+ .description("Password to access the IoTDB.")
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .required(true)
+ .sensitive(true)
+ .build();
+
+ protected static List<PropertyDescriptor> descriptors = new ArrayList<>();
+
+ protected static final Relationship REL_SUCCESS =
+ new Relationship.Builder()
+ .name("success")
+ .description("files that were successfully processed")
+ .build();
+ protected static final Relationship REL_FAILURE =
+ new Relationship.Builder()
+ .name("failure")
+ .description("files that were not successfully processed")
+ .build();
+
+ protected static Set<Relationship> relationships = new HashSet<>();
+
+ static {
+ descriptors.add(IOTDB_HOST);
+ descriptors.add(IOTDB_PORT);
+ descriptors.add(USERNAME);
+ descriptors.add(PASSWORD);
+
+ relationships.add(REL_SUCCESS);
+ relationships.add(REL_FAILURE);
+
+ typeMap.put(RecordFieldType.STRING, TSDataType.TEXT);
+ typeMap.put(RecordFieldType.BOOLEAN, TSDataType.BOOLEAN);
+ typeMap.put(RecordFieldType.INT, TSDataType.INT32);
+ typeMap.put(RecordFieldType.LONG, TSDataType.INT64);
+ typeMap.put(RecordFieldType.FLOAT, TSDataType.FLOAT);
+ typeMap.put(RecordFieldType.DOUBLE, TSDataType.DOUBLE);
+
+ supportedType.add(RecordFieldType.BOOLEAN);
+ supportedType.add(RecordFieldType.STRING);
+ supportedType.add(RecordFieldType.INT);
+ supportedType.add(RecordFieldType.LONG);
+ supportedType.add(RecordFieldType.FLOAT);
+ supportedType.add(RecordFieldType.DOUBLE);
+
+ supportedTimeType.add(RecordFieldType.STRING);
+ supportedTimeType.add(RecordFieldType.LONG);
+ }
+
+ protected final AtomicReference<Session> session = new AtomicReference<>(null);
+
+ @OnScheduled
+ public void onScheduled(ProcessContext context) {
+ connectToIoTDB(context);
+ }
+
+ void connectToIoTDB(ProcessContext context) {
+ if (session.get() == null) {
+ final String host, username, password;
+ final int port;
+
+ host = context.getProperty(IOTDB_HOST).getValue();
+ port = Integer.parseInt(context.getProperty(IOTDB_PORT).getValue());
+ username = context.getProperty(USERNAME).getValue();
+ password = context.getProperty(PASSWORD).getValue();
+
+ session.set(
+ new Session.Builder()
+ .host(host)
+ .port(port)
+ .username(username)
+ .password(password)
+ .build());
+ try {
+ session.get().open();
+ } catch (IoTDBConnectionException e) {
+ getLogger().error(e.getMessage());
Review Comment:
This should be changed to log a message with the exception:
```suggestion
getLogger().error("IoTDB connection failed", e);
```
##########
nifi-nar-bundles/nifi-iotdb-bundle/nifi-iotdb-processors/src/main/java/org/apache/nifi/processors/PutIoTDB.java:
##########
@@ -0,0 +1,286 @@
+/*
+ * 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;
+
+import java.io.InputStream;
+import java.text.SimpleDateFormat;
+import java.util.Set;
+import java.util.List;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.nifi.processors.model.Schema;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.write.record.Tablet;
+import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnUnscheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.expression.ExpressionLanguageScope;
+import org.apache.nifi.flowfile.FlowFile;
+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.serialization.RecordReader;
+import org.apache.nifi.serialization.RecordReaderFactory;
+import org.apache.nifi.serialization.record.Record;
+import org.apache.nifi.util.Tuple;
+
+@Tags({"iotdb", "insert", "tablet"})
+@InputRequirement(InputRequirement.Requirement.INPUT_REQUIRED)
+@CapabilityDescription(
+ "This is a record aware processor that reads the content of the incoming FlowFile as individual records using the "
+ + "configured 'Record Reader' and writes them to Apache IoTDB using native interface.")
+public class PutIoTDB extends AbstractIoTDB {
+
+ static final PropertyDescriptor RECORD_READER_FACTORY =
+ new PropertyDescriptor.Builder()
+ .name("Record Reader")
+ .description(
+ "Specifies the type of Record Reader controller service to use for parsing the incoming data "
+ + "and determining the schema")
+ .identifiesControllerService(RecordReaderFactory.class)
+ .required(true)
+ .build();
+
+ static final PropertyDescriptor TIME_FIELD =
+ new PropertyDescriptor.Builder()
+ .name("Time Field")
+ .description(
+ "The field name which represents time. It can be updated by expression language.")
+ .defaultValue("Time")
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .required(false)
+ .build();
+
+ static final PropertyDescriptor SCHEMA =
+ new PropertyDescriptor.Builder()
+ .name("Schema")
+ .description(
+ "The schema that IoTDB needs doesn't support good by NiFi.\n"
+ + "Therefore, you can define the schema here.\n"
+ + "Besides, you can set encoding type and compression type by this method.\n"
+ + "If you don't set this property, the inferred schema will be used.\n"
+ + "It can be updated by expression language.")
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .required(false)
+ .build();
+
+ static final PropertyDescriptor ALIGNED =
+ new PropertyDescriptor.Builder()
+ .name("Aligned")
+ .description("Whether using aligned interface? It can be updated by expression language.")
+ .allowableValues(new String[] {"true", "false"})
+ .addValidator(StandardValidators.BOOLEAN_VALIDATOR)
+ .defaultValue("false")
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .required(false)
+ .build();
+
+ static final PropertyDescriptor MAX_ROW_NUMBER =
+ new PropertyDescriptor.Builder()
+ .name("Max Row Number")
+ .description(
+ "Specifies the max row number of each tablet. It can be updated by expression language.")
Review Comment:
```suggestion
"Specifies the max row number of each Apache IoTDB Tablet")
```
##########
nifi-nar-bundles/nifi-iotdb-bundle/nifi-iotdb-processors/src/main/java/org/apache/nifi/processors/PutIoTDB.java:
##########
@@ -0,0 +1,286 @@
+/*
+ * 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;
+
+import java.io.InputStream;
+import java.text.SimpleDateFormat;
+import java.util.Set;
+import java.util.List;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.nifi.processors.model.Schema;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.write.record.Tablet;
+import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnUnscheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.expression.ExpressionLanguageScope;
+import org.apache.nifi.flowfile.FlowFile;
+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.serialization.RecordReader;
+import org.apache.nifi.serialization.RecordReaderFactory;
+import org.apache.nifi.serialization.record.Record;
+import org.apache.nifi.util.Tuple;
+
+@Tags({"iotdb", "insert", "tablet"})
+@InputRequirement(InputRequirement.Requirement.INPUT_REQUIRED)
+@CapabilityDescription(
+ "This is a record aware processor that reads the content of the incoming FlowFile as individual records using the "
+ + "configured 'Record Reader' and writes them to Apache IoTDB using native interface.")
+public class PutIoTDB extends AbstractIoTDB {
+
+ static final PropertyDescriptor RECORD_READER_FACTORY =
+ new PropertyDescriptor.Builder()
+ .name("Record Reader")
+ .description(
+ "Specifies the type of Record Reader controller service to use for parsing the incoming data "
+ + "and determining the schema")
+ .identifiesControllerService(RecordReaderFactory.class)
+ .required(true)
+ .build();
+
+ static final PropertyDescriptor TIME_FIELD =
+ new PropertyDescriptor.Builder()
+ .name("Time Field")
+ .description(
+ "The field name which represents time. It can be updated by expression language.")
+ .defaultValue("Time")
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .required(false)
+ .build();
+
+ static final PropertyDescriptor SCHEMA =
+ new PropertyDescriptor.Builder()
+ .name("Schema")
+ .description(
+ "The schema that IoTDB needs doesn't support good by NiFi.\n"
+ + "Therefore, you can define the schema here.\n"
+ + "Besides, you can set encoding type and compression type by this method.\n"
+ + "If you don't set this property, the inferred schema will be used.\n"
+ + "It can be updated by expression language.")
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .required(false)
+ .build();
+
+ static final PropertyDescriptor ALIGNED =
+ new PropertyDescriptor.Builder()
+ .name("Aligned")
+ .description("Whether using aligned interface? It can be updated by expression language.")
+ .allowableValues(new String[] {"true", "false"})
+ .addValidator(StandardValidators.BOOLEAN_VALIDATOR)
+ .defaultValue("false")
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .required(false)
+ .build();
+
+ static final PropertyDescriptor MAX_ROW_NUMBER =
+ new PropertyDescriptor.Builder()
+ .name("Max Row Number")
+ .description(
+ "Specifies the max row number of each tablet. It can be updated by expression language.")
+ .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
+ .defaultValue("false")
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .required(false)
+ .build();
+
+ static {
+ descriptors.add(RECORD_READER_FACTORY);
+ descriptors.add(TIME_FIELD);
+ descriptors.add(SCHEMA);
+ descriptors.add(ALIGNED);
+ descriptors.add(MAX_ROW_NUMBER);
+ }
+
+ @Override
+ protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ return descriptors;
+ }
+
+ @Override
+ public Set<Relationship> getRelationships() {
+ return relationships;
+ }
+
+ @Override
+ public void onTrigger(ProcessContext processContext, ProcessSession processSession)
+ throws ProcessException {
+ final RecordReaderFactory recordParserFactory =
+ processContext
+ .getProperty(RECORD_READER_FACTORY)
+ .asControllerService(RecordReaderFactory.class);
+
+ String timeFieldProperty = processContext.getProperty(TIME_FIELD).getValue();
+ String schemaProperty = processContext.getProperty(SCHEMA).getValue();
+ String alignedProperty = processContext.getProperty(ALIGNED).getValue();
+ String maxRowNumberProperty = processContext.getProperty(MAX_ROW_NUMBER).getValue();
+
+ FlowFile flowFile = processSession.get();
+
+ String timeField = timeFieldProperty != null ? timeFieldProperty : "Time";
+ Boolean aligned = alignedProperty != null ? Boolean.valueOf(alignedProperty) : false;
+ int maxRowNumber = maxRowNumberProperty != null ? Integer.valueOf(maxRowNumberProperty) : 1024;
+
+ try (final InputStream inputStream = processSession.read(flowFile);
+ final RecordReader recordReader =
+ recordParserFactory.createRecordReader(flowFile, inputStream, getLogger())) {
+ if (flowFile == null) {
+ inputStream.close();
+ recordReader.close();
+ processSession.transfer(flowFile, REL_SUCCESS);
+ return;
+ }
+
+ HashMap<String, Tablet> tablets;
+ boolean needInitFormatter;
+ Schema schema;
+ Tuple<Boolean, String> result;
+
+ result =
+ schemaProperty != null
+ ? validateSchemaAttribute(schemaProperty)
+ : validateSchema(recordReader.getSchema());
+
+ if (!result.getKey()) {
+ getLogger().error(result.getValue());
+ inputStream.close();
+ recordReader.close();
+ processSession.transfer(flowFile, REL_FAILURE);
+ return;
+ } else {
+ if (result.getValue() != null) {
+ getLogger().warn(result.getValue());
+ }
+ }
+
+ schema =
+ schemaProperty != null
+ ? new ObjectMapper().readValue(schemaProperty, Schema.class)
Review Comment:
`ObjectMapper` should be declared as a static member and reused.
##########
nifi-nar-bundles/nifi-iotdb-bundle/nifi-iotdb-processors/src/main/java/org/apache/nifi/processors/model/Schema.java:
##########
@@ -0,0 +1,99 @@
+/*
+ * 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.model;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Set;
+import java.util.HashSet;
+import java.util.stream.Collectors;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import org.apache.iotdb.tsfile.file.metadata.enums.CompressionType;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSEncoding;
+
+public class Schema {
Review Comment:
It would be helpful to name this class more specifically given the various types of schemas in the context of Apache NiFi. What do you think of `DatabaseSchema` or `TimeDatabaseSchema`?
##########
nifi-nar-bundles/nifi-iotdb-bundle/nifi-iotdb-processors/src/main/java/org/apache/nifi/processors/PutIoTDB.java:
##########
@@ -0,0 +1,286 @@
+/*
+ * 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;
+
+import java.io.InputStream;
+import java.text.SimpleDateFormat;
+import java.util.Set;
+import java.util.List;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.nifi.processors.model.Schema;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.write.record.Tablet;
+import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnUnscheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.expression.ExpressionLanguageScope;
+import org.apache.nifi.flowfile.FlowFile;
+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.serialization.RecordReader;
+import org.apache.nifi.serialization.RecordReaderFactory;
+import org.apache.nifi.serialization.record.Record;
+import org.apache.nifi.util.Tuple;
+
+@Tags({"iotdb", "insert", "tablet"})
+@InputRequirement(InputRequirement.Requirement.INPUT_REQUIRED)
+@CapabilityDescription(
+ "This is a record aware processor that reads the content of the incoming FlowFile as individual records using the "
+ + "configured 'Record Reader' and writes them to Apache IoTDB using native interface.")
+public class PutIoTDB extends AbstractIoTDB {
+
+ static final PropertyDescriptor RECORD_READER_FACTORY =
+ new PropertyDescriptor.Builder()
+ .name("Record Reader")
+ .description(
+ "Specifies the type of Record Reader controller service to use for parsing the incoming data "
+ + "and determining the schema")
+ .identifiesControllerService(RecordReaderFactory.class)
+ .required(true)
+ .build();
+
+ static final PropertyDescriptor TIME_FIELD =
+ new PropertyDescriptor.Builder()
+ .name("Time Field")
+ .description(
+ "The field name which represents time. It can be updated by expression language.")
+ .defaultValue("Time")
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .required(false)
+ .build();
+
+ static final PropertyDescriptor SCHEMA =
+ new PropertyDescriptor.Builder()
+ .name("Schema")
+ .description(
+ "The schema that IoTDB needs doesn't support good by NiFi.\n"
+ + "Therefore, you can define the schema here.\n"
+ + "Besides, you can set encoding type and compression type by this method.\n"
+ + "If you don't set this property, the inferred schema will be used.\n"
+ + "It can be updated by expression language.")
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .required(false)
+ .build();
+
+ static final PropertyDescriptor ALIGNED =
+ new PropertyDescriptor.Builder()
+ .name("Aligned")
+ .description("Whether using aligned interface? It can be updated by expression language.")
+ .allowableValues(new String[] {"true", "false"})
+ .addValidator(StandardValidators.BOOLEAN_VALIDATOR)
+ .defaultValue("false")
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .required(false)
+ .build();
+
+ static final PropertyDescriptor MAX_ROW_NUMBER =
+ new PropertyDescriptor.Builder()
+ .name("Max Row Number")
+ .description(
+ "Specifies the max row number of each tablet. It can be updated by expression language.")
+ .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
+ .defaultValue("false")
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .required(false)
+ .build();
+
+ static {
+ descriptors.add(RECORD_READER_FACTORY);
+ descriptors.add(TIME_FIELD);
+ descriptors.add(SCHEMA);
+ descriptors.add(ALIGNED);
+ descriptors.add(MAX_ROW_NUMBER);
+ }
+
+ @Override
+ protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ return descriptors;
+ }
+
+ @Override
+ public Set<Relationship> getRelationships() {
+ return relationships;
+ }
+
+ @Override
+ public void onTrigger(ProcessContext processContext, ProcessSession processSession)
+ throws ProcessException {
+ final RecordReaderFactory recordParserFactory =
+ processContext
+ .getProperty(RECORD_READER_FACTORY)
+ .asControllerService(RecordReaderFactory.class);
+
+ String timeFieldProperty = processContext.getProperty(TIME_FIELD).getValue();
+ String schemaProperty = processContext.getProperty(SCHEMA).getValue();
+ String alignedProperty = processContext.getProperty(ALIGNED).getValue();
+ String maxRowNumberProperty = processContext.getProperty(MAX_ROW_NUMBER).getValue();
+
+ FlowFile flowFile = processSession.get();
+
+ String timeField = timeFieldProperty != null ? timeFieldProperty : "Time";
+ Boolean aligned = alignedProperty != null ? Boolean.valueOf(alignedProperty) : false;
+ int maxRowNumber = maxRowNumberProperty != null ? Integer.valueOf(maxRowNumberProperty) : 1024;
+
+ try (final InputStream inputStream = processSession.read(flowFile);
+ final RecordReader recordReader =
+ recordParserFactory.createRecordReader(flowFile, inputStream, getLogger())) {
+ if (flowFile == null) {
+ inputStream.close();
+ recordReader.close();
+ processSession.transfer(flowFile, REL_SUCCESS);
+ return;
+ }
+
+ HashMap<String, Tablet> tablets;
+ boolean needInitFormatter;
+ Schema schema;
+ Tuple<Boolean, String> result;
+
+ result =
+ schemaProperty != null
+ ? validateSchemaAttribute(schemaProperty)
+ : validateSchema(recordReader.getSchema());
+
+ if (!result.getKey()) {
+ getLogger().error(result.getValue());
+ inputStream.close();
+ recordReader.close();
+ processSession.transfer(flowFile, REL_FAILURE);
+ return;
+ } else {
+ if (result.getValue() != null) {
+ getLogger().warn(result.getValue());
+ }
+ }
Review Comment:
All log messages should have a readable message, in addition to passing the relevant value.
##########
nifi-nar-bundles/nifi-iotdb-bundle/nifi-iotdb-processors/src/main/java/org/apache/nifi/processors/PutIoTDB.java:
##########
@@ -0,0 +1,286 @@
+/*
+ * 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;
+
+import java.io.InputStream;
+import java.text.SimpleDateFormat;
+import java.util.Set;
+import java.util.List;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.nifi.processors.model.Schema;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.write.record.Tablet;
+import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnUnscheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.expression.ExpressionLanguageScope;
+import org.apache.nifi.flowfile.FlowFile;
+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.serialization.RecordReader;
+import org.apache.nifi.serialization.RecordReaderFactory;
+import org.apache.nifi.serialization.record.Record;
+import org.apache.nifi.util.Tuple;
+
+@Tags({"iotdb", "insert", "tablet"})
+@InputRequirement(InputRequirement.Requirement.INPUT_REQUIRED)
+@CapabilityDescription(
+ "This is a record aware processor that reads the content of the incoming FlowFile as individual records using the "
+ + "configured 'Record Reader' and writes them to Apache IoTDB using native interface.")
+public class PutIoTDB extends AbstractIoTDB {
+
+ static final PropertyDescriptor RECORD_READER_FACTORY =
+ new PropertyDescriptor.Builder()
+ .name("Record Reader")
+ .description(
+ "Specifies the type of Record Reader controller service to use for parsing the incoming data "
+ + "and determining the schema")
+ .identifiesControllerService(RecordReaderFactory.class)
+ .required(true)
+ .build();
+
+ static final PropertyDescriptor TIME_FIELD =
+ new PropertyDescriptor.Builder()
+ .name("Time Field")
+ .description(
+ "The field name which represents time. It can be updated by expression language.")
+ .defaultValue("Time")
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .required(false)
+ .build();
+
+ static final PropertyDescriptor SCHEMA =
+ new PropertyDescriptor.Builder()
+ .name("Schema")
+ .description(
+ "The schema that IoTDB needs doesn't support good by NiFi.\n"
+ + "Therefore, you can define the schema here.\n"
+ + "Besides, you can set encoding type and compression type by this method.\n"
+ + "If you don't set this property, the inferred schema will be used.\n"
+ + "It can be updated by expression language.")
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .required(false)
+ .build();
+
+ static final PropertyDescriptor ALIGNED =
+ new PropertyDescriptor.Builder()
+ .name("Aligned")
+ .description("Whether using aligned interface? It can be updated by expression language.")
+ .allowableValues(new String[] {"true", "false"})
+ .addValidator(StandardValidators.BOOLEAN_VALIDATOR)
+ .defaultValue("false")
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .required(false)
+ .build();
+
+ static final PropertyDescriptor MAX_ROW_NUMBER =
+ new PropertyDescriptor.Builder()
+ .name("Max Row Number")
+ .description(
+ "Specifies the max row number of each tablet. It can be updated by expression language.")
+ .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
+ .defaultValue("false")
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .required(false)
+ .build();
+
+ static {
+ descriptors.add(RECORD_READER_FACTORY);
+ descriptors.add(TIME_FIELD);
+ descriptors.add(SCHEMA);
+ descriptors.add(ALIGNED);
+ descriptors.add(MAX_ROW_NUMBER);
+ }
+
+ @Override
+ protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ return descriptors;
+ }
+
+ @Override
+ public Set<Relationship> getRelationships() {
+ return relationships;
+ }
+
+ @Override
+ public void onTrigger(ProcessContext processContext, ProcessSession processSession)
+ throws ProcessException {
+ final RecordReaderFactory recordParserFactory =
+ processContext
+ .getProperty(RECORD_READER_FACTORY)
+ .asControllerService(RecordReaderFactory.class);
+
+ String timeFieldProperty = processContext.getProperty(TIME_FIELD).getValue();
+ String schemaProperty = processContext.getProperty(SCHEMA).getValue();
+ String alignedProperty = processContext.getProperty(ALIGNED).getValue();
+ String maxRowNumberProperty = processContext.getProperty(MAX_ROW_NUMBER).getValue();
+
+ FlowFile flowFile = processSession.get();
+
+ String timeField = timeFieldProperty != null ? timeFieldProperty : "Time";
+ Boolean aligned = alignedProperty != null ? Boolean.valueOf(alignedProperty) : false;
+ int maxRowNumber = maxRowNumberProperty != null ? Integer.valueOf(maxRowNumberProperty) : 1024;
+
+ try (final InputStream inputStream = processSession.read(flowFile);
+ final RecordReader recordReader =
+ recordParserFactory.createRecordReader(flowFile, inputStream, getLogger())) {
+ if (flowFile == null) {
+ inputStream.close();
+ recordReader.close();
+ processSession.transfer(flowFile, REL_SUCCESS);
+ return;
+ }
+
+ HashMap<String, Tablet> tablets;
+ boolean needInitFormatter;
+ Schema schema;
+ Tuple<Boolean, String> result;
+
+ result =
+ schemaProperty != null
+ ? validateSchemaAttribute(schemaProperty)
+ : validateSchema(recordReader.getSchema());
+
+ if (!result.getKey()) {
+ getLogger().error(result.getValue());
+ inputStream.close();
+ recordReader.close();
+ processSession.transfer(flowFile, REL_FAILURE);
+ return;
+ } else {
+ if (result.getValue() != null) {
+ getLogger().warn(result.getValue());
+ }
+ }
+
+ schema =
+ schemaProperty != null
+ ? new ObjectMapper().readValue(schemaProperty, Schema.class)
+ : convertSchema(recordReader.getSchema());
+
+ List<String> fieldNames = schema.getFieldNames();
+
+ needInitFormatter = schema.getTimeType() != Schema.TimeType.LONG;
+
+ tablets = generateTablets(schema, maxRowNumber);
+ SimpleDateFormat timeFormatter = null;
+
+ Record record;
+
+ while ((record = recordReader.nextRecord()) != null) {
+ Object[] values = record.getValues();
+ if (timeFormatter == null && needInitFormatter) {
+ timeFormatter = initFormatter((String) values[0]);
+ if (timeFormatter == null) {
+ getLogger().error("The format of time is not supported.");
+ inputStream.close();
+ recordReader.close();
+ processSession.transfer(flowFile, REL_FAILURE);
+ return;
+ }
+ }
+
+ long timestamp;
+ if (needInitFormatter) {
+ timestamp = timeFormatter.parse((String) values[0]).getTime();
+ } else {
+ timestamp = (Long) values[0];
+ }
+
+ boolean flag = false;
Review Comment:
The purpose of this variable is somewhat unclear based on the name.
##########
nifi-nar-bundles/nifi-iotdb-bundle/nifi-iotdb-processors/src/main/java/org/apache/nifi/processors/AbstractIoTDB.java:
##########
@@ -0,0 +1,442 @@
+/*
+ * 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;
+
+import java.text.SimpleDateFormat;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicReference;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.nifi.processors.model.Field;
+import org.apache.nifi.processors.model.Schema;
+import org.apache.iotdb.rpc.IoTDBConnectionException;
+import org.apache.iotdb.session.Session;
+import org.apache.iotdb.tsfile.file.metadata.enums.CompressionType;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSEncoding;
+import org.apache.iotdb.tsfile.utils.Binary;
+import org.apache.iotdb.tsfile.write.record.Tablet;
+import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.serialization.record.DataType;
+import org.apache.nifi.serialization.record.RecordFieldType;
+import org.apache.nifi.serialization.record.RecordSchema;
+import org.apache.nifi.util.Tuple;
+
+public abstract class AbstractIoTDB extends AbstractProcessor {
+ private static final int DEFAULT_IOTDB_PORT = 6667;
+
+ private static Map<RecordFieldType, TSDataType> typeMap =
+ new HashMap<>();
+
+ static final Set<RecordFieldType> supportedType =
+ new HashSet<>();
+
+ static final Set<RecordFieldType> supportedTimeType =
+ new HashSet<>();
+
+ private static final String[] STRING_TIME_FORMAT =
+ new String[]{
+ "yyyy-MM-dd HH:mm:ss.SSSX",
+ "yyyy/MM/dd HH:mm:ss.SSSX",
+ "yyyy.MM.dd HH:mm:ss.SSSX",
+ "yyyy-MM-dd HH:mm:ssX",
+ "yyyy/MM/dd HH:mm:ssX",
+ "yyyy.MM.dd HH:mm:ssX",
+ "yyyy-MM-dd HH:mm:ss.SSSz",
+ "yyyy/MM/dd HH:mm:ss.SSSz",
+ "yyyy.MM.dd HH:mm:ss.SSSz",
+ "yyyy-MM-dd HH:mm:ssz",
+ "yyyy/MM/dd HH:mm:ssz",
+ "yyyy.MM.dd HH:mm:ssz",
+ "yyyy-MM-dd HH:mm:ss.SSS",
+ "yyyy/MM/dd HH:mm:ss.SSS",
+ "yyyy.MM.dd HH:mm:ss.SSS",
+ "yyyy-MM-dd HH:mm:ss",
+ "yyyy/MM/dd HH:mm:ss",
+ "yyyy.MM.dd HH:mm:ss",
+ "yyyy-MM-dd'T'HH:mm:ss.SSSX",
+ "yyyy/MM/dd'T'HH:mm:ss.SSSX",
+ "yyyy.MM.dd'T'HH:mm:ss.SSSX",
+ "yyyy-MM-dd'T'HH:mm:ssX",
+ "yyyy/MM/dd'T'HH:mm:ssX",
+ "yyyy.MM.dd'T'HH:mm:ssX",
+ "yyyy-MM-dd'T'HH:mm:ss.SSSz",
+ "yyyy/MM/dd'T'HH:mm:ss.SSSz",
+ "yyyy.MM.dd'T'HH:mm:ss.SSSz",
+ "yyyy-MM-dd'T'HH:mm:ssz",
+ "yyyy/MM/dd'T'HH:mm:ssz",
+ "yyyy.MM.dd'T'HH:mm:ssz",
+ "yyyy-MM-dd'T'HH:mm:ss.SSS",
+ "yyyy/MM/dd'T'HH:mm:ss.SSS",
+ "yyyy.MM.dd'T'HH:mm:ss.SSS",
+ "yyyy-MM-dd'T'HH:mm:ss",
+ "yyyy/MM/dd'T'HH:mm:ss",
+ "yyyy.MM.dd'T'HH:mm:ss"
+ };
+
+ static final PropertyDescriptor IOTDB_HOST =
+ new PropertyDescriptor.Builder()
+ .name("Host")
+ .description("The host of IoTDB.")
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .required(true)
+ .build();
+
+ static final PropertyDescriptor IOTDB_PORT =
+ new PropertyDescriptor.Builder()
+ .name("Port")
+ .description("The port of IoTDB.")
+ .defaultValue(String.valueOf(DEFAULT_IOTDB_PORT))
+ .addValidator(StandardValidators.PORT_VALIDATOR)
+ .required(true)
+ .build();
+
+ static final PropertyDescriptor USERNAME =
+ new PropertyDescriptor.Builder()
+ .name("Username")
+ .description("Username to access the IoTDB.")
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .required(true)
+ .build();
+
+ static final PropertyDescriptor PASSWORD =
+ new PropertyDescriptor.Builder()
+ .name("Password")
+ .description("Password to access the IoTDB.")
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .required(true)
+ .sensitive(true)
+ .build();
+
+ protected static List<PropertyDescriptor> descriptors = new ArrayList<>();
+
+ protected static final Relationship REL_SUCCESS =
+ new Relationship.Builder()
+ .name("success")
+ .description("files that were successfully processed")
+ .build();
+ protected static final Relationship REL_FAILURE =
+ new Relationship.Builder()
+ .name("failure")
+ .description("files that were not successfully processed")
+ .build();
+
+ protected static Set<Relationship> relationships = new HashSet<>();
+
+ static {
+ descriptors.add(IOTDB_HOST);
+ descriptors.add(IOTDB_PORT);
+ descriptors.add(USERNAME);
+ descriptors.add(PASSWORD);
+
+ relationships.add(REL_SUCCESS);
+ relationships.add(REL_FAILURE);
+
+ typeMap.put(RecordFieldType.STRING, TSDataType.TEXT);
+ typeMap.put(RecordFieldType.BOOLEAN, TSDataType.BOOLEAN);
+ typeMap.put(RecordFieldType.INT, TSDataType.INT32);
+ typeMap.put(RecordFieldType.LONG, TSDataType.INT64);
+ typeMap.put(RecordFieldType.FLOAT, TSDataType.FLOAT);
+ typeMap.put(RecordFieldType.DOUBLE, TSDataType.DOUBLE);
+
+ supportedType.add(RecordFieldType.BOOLEAN);
+ supportedType.add(RecordFieldType.STRING);
+ supportedType.add(RecordFieldType.INT);
+ supportedType.add(RecordFieldType.LONG);
+ supportedType.add(RecordFieldType.FLOAT);
+ supportedType.add(RecordFieldType.DOUBLE);
+
+ supportedTimeType.add(RecordFieldType.STRING);
+ supportedTimeType.add(RecordFieldType.LONG);
+ }
+
+ protected final AtomicReference<Session> session = new AtomicReference<>(null);
+
+ @OnScheduled
+ public void onScheduled(ProcessContext context) {
+ connectToIoTDB(context);
+ }
+
+ void connectToIoTDB(ProcessContext context) {
+ if (session.get() == null) {
+ final String host, username, password;
+ final int port;
+
+ host = context.getProperty(IOTDB_HOST).getValue();
+ port = Integer.parseInt(context.getProperty(IOTDB_PORT).getValue());
+ username = context.getProperty(USERNAME).getValue();
+ password = context.getProperty(PASSWORD).getValue();
+
+ session.set(
+ new Session.Builder()
+ .host(host)
+ .port(port)
+ .username(username)
+ .password(password)
+ .build());
+ try {
+ session.get().open();
+ } catch (IoTDBConnectionException e) {
Review Comment:
In addition to logging the message, it would be best to throw an exception, because the processor will not be functional without a working connection to IoTDB.
##########
nifi-nar-bundles/nifi-iotdb-bundle/nifi-iotdb-processors/pom.xml:
##########
@@ -0,0 +1,116 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ 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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <parent>
+ <artifactId>nifi-iotdb-bundle</artifactId>
+ <groupId>org.apache.nifi</groupId>
+ <version>1.18.0-SNAPSHOT</version>
+ </parent>
+ <modelVersion>4.0.0</modelVersion>
+
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-iotdb-processors</artifactId>
+ <packaging>jar</packaging>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.iotdb</groupId>
+ <artifactId>iotdb-session</artifactId>
+ <version>${iotdb.sdk.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.iotdb</groupId>
+ <artifactId>iotdb-server</artifactId>
+ <version>${iotdb.sdk.version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.iotdb</groupId>
+ <artifactId>iotdb-server</artifactId>
+ <version>${iotdb.sdk.version}</version>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-api</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-mock</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-record-serialization-service-api</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-record</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-mock-record-utils</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-dbcp-service-api</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-standard-services-api-nar</artifactId>
+ <version>${project.version}</version>
+ <type>nar</type>
+ </dependency>
+
+ <dependency>
+ <groupId>commons-io</groupId>
+ <artifactId>commons-io</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.commons</groupId>
+ <artifactId>commons-lang3</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-core</artifactId>
+ <version>${jackson.bom.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-annotations</artifactId>
+ <version>${jackson.bom.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-databind</artifactId>
+ <version>${jackson.bom.version}</version>
Review Comment:
The `version` property can be removed since the Bill-of-Materials dependency in the root configuration manages the version.
##########
nifi-nar-bundles/nifi-iotdb-bundle/nifi-iotdb-processors/src/main/java/org/apache/nifi/processors/AbstractIoTDB.java:
##########
@@ -0,0 +1,442 @@
+/*
+ * 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;
+
+import java.text.SimpleDateFormat;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicReference;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.nifi.processors.model.Field;
+import org.apache.nifi.processors.model.Schema;
+import org.apache.iotdb.rpc.IoTDBConnectionException;
+import org.apache.iotdb.session.Session;
+import org.apache.iotdb.tsfile.file.metadata.enums.CompressionType;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSEncoding;
+import org.apache.iotdb.tsfile.utils.Binary;
+import org.apache.iotdb.tsfile.write.record.Tablet;
+import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.serialization.record.DataType;
+import org.apache.nifi.serialization.record.RecordFieldType;
+import org.apache.nifi.serialization.record.RecordSchema;
+import org.apache.nifi.util.Tuple;
+
+public abstract class AbstractIoTDB extends AbstractProcessor {
+ private static final int DEFAULT_IOTDB_PORT = 6667;
+
+ private static Map<RecordFieldType, TSDataType> typeMap =
+ new HashMap<>();
+
+ static final Set<RecordFieldType> supportedType =
+ new HashSet<>();
+
+ static final Set<RecordFieldType> supportedTimeType =
+ new HashSet<>();
+
+ private static final String[] STRING_TIME_FORMAT =
+ new String[]{
+ "yyyy-MM-dd HH:mm:ss.SSSX",
+ "yyyy/MM/dd HH:mm:ss.SSSX",
+ "yyyy.MM.dd HH:mm:ss.SSSX",
+ "yyyy-MM-dd HH:mm:ssX",
+ "yyyy/MM/dd HH:mm:ssX",
+ "yyyy.MM.dd HH:mm:ssX",
+ "yyyy-MM-dd HH:mm:ss.SSSz",
+ "yyyy/MM/dd HH:mm:ss.SSSz",
+ "yyyy.MM.dd HH:mm:ss.SSSz",
+ "yyyy-MM-dd HH:mm:ssz",
+ "yyyy/MM/dd HH:mm:ssz",
+ "yyyy.MM.dd HH:mm:ssz",
+ "yyyy-MM-dd HH:mm:ss.SSS",
+ "yyyy/MM/dd HH:mm:ss.SSS",
+ "yyyy.MM.dd HH:mm:ss.SSS",
+ "yyyy-MM-dd HH:mm:ss",
+ "yyyy/MM/dd HH:mm:ss",
+ "yyyy.MM.dd HH:mm:ss",
+ "yyyy-MM-dd'T'HH:mm:ss.SSSX",
+ "yyyy/MM/dd'T'HH:mm:ss.SSSX",
+ "yyyy.MM.dd'T'HH:mm:ss.SSSX",
+ "yyyy-MM-dd'T'HH:mm:ssX",
+ "yyyy/MM/dd'T'HH:mm:ssX",
+ "yyyy.MM.dd'T'HH:mm:ssX",
+ "yyyy-MM-dd'T'HH:mm:ss.SSSz",
+ "yyyy/MM/dd'T'HH:mm:ss.SSSz",
+ "yyyy.MM.dd'T'HH:mm:ss.SSSz",
+ "yyyy-MM-dd'T'HH:mm:ssz",
+ "yyyy/MM/dd'T'HH:mm:ssz",
+ "yyyy.MM.dd'T'HH:mm:ssz",
+ "yyyy-MM-dd'T'HH:mm:ss.SSS",
+ "yyyy/MM/dd'T'HH:mm:ss.SSS",
+ "yyyy.MM.dd'T'HH:mm:ss.SSS",
+ "yyyy-MM-dd'T'HH:mm:ss",
+ "yyyy/MM/dd'T'HH:mm:ss",
+ "yyyy.MM.dd'T'HH:mm:ss"
+ };
+
+ static final PropertyDescriptor IOTDB_HOST =
+ new PropertyDescriptor.Builder()
+ .name("Host")
+ .description("The host of IoTDB.")
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .required(true)
+ .build();
+
+ static final PropertyDescriptor IOTDB_PORT =
+ new PropertyDescriptor.Builder()
+ .name("Port")
+ .description("The port of IoTDB.")
+ .defaultValue(String.valueOf(DEFAULT_IOTDB_PORT))
+ .addValidator(StandardValidators.PORT_VALIDATOR)
+ .required(true)
+ .build();
+
+ static final PropertyDescriptor USERNAME =
+ new PropertyDescriptor.Builder()
+ .name("Username")
+ .description("Username to access the IoTDB.")
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .required(true)
+ .build();
+
+ static final PropertyDescriptor PASSWORD =
+ new PropertyDescriptor.Builder()
+ .name("Password")
+ .description("Password to access the IoTDB.")
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .required(true)
+ .sensitive(true)
+ .build();
+
+ protected static List<PropertyDescriptor> descriptors = new ArrayList<>();
+
+ protected static final Relationship REL_SUCCESS =
+ new Relationship.Builder()
+ .name("success")
+ .description("files that were successfully processed")
+ .build();
+ protected static final Relationship REL_FAILURE =
+ new Relationship.Builder()
+ .name("failure")
+ .description("files that were not successfully processed")
+ .build();
+
+ protected static Set<Relationship> relationships = new HashSet<>();
+
+ static {
+ descriptors.add(IOTDB_HOST);
+ descriptors.add(IOTDB_PORT);
+ descriptors.add(USERNAME);
+ descriptors.add(PASSWORD);
+
+ relationships.add(REL_SUCCESS);
+ relationships.add(REL_FAILURE);
+
+ typeMap.put(RecordFieldType.STRING, TSDataType.TEXT);
+ typeMap.put(RecordFieldType.BOOLEAN, TSDataType.BOOLEAN);
+ typeMap.put(RecordFieldType.INT, TSDataType.INT32);
+ typeMap.put(RecordFieldType.LONG, TSDataType.INT64);
+ typeMap.put(RecordFieldType.FLOAT, TSDataType.FLOAT);
+ typeMap.put(RecordFieldType.DOUBLE, TSDataType.DOUBLE);
+
+ supportedType.add(RecordFieldType.BOOLEAN);
+ supportedType.add(RecordFieldType.STRING);
+ supportedType.add(RecordFieldType.INT);
+ supportedType.add(RecordFieldType.LONG);
+ supportedType.add(RecordFieldType.FLOAT);
+ supportedType.add(RecordFieldType.DOUBLE);
+
+ supportedTimeType.add(RecordFieldType.STRING);
+ supportedTimeType.add(RecordFieldType.LONG);
+ }
+
+ protected final AtomicReference<Session> session = new AtomicReference<>(null);
+
+ @OnScheduled
+ public void onScheduled(ProcessContext context) {
+ connectToIoTDB(context);
+ }
+
+ void connectToIoTDB(ProcessContext context) {
+ if (session.get() == null) {
+ final String host, username, password;
+ final int port;
+
+ host = context.getProperty(IOTDB_HOST).getValue();
+ port = Integer.parseInt(context.getProperty(IOTDB_PORT).getValue());
+ username = context.getProperty(USERNAME).getValue();
+ password = context.getProperty(PASSWORD).getValue();
+
+ session.set(
+ new Session.Builder()
+ .host(host)
+ .port(port)
+ .username(username)
+ .password(password)
+ .build());
+ try {
+ session.get().open();
+ } catch (IoTDBConnectionException e) {
+ getLogger().error(e.getMessage());
+ e.printStackTrace();
+ }
+ }
+ }
+
+ public void stop(ProcessContext context) {
+ if (session.get() != null) {
+ try {
+ session.get().close();
+ } catch (IoTDBConnectionException e) {
+ e.printStackTrace();
+ }
+ session.set(null);
+ }
+ }
+
+ protected TSDataType getType(RecordFieldType type) {
+ return typeMap.get(type);
+ }
+
+ protected Tuple<Boolean, String> validateSchemaAttribute(String schemaAttribute) {
+ ObjectMapper mapper = new ObjectMapper();
+ JsonNode schema = null;
+ try {
+ schema = mapper.readTree(schemaAttribute);
+ } catch (JsonProcessingException e) {
+ return new Tuple<>(false, e.getMessage());
+ }
+ Set<String> keySet = new HashSet<>();
+ schema.fieldNames().forEachRemaining(field -> keySet.add(field));
+
+ if (!keySet.contains("timeType") || !keySet.contains("fields")) {
+ String msg = "The JSON of schema must contain `timeType` and `fields`.";
+ return new Tuple<>(false, msg);
+ }
+
+ if (!Schema.getSupportedTimeType().contains(schema.get("timeType").asText().toLowerCase())) {
+ String msg =
+ String.format(
+ "Unknown `timeType`: %s, there are only two options `long` and `string` for this property.",
+ schema.get("timeType").asText());
+ return new Tuple<>(false, msg);
+ }
+
+ for (int i = 0; i < schema.get("fields").size(); i++) {
+ JsonNode field = schema.get("fields").get(i);
+ Set<String> fieldKeySet = new HashSet<>();
+
+ field.fieldNames().forEachRemaining(fieldName -> fieldKeySet.add(fieldName));
+ if (!fieldKeySet.contains("tsName") || !fieldKeySet.contains("dataType")) {
+ String msg = "`tsName` or `dataType` has not been set.";
+ return new Tuple<>(false, msg);
+ }
+
+ if (!field.get("tsName").asText().startsWith("root.")) {
+ String msg =
+ String.format("The tsName `%s` is not start with 'root.'.", field.get("tsName").asText());
+ return new Tuple<>(false, msg);
+ }
+
+ if (!Field.getSupportedDataType().contains(field.get("dataType").asText())) {
+ String msg =
+ String.format(
+ "Unknown `dataType`: %s. The supported dataTypes are %s",
+ field.get("dataType").asText(), Field.getSupportedDataType());
+ return new Tuple<>(false, msg);
+ }
+
+ Set<String> supportedKeySet = new HashSet<>();
+ supportedKeySet.add("tsName");
+ supportedKeySet.add("dataType");
+ supportedKeySet.add("encoding");
+ supportedKeySet.add("compressionType");
+
+ HashSet<String> tmpKetSet = new HashSet<>();
+ tmpKetSet.addAll(supportedKeySet);
+ tmpKetSet.addAll(fieldKeySet);
+ tmpKetSet.removeAll(supportedKeySet);
+ if (!tmpKetSet.isEmpty()) {
+ String msg = "Unknown property or properties: " + tmpKetSet;
+ return new Tuple<>(false, msg);
+ }
+
+ if (fieldKeySet.contains("compressionType") && !fieldKeySet.contains("encoding")) {
+ String msg =
+ "The `compressionType` has been set, but the `encoding` has not. The property `compressionType` will not take effect.";
+ return new Tuple<>(true, msg);
+ }
+
+ if (field.get("encoding") != null
+ && !Field.getSupportedEncoding().contains(field.get("encoding").asText())) {
+ String msg =
+ String.format(
+ "Unknown `encoding`: %s, The supported encoding types are %s",
+ field.get("encoding").asText(), Field.getSupportedEncoding());
+ return new Tuple<>(false, msg);
+ }
+
+ if (field.get("compressionType") != null
+ && !Field.getSupportedCompressionType().contains(field.get("compressionType"))) {
+ String msg =
+ String.format(
+ "Unknown `compressionType`: %s, The supported compressionType are %s",
+ field.get("compressionType").asText(), Field.getSupportedCompressionType());
+ return new Tuple<>(false, msg);
+ }
+ }
+
+ return new Tuple<>(true, null);
+ }
+
+ protected Tuple<Boolean, String> validateSchema(RecordSchema recordSchema) {
+ List<String> fieldNames = recordSchema.getFieldNames();
+ List<DataType> dataTypes = recordSchema.getDataTypes();
+ if (!fieldNames.contains("Time")) {
+ return new Tuple<>(false, "The fields must contain `Time`.");
Review Comment:
It is best to avoid the period character in messages:
```suggestion
return new Tuple<>(false, "The fields must contain `Time`");
```
##########
nifi-nar-bundles/nifi-iotdb-bundle/nifi-iotdb-processors/pom.xml:
##########
@@ -0,0 +1,116 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ 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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <parent>
+ <artifactId>nifi-iotdb-bundle</artifactId>
+ <groupId>org.apache.nifi</groupId>
+ <version>1.18.0-SNAPSHOT</version>
+ </parent>
+ <modelVersion>4.0.0</modelVersion>
+
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-iotdb-processors</artifactId>
+ <packaging>jar</packaging>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.iotdb</groupId>
+ <artifactId>iotdb-session</artifactId>
+ <version>${iotdb.sdk.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.iotdb</groupId>
+ <artifactId>iotdb-server</artifactId>
+ <version>${iotdb.sdk.version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.iotdb</groupId>
+ <artifactId>iotdb-server</artifactId>
+ <version>${iotdb.sdk.version}</version>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-api</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-mock</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-record-serialization-service-api</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-record</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-mock-record-utils</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-dbcp-service-api</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-standard-services-api-nar</artifactId>
+ <version>${project.version}</version>
+ <type>nar</type>
+ </dependency>
+
+ <dependency>
+ <groupId>commons-io</groupId>
+ <artifactId>commons-io</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.commons</groupId>
+ <artifactId>commons-lang3</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-core</artifactId>
+ <version>${jackson.bom.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-annotations</artifactId>
+ <version>${jackson.bom.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-databind</artifactId>
+ <version>${jackson.bom.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>junit</groupId>
+ <artifactId>junit</artifactId>
+ </dependency>
Review Comment:
JUnit 5 is included as a default dependency, so this dependency on JUnit 4 should be removed.
##########
nifi-nar-bundles/nifi-iotdb-bundle/nifi-iotdb-processors/src/main/java/org/apache/nifi/processors/AbstractIoTDB.java:
##########
@@ -0,0 +1,442 @@
+/*
+ * 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;
+
+import java.text.SimpleDateFormat;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicReference;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.nifi.processors.model.Field;
+import org.apache.nifi.processors.model.Schema;
+import org.apache.iotdb.rpc.IoTDBConnectionException;
+import org.apache.iotdb.session.Session;
+import org.apache.iotdb.tsfile.file.metadata.enums.CompressionType;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSEncoding;
+import org.apache.iotdb.tsfile.utils.Binary;
+import org.apache.iotdb.tsfile.write.record.Tablet;
+import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.serialization.record.DataType;
+import org.apache.nifi.serialization.record.RecordFieldType;
+import org.apache.nifi.serialization.record.RecordSchema;
+import org.apache.nifi.util.Tuple;
+
+public abstract class AbstractIoTDB extends AbstractProcessor {
+ private static final int DEFAULT_IOTDB_PORT = 6667;
+
+ private static Map<RecordFieldType, TSDataType> typeMap =
+ new HashMap<>();
+
+ static final Set<RecordFieldType> supportedType =
+ new HashSet<>();
+
+ static final Set<RecordFieldType> supportedTimeType =
+ new HashSet<>();
+
+ private static final String[] STRING_TIME_FORMAT =
+ new String[]{
+ "yyyy-MM-dd HH:mm:ss.SSSX",
+ "yyyy/MM/dd HH:mm:ss.SSSX",
+ "yyyy.MM.dd HH:mm:ss.SSSX",
+ "yyyy-MM-dd HH:mm:ssX",
+ "yyyy/MM/dd HH:mm:ssX",
+ "yyyy.MM.dd HH:mm:ssX",
+ "yyyy-MM-dd HH:mm:ss.SSSz",
+ "yyyy/MM/dd HH:mm:ss.SSSz",
+ "yyyy.MM.dd HH:mm:ss.SSSz",
+ "yyyy-MM-dd HH:mm:ssz",
+ "yyyy/MM/dd HH:mm:ssz",
+ "yyyy.MM.dd HH:mm:ssz",
+ "yyyy-MM-dd HH:mm:ss.SSS",
+ "yyyy/MM/dd HH:mm:ss.SSS",
+ "yyyy.MM.dd HH:mm:ss.SSS",
+ "yyyy-MM-dd HH:mm:ss",
+ "yyyy/MM/dd HH:mm:ss",
+ "yyyy.MM.dd HH:mm:ss",
+ "yyyy-MM-dd'T'HH:mm:ss.SSSX",
+ "yyyy/MM/dd'T'HH:mm:ss.SSSX",
+ "yyyy.MM.dd'T'HH:mm:ss.SSSX",
+ "yyyy-MM-dd'T'HH:mm:ssX",
+ "yyyy/MM/dd'T'HH:mm:ssX",
+ "yyyy.MM.dd'T'HH:mm:ssX",
+ "yyyy-MM-dd'T'HH:mm:ss.SSSz",
+ "yyyy/MM/dd'T'HH:mm:ss.SSSz",
+ "yyyy.MM.dd'T'HH:mm:ss.SSSz",
+ "yyyy-MM-dd'T'HH:mm:ssz",
+ "yyyy/MM/dd'T'HH:mm:ssz",
+ "yyyy.MM.dd'T'HH:mm:ssz",
+ "yyyy-MM-dd'T'HH:mm:ss.SSS",
+ "yyyy/MM/dd'T'HH:mm:ss.SSS",
+ "yyyy.MM.dd'T'HH:mm:ss.SSS",
+ "yyyy-MM-dd'T'HH:mm:ss",
+ "yyyy/MM/dd'T'HH:mm:ss",
+ "yyyy.MM.dd'T'HH:mm:ss"
+ };
+
+ static final PropertyDescriptor IOTDB_HOST =
+ new PropertyDescriptor.Builder()
+ .name("Host")
+ .description("The host of IoTDB.")
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .required(true)
+ .build();
+
+ static final PropertyDescriptor IOTDB_PORT =
+ new PropertyDescriptor.Builder()
+ .name("Port")
+ .description("The port of IoTDB.")
+ .defaultValue(String.valueOf(DEFAULT_IOTDB_PORT))
+ .addValidator(StandardValidators.PORT_VALIDATOR)
+ .required(true)
+ .build();
+
+ static final PropertyDescriptor USERNAME =
+ new PropertyDescriptor.Builder()
+ .name("Username")
+ .description("Username to access the IoTDB.")
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .required(true)
+ .build();
+
+ static final PropertyDescriptor PASSWORD =
+ new PropertyDescriptor.Builder()
+ .name("Password")
+ .description("Password to access the IoTDB.")
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .required(true)
+ .sensitive(true)
+ .build();
+
+ protected static List<PropertyDescriptor> descriptors = new ArrayList<>();
+
+ protected static final Relationship REL_SUCCESS =
+ new Relationship.Builder()
+ .name("success")
+ .description("files that were successfully processed")
+ .build();
+ protected static final Relationship REL_FAILURE =
+ new Relationship.Builder()
+ .name("failure")
+ .description("files that were not successfully processed")
+ .build();
+
+ protected static Set<Relationship> relationships = new HashSet<>();
+
+ static {
+ descriptors.add(IOTDB_HOST);
+ descriptors.add(IOTDB_PORT);
+ descriptors.add(USERNAME);
+ descriptors.add(PASSWORD);
+
+ relationships.add(REL_SUCCESS);
+ relationships.add(REL_FAILURE);
+
+ typeMap.put(RecordFieldType.STRING, TSDataType.TEXT);
+ typeMap.put(RecordFieldType.BOOLEAN, TSDataType.BOOLEAN);
+ typeMap.put(RecordFieldType.INT, TSDataType.INT32);
+ typeMap.put(RecordFieldType.LONG, TSDataType.INT64);
+ typeMap.put(RecordFieldType.FLOAT, TSDataType.FLOAT);
+ typeMap.put(RecordFieldType.DOUBLE, TSDataType.DOUBLE);
+
+ supportedType.add(RecordFieldType.BOOLEAN);
+ supportedType.add(RecordFieldType.STRING);
+ supportedType.add(RecordFieldType.INT);
+ supportedType.add(RecordFieldType.LONG);
+ supportedType.add(RecordFieldType.FLOAT);
+ supportedType.add(RecordFieldType.DOUBLE);
+
+ supportedTimeType.add(RecordFieldType.STRING);
+ supportedTimeType.add(RecordFieldType.LONG);
+ }
+
+ protected final AtomicReference<Session> session = new AtomicReference<>(null);
+
+ @OnScheduled
+ public void onScheduled(ProcessContext context) {
+ connectToIoTDB(context);
+ }
+
+ void connectToIoTDB(ProcessContext context) {
+ if (session.get() == null) {
+ final String host, username, password;
+ final int port;
+
+ host = context.getProperty(IOTDB_HOST).getValue();
+ port = Integer.parseInt(context.getProperty(IOTDB_PORT).getValue());
+ username = context.getProperty(USERNAME).getValue();
+ password = context.getProperty(PASSWORD).getValue();
+
+ session.set(
+ new Session.Builder()
+ .host(host)
+ .port(port)
+ .username(username)
+ .password(password)
+ .build());
+ try {
+ session.get().open();
+ } catch (IoTDBConnectionException e) {
+ getLogger().error(e.getMessage());
+ e.printStackTrace();
+ }
+ }
+ }
+
+ public void stop(ProcessContext context) {
+ if (session.get() != null) {
+ try {
+ session.get().close();
+ } catch (IoTDBConnectionException e) {
+ e.printStackTrace();
+ }
+ session.set(null);
+ }
+ }
+
+ protected TSDataType getType(RecordFieldType type) {
+ return typeMap.get(type);
+ }
+
+ protected Tuple<Boolean, String> validateSchemaAttribute(String schemaAttribute) {
+ ObjectMapper mapper = new ObjectMapper();
+ JsonNode schema = null;
+ try {
+ schema = mapper.readTree(schemaAttribute);
+ } catch (JsonProcessingException e) {
+ return new Tuple<>(false, e.getMessage());
+ }
+ Set<String> keySet = new HashSet<>();
+ schema.fieldNames().forEachRemaining(field -> keySet.add(field));
+
+ if (!keySet.contains("timeType") || !keySet.contains("fields")) {
+ String msg = "The JSON of schema must contain `timeType` and `fields`.";
+ return new Tuple<>(false, msg);
+ }
+
+ if (!Schema.getSupportedTimeType().contains(schema.get("timeType").asText().toLowerCase())) {
+ String msg =
+ String.format(
+ "Unknown `timeType`: %s, there are only two options `long` and `string` for this property.",
+ schema.get("timeType").asText());
+ return new Tuple<>(false, msg);
+ }
+
+ for (int i = 0; i < schema.get("fields").size(); i++) {
+ JsonNode field = schema.get("fields").get(i);
+ Set<String> fieldKeySet = new HashSet<>();
+
+ field.fieldNames().forEachRemaining(fieldName -> fieldKeySet.add(fieldName));
+ if (!fieldKeySet.contains("tsName") || !fieldKeySet.contains("dataType")) {
+ String msg = "`tsName` or `dataType` has not been set.";
+ return new Tuple<>(false, msg);
+ }
+
+ if (!field.get("tsName").asText().startsWith("root.")) {
+ String msg =
+ String.format("The tsName `%s` is not start with 'root.'.", field.get("tsName").asText());
+ return new Tuple<>(false, msg);
+ }
+
+ if (!Field.getSupportedDataType().contains(field.get("dataType").asText())) {
+ String msg =
+ String.format(
+ "Unknown `dataType`: %s. The supported dataTypes are %s",
+ field.get("dataType").asText(), Field.getSupportedDataType());
+ return new Tuple<>(false, msg);
+ }
+
+ Set<String> supportedKeySet = new HashSet<>();
+ supportedKeySet.add("tsName");
+ supportedKeySet.add("dataType");
+ supportedKeySet.add("encoding");
+ supportedKeySet.add("compressionType");
+
+ HashSet<String> tmpKetSet = new HashSet<>();
+ tmpKetSet.addAll(supportedKeySet);
+ tmpKetSet.addAll(fieldKeySet);
+ tmpKetSet.removeAll(supportedKeySet);
+ if (!tmpKetSet.isEmpty()) {
+ String msg = "Unknown property or properties: " + tmpKetSet;
+ return new Tuple<>(false, msg);
+ }
+
+ if (fieldKeySet.contains("compressionType") && !fieldKeySet.contains("encoding")) {
+ String msg =
+ "The `compressionType` has been set, but the `encoding` has not. The property `compressionType` will not take effect.";
+ return new Tuple<>(true, msg);
+ }
+
+ if (field.get("encoding") != null
+ && !Field.getSupportedEncoding().contains(field.get("encoding").asText())) {
+ String msg =
+ String.format(
+ "Unknown `encoding`: %s, The supported encoding types are %s",
+ field.get("encoding").asText(), Field.getSupportedEncoding());
+ return new Tuple<>(false, msg);
+ }
+
+ if (field.get("compressionType") != null
+ && !Field.getSupportedCompressionType().contains(field.get("compressionType"))) {
+ String msg =
+ String.format(
+ "Unknown `compressionType`: %s, The supported compressionType are %s",
+ field.get("compressionType").asText(), Field.getSupportedCompressionType());
+ return new Tuple<>(false, msg);
+ }
+ }
+
+ return new Tuple<>(true, null);
+ }
+
+ protected Tuple<Boolean, String> validateSchema(RecordSchema recordSchema) {
+ List<String> fieldNames = recordSchema.getFieldNames();
+ List<DataType> dataTypes = recordSchema.getDataTypes();
+ if (!fieldNames.contains("Time")) {
Review Comment:
As with other strings, `Time` should be declared statically and reused.
##########
nifi-nar-bundles/nifi-iotdb-bundle/nifi-iotdb-processors/src/main/java/org/apache/nifi/processors/AbstractIoTDB.java:
##########
@@ -0,0 +1,442 @@
+/*
+ * 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;
+
+import java.text.SimpleDateFormat;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicReference;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.nifi.processors.model.Field;
+import org.apache.nifi.processors.model.Schema;
+import org.apache.iotdb.rpc.IoTDBConnectionException;
+import org.apache.iotdb.session.Session;
+import org.apache.iotdb.tsfile.file.metadata.enums.CompressionType;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSEncoding;
+import org.apache.iotdb.tsfile.utils.Binary;
+import org.apache.iotdb.tsfile.write.record.Tablet;
+import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.serialization.record.DataType;
+import org.apache.nifi.serialization.record.RecordFieldType;
+import org.apache.nifi.serialization.record.RecordSchema;
+import org.apache.nifi.util.Tuple;
+
+public abstract class AbstractIoTDB extends AbstractProcessor {
+ private static final int DEFAULT_IOTDB_PORT = 6667;
+
+ private static Map<RecordFieldType, TSDataType> typeMap =
+ new HashMap<>();
+
+ static final Set<RecordFieldType> supportedType =
+ new HashSet<>();
+
+ static final Set<RecordFieldType> supportedTimeType =
+ new HashSet<>();
+
+ private static final String[] STRING_TIME_FORMAT =
+ new String[]{
+ "yyyy-MM-dd HH:mm:ss.SSSX",
+ "yyyy/MM/dd HH:mm:ss.SSSX",
+ "yyyy.MM.dd HH:mm:ss.SSSX",
+ "yyyy-MM-dd HH:mm:ssX",
+ "yyyy/MM/dd HH:mm:ssX",
+ "yyyy.MM.dd HH:mm:ssX",
+ "yyyy-MM-dd HH:mm:ss.SSSz",
+ "yyyy/MM/dd HH:mm:ss.SSSz",
+ "yyyy.MM.dd HH:mm:ss.SSSz",
+ "yyyy-MM-dd HH:mm:ssz",
+ "yyyy/MM/dd HH:mm:ssz",
+ "yyyy.MM.dd HH:mm:ssz",
+ "yyyy-MM-dd HH:mm:ss.SSS",
+ "yyyy/MM/dd HH:mm:ss.SSS",
+ "yyyy.MM.dd HH:mm:ss.SSS",
+ "yyyy-MM-dd HH:mm:ss",
+ "yyyy/MM/dd HH:mm:ss",
+ "yyyy.MM.dd HH:mm:ss",
+ "yyyy-MM-dd'T'HH:mm:ss.SSSX",
+ "yyyy/MM/dd'T'HH:mm:ss.SSSX",
+ "yyyy.MM.dd'T'HH:mm:ss.SSSX",
+ "yyyy-MM-dd'T'HH:mm:ssX",
+ "yyyy/MM/dd'T'HH:mm:ssX",
+ "yyyy.MM.dd'T'HH:mm:ssX",
+ "yyyy-MM-dd'T'HH:mm:ss.SSSz",
+ "yyyy/MM/dd'T'HH:mm:ss.SSSz",
+ "yyyy.MM.dd'T'HH:mm:ss.SSSz",
+ "yyyy-MM-dd'T'HH:mm:ssz",
+ "yyyy/MM/dd'T'HH:mm:ssz",
+ "yyyy.MM.dd'T'HH:mm:ssz",
+ "yyyy-MM-dd'T'HH:mm:ss.SSS",
+ "yyyy/MM/dd'T'HH:mm:ss.SSS",
+ "yyyy.MM.dd'T'HH:mm:ss.SSS",
+ "yyyy-MM-dd'T'HH:mm:ss",
+ "yyyy/MM/dd'T'HH:mm:ss",
+ "yyyy.MM.dd'T'HH:mm:ss"
+ };
+
+ static final PropertyDescriptor IOTDB_HOST =
+ new PropertyDescriptor.Builder()
+ .name("Host")
+ .description("The host of IoTDB.")
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .required(true)
+ .build();
+
+ static final PropertyDescriptor IOTDB_PORT =
+ new PropertyDescriptor.Builder()
+ .name("Port")
+ .description("The port of IoTDB.")
+ .defaultValue(String.valueOf(DEFAULT_IOTDB_PORT))
+ .addValidator(StandardValidators.PORT_VALIDATOR)
+ .required(true)
+ .build();
+
+ static final PropertyDescriptor USERNAME =
+ new PropertyDescriptor.Builder()
+ .name("Username")
+ .description("Username to access the IoTDB.")
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .required(true)
+ .build();
+
+ static final PropertyDescriptor PASSWORD =
+ new PropertyDescriptor.Builder()
+ .name("Password")
+ .description("Password to access the IoTDB.")
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .required(true)
+ .sensitive(true)
+ .build();
+
+ protected static List<PropertyDescriptor> descriptors = new ArrayList<>();
+
+ protected static final Relationship REL_SUCCESS =
+ new Relationship.Builder()
+ .name("success")
+ .description("files that were successfully processed")
+ .build();
+ protected static final Relationship REL_FAILURE =
+ new Relationship.Builder()
+ .name("failure")
+ .description("files that were not successfully processed")
+ .build();
+
+ protected static Set<Relationship> relationships = new HashSet<>();
+
+ static {
+ descriptors.add(IOTDB_HOST);
+ descriptors.add(IOTDB_PORT);
+ descriptors.add(USERNAME);
+ descriptors.add(PASSWORD);
+
+ relationships.add(REL_SUCCESS);
+ relationships.add(REL_FAILURE);
+
+ typeMap.put(RecordFieldType.STRING, TSDataType.TEXT);
+ typeMap.put(RecordFieldType.BOOLEAN, TSDataType.BOOLEAN);
+ typeMap.put(RecordFieldType.INT, TSDataType.INT32);
+ typeMap.put(RecordFieldType.LONG, TSDataType.INT64);
+ typeMap.put(RecordFieldType.FLOAT, TSDataType.FLOAT);
+ typeMap.put(RecordFieldType.DOUBLE, TSDataType.DOUBLE);
+
+ supportedType.add(RecordFieldType.BOOLEAN);
+ supportedType.add(RecordFieldType.STRING);
+ supportedType.add(RecordFieldType.INT);
+ supportedType.add(RecordFieldType.LONG);
+ supportedType.add(RecordFieldType.FLOAT);
+ supportedType.add(RecordFieldType.DOUBLE);
+
+ supportedTimeType.add(RecordFieldType.STRING);
+ supportedTimeType.add(RecordFieldType.LONG);
+ }
+
+ protected final AtomicReference<Session> session = new AtomicReference<>(null);
+
+ @OnScheduled
+ public void onScheduled(ProcessContext context) {
+ connectToIoTDB(context);
+ }
+
+ void connectToIoTDB(ProcessContext context) {
+ if (session.get() == null) {
+ final String host, username, password;
+ final int port;
+
+ host = context.getProperty(IOTDB_HOST).getValue();
+ port = Integer.parseInt(context.getProperty(IOTDB_PORT).getValue());
+ username = context.getProperty(USERNAME).getValue();
+ password = context.getProperty(PASSWORD).getValue();
+
+ session.set(
+ new Session.Builder()
+ .host(host)
+ .port(port)
+ .username(username)
+ .password(password)
+ .build());
+ try {
+ session.get().open();
+ } catch (IoTDBConnectionException e) {
+ getLogger().error(e.getMessage());
+ e.printStackTrace();
+ }
+ }
+ }
+
+ public void stop(ProcessContext context) {
+ if (session.get() != null) {
+ try {
+ session.get().close();
+ } catch (IoTDBConnectionException e) {
+ e.printStackTrace();
Review Comment:
`printStackTrace` should be replaced an error log.
##########
nifi-nar-bundles/nifi-iotdb-bundle/nifi-iotdb-processors/src/main/java/org/apache/nifi/processors/AbstractIoTDB.java:
##########
@@ -0,0 +1,442 @@
+/*
+ * 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;
+
+import java.text.SimpleDateFormat;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicReference;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.nifi.processors.model.Field;
+import org.apache.nifi.processors.model.Schema;
+import org.apache.iotdb.rpc.IoTDBConnectionException;
+import org.apache.iotdb.session.Session;
+import org.apache.iotdb.tsfile.file.metadata.enums.CompressionType;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSEncoding;
+import org.apache.iotdb.tsfile.utils.Binary;
+import org.apache.iotdb.tsfile.write.record.Tablet;
+import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.serialization.record.DataType;
+import org.apache.nifi.serialization.record.RecordFieldType;
+import org.apache.nifi.serialization.record.RecordSchema;
+import org.apache.nifi.util.Tuple;
+
+public abstract class AbstractIoTDB extends AbstractProcessor {
+ private static final int DEFAULT_IOTDB_PORT = 6667;
+
+ private static Map<RecordFieldType, TSDataType> typeMap =
+ new HashMap<>();
+
+ static final Set<RecordFieldType> supportedType =
+ new HashSet<>();
+
+ static final Set<RecordFieldType> supportedTimeType =
+ new HashSet<>();
+
+ private static final String[] STRING_TIME_FORMAT =
+ new String[]{
+ "yyyy-MM-dd HH:mm:ss.SSSX",
+ "yyyy/MM/dd HH:mm:ss.SSSX",
+ "yyyy.MM.dd HH:mm:ss.SSSX",
+ "yyyy-MM-dd HH:mm:ssX",
+ "yyyy/MM/dd HH:mm:ssX",
+ "yyyy.MM.dd HH:mm:ssX",
+ "yyyy-MM-dd HH:mm:ss.SSSz",
+ "yyyy/MM/dd HH:mm:ss.SSSz",
+ "yyyy.MM.dd HH:mm:ss.SSSz",
+ "yyyy-MM-dd HH:mm:ssz",
+ "yyyy/MM/dd HH:mm:ssz",
+ "yyyy.MM.dd HH:mm:ssz",
+ "yyyy-MM-dd HH:mm:ss.SSS",
+ "yyyy/MM/dd HH:mm:ss.SSS",
+ "yyyy.MM.dd HH:mm:ss.SSS",
+ "yyyy-MM-dd HH:mm:ss",
+ "yyyy/MM/dd HH:mm:ss",
+ "yyyy.MM.dd HH:mm:ss",
+ "yyyy-MM-dd'T'HH:mm:ss.SSSX",
+ "yyyy/MM/dd'T'HH:mm:ss.SSSX",
+ "yyyy.MM.dd'T'HH:mm:ss.SSSX",
+ "yyyy-MM-dd'T'HH:mm:ssX",
+ "yyyy/MM/dd'T'HH:mm:ssX",
+ "yyyy.MM.dd'T'HH:mm:ssX",
+ "yyyy-MM-dd'T'HH:mm:ss.SSSz",
+ "yyyy/MM/dd'T'HH:mm:ss.SSSz",
+ "yyyy.MM.dd'T'HH:mm:ss.SSSz",
+ "yyyy-MM-dd'T'HH:mm:ssz",
+ "yyyy/MM/dd'T'HH:mm:ssz",
+ "yyyy.MM.dd'T'HH:mm:ssz",
+ "yyyy-MM-dd'T'HH:mm:ss.SSS",
+ "yyyy/MM/dd'T'HH:mm:ss.SSS",
+ "yyyy.MM.dd'T'HH:mm:ss.SSS",
+ "yyyy-MM-dd'T'HH:mm:ss",
+ "yyyy/MM/dd'T'HH:mm:ss",
+ "yyyy.MM.dd'T'HH:mm:ss"
+ };
+
+ static final PropertyDescriptor IOTDB_HOST =
+ new PropertyDescriptor.Builder()
+ .name("Host")
+ .description("The host of IoTDB.")
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .required(true)
+ .build();
+
+ static final PropertyDescriptor IOTDB_PORT =
+ new PropertyDescriptor.Builder()
+ .name("Port")
+ .description("The port of IoTDB.")
+ .defaultValue(String.valueOf(DEFAULT_IOTDB_PORT))
+ .addValidator(StandardValidators.PORT_VALIDATOR)
+ .required(true)
+ .build();
+
+ static final PropertyDescriptor USERNAME =
+ new PropertyDescriptor.Builder()
+ .name("Username")
+ .description("Username to access the IoTDB.")
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .required(true)
+ .build();
+
+ static final PropertyDescriptor PASSWORD =
+ new PropertyDescriptor.Builder()
+ .name("Password")
+ .description("Password to access the IoTDB.")
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .required(true)
+ .sensitive(true)
+ .build();
+
+ protected static List<PropertyDescriptor> descriptors = new ArrayList<>();
+
+ protected static final Relationship REL_SUCCESS =
+ new Relationship.Builder()
+ .name("success")
+ .description("files that were successfully processed")
+ .build();
+ protected static final Relationship REL_FAILURE =
+ new Relationship.Builder()
+ .name("failure")
+ .description("files that were not successfully processed")
+ .build();
+
+ protected static Set<Relationship> relationships = new HashSet<>();
+
+ static {
+ descriptors.add(IOTDB_HOST);
+ descriptors.add(IOTDB_PORT);
+ descriptors.add(USERNAME);
+ descriptors.add(PASSWORD);
+
+ relationships.add(REL_SUCCESS);
+ relationships.add(REL_FAILURE);
+
+ typeMap.put(RecordFieldType.STRING, TSDataType.TEXT);
+ typeMap.put(RecordFieldType.BOOLEAN, TSDataType.BOOLEAN);
+ typeMap.put(RecordFieldType.INT, TSDataType.INT32);
+ typeMap.put(RecordFieldType.LONG, TSDataType.INT64);
+ typeMap.put(RecordFieldType.FLOAT, TSDataType.FLOAT);
+ typeMap.put(RecordFieldType.DOUBLE, TSDataType.DOUBLE);
+
+ supportedType.add(RecordFieldType.BOOLEAN);
+ supportedType.add(RecordFieldType.STRING);
+ supportedType.add(RecordFieldType.INT);
+ supportedType.add(RecordFieldType.LONG);
+ supportedType.add(RecordFieldType.FLOAT);
+ supportedType.add(RecordFieldType.DOUBLE);
+
+ supportedTimeType.add(RecordFieldType.STRING);
+ supportedTimeType.add(RecordFieldType.LONG);
+ }
+
+ protected final AtomicReference<Session> session = new AtomicReference<>(null);
+
+ @OnScheduled
+ public void onScheduled(ProcessContext context) {
+ connectToIoTDB(context);
+ }
+
+ void connectToIoTDB(ProcessContext context) {
+ if (session.get() == null) {
+ final String host, username, password;
+ final int port;
+
+ host = context.getProperty(IOTDB_HOST).getValue();
+ port = Integer.parseInt(context.getProperty(IOTDB_PORT).getValue());
+ username = context.getProperty(USERNAME).getValue();
+ password = context.getProperty(PASSWORD).getValue();
Review Comment:
Variable declaration and assignment should be combined into single lines as follows:
```suggestion
final String host = context.getProperty(IOTDB_HOST).getValue();
final int port = Integer.parseInt(context.getProperty(IOTDB_PORT).getValue());
final String username = context.getProperty(USERNAME).getValue();
final String password = context.getProperty(PASSWORD).getValue();
```
##########
nifi-nar-bundles/nifi-iotdb-bundle/nifi-iotdb-processors/src/main/java/org/apache/nifi/processors/AbstractIoTDB.java:
##########
@@ -0,0 +1,442 @@
+/*
+ * 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;
+
+import java.text.SimpleDateFormat;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicReference;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.nifi.processors.model.Field;
+import org.apache.nifi.processors.model.Schema;
+import org.apache.iotdb.rpc.IoTDBConnectionException;
+import org.apache.iotdb.session.Session;
+import org.apache.iotdb.tsfile.file.metadata.enums.CompressionType;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSEncoding;
+import org.apache.iotdb.tsfile.utils.Binary;
+import org.apache.iotdb.tsfile.write.record.Tablet;
+import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.serialization.record.DataType;
+import org.apache.nifi.serialization.record.RecordFieldType;
+import org.apache.nifi.serialization.record.RecordSchema;
+import org.apache.nifi.util.Tuple;
+
+public abstract class AbstractIoTDB extends AbstractProcessor {
+ private static final int DEFAULT_IOTDB_PORT = 6667;
+
+ private static Map<RecordFieldType, TSDataType> typeMap =
+ new HashMap<>();
+
+ static final Set<RecordFieldType> supportedType =
+ new HashSet<>();
+
+ static final Set<RecordFieldType> supportedTimeType =
+ new HashSet<>();
+
+ private static final String[] STRING_TIME_FORMAT =
+ new String[]{
+ "yyyy-MM-dd HH:mm:ss.SSSX",
+ "yyyy/MM/dd HH:mm:ss.SSSX",
+ "yyyy.MM.dd HH:mm:ss.SSSX",
+ "yyyy-MM-dd HH:mm:ssX",
+ "yyyy/MM/dd HH:mm:ssX",
+ "yyyy.MM.dd HH:mm:ssX",
+ "yyyy-MM-dd HH:mm:ss.SSSz",
+ "yyyy/MM/dd HH:mm:ss.SSSz",
+ "yyyy.MM.dd HH:mm:ss.SSSz",
+ "yyyy-MM-dd HH:mm:ssz",
+ "yyyy/MM/dd HH:mm:ssz",
+ "yyyy.MM.dd HH:mm:ssz",
+ "yyyy-MM-dd HH:mm:ss.SSS",
+ "yyyy/MM/dd HH:mm:ss.SSS",
+ "yyyy.MM.dd HH:mm:ss.SSS",
+ "yyyy-MM-dd HH:mm:ss",
+ "yyyy/MM/dd HH:mm:ss",
+ "yyyy.MM.dd HH:mm:ss",
+ "yyyy-MM-dd'T'HH:mm:ss.SSSX",
+ "yyyy/MM/dd'T'HH:mm:ss.SSSX",
+ "yyyy.MM.dd'T'HH:mm:ss.SSSX",
+ "yyyy-MM-dd'T'HH:mm:ssX",
+ "yyyy/MM/dd'T'HH:mm:ssX",
+ "yyyy.MM.dd'T'HH:mm:ssX",
+ "yyyy-MM-dd'T'HH:mm:ss.SSSz",
+ "yyyy/MM/dd'T'HH:mm:ss.SSSz",
+ "yyyy.MM.dd'T'HH:mm:ss.SSSz",
+ "yyyy-MM-dd'T'HH:mm:ssz",
+ "yyyy/MM/dd'T'HH:mm:ssz",
+ "yyyy.MM.dd'T'HH:mm:ssz",
+ "yyyy-MM-dd'T'HH:mm:ss.SSS",
+ "yyyy/MM/dd'T'HH:mm:ss.SSS",
+ "yyyy.MM.dd'T'HH:mm:ss.SSS",
+ "yyyy-MM-dd'T'HH:mm:ss",
+ "yyyy/MM/dd'T'HH:mm:ss",
+ "yyyy.MM.dd'T'HH:mm:ss"
+ };
+
+ static final PropertyDescriptor IOTDB_HOST =
+ new PropertyDescriptor.Builder()
+ .name("Host")
+ .description("The host of IoTDB.")
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .required(true)
+ .build();
+
+ static final PropertyDescriptor IOTDB_PORT =
+ new PropertyDescriptor.Builder()
+ .name("Port")
+ .description("The port of IoTDB.")
+ .defaultValue(String.valueOf(DEFAULT_IOTDB_PORT))
+ .addValidator(StandardValidators.PORT_VALIDATOR)
+ .required(true)
+ .build();
+
+ static final PropertyDescriptor USERNAME =
+ new PropertyDescriptor.Builder()
+ .name("Username")
+ .description("Username to access the IoTDB.")
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .required(true)
+ .build();
+
+ static final PropertyDescriptor PASSWORD =
+ new PropertyDescriptor.Builder()
+ .name("Password")
+ .description("Password to access the IoTDB.")
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .required(true)
+ .sensitive(true)
+ .build();
+
+ protected static List<PropertyDescriptor> descriptors = new ArrayList<>();
+
+ protected static final Relationship REL_SUCCESS =
+ new Relationship.Builder()
+ .name("success")
+ .description("files that were successfully processed")
+ .build();
+ protected static final Relationship REL_FAILURE =
+ new Relationship.Builder()
+ .name("failure")
+ .description("files that were not successfully processed")
+ .build();
+
+ protected static Set<Relationship> relationships = new HashSet<>();
+
+ static {
+ descriptors.add(IOTDB_HOST);
+ descriptors.add(IOTDB_PORT);
+ descriptors.add(USERNAME);
+ descriptors.add(PASSWORD);
+
+ relationships.add(REL_SUCCESS);
+ relationships.add(REL_FAILURE);
+
+ typeMap.put(RecordFieldType.STRING, TSDataType.TEXT);
+ typeMap.put(RecordFieldType.BOOLEAN, TSDataType.BOOLEAN);
+ typeMap.put(RecordFieldType.INT, TSDataType.INT32);
+ typeMap.put(RecordFieldType.LONG, TSDataType.INT64);
+ typeMap.put(RecordFieldType.FLOAT, TSDataType.FLOAT);
+ typeMap.put(RecordFieldType.DOUBLE, TSDataType.DOUBLE);
+
+ supportedType.add(RecordFieldType.BOOLEAN);
+ supportedType.add(RecordFieldType.STRING);
+ supportedType.add(RecordFieldType.INT);
+ supportedType.add(RecordFieldType.LONG);
+ supportedType.add(RecordFieldType.FLOAT);
+ supportedType.add(RecordFieldType.DOUBLE);
+
+ supportedTimeType.add(RecordFieldType.STRING);
+ supportedTimeType.add(RecordFieldType.LONG);
+ }
+
+ protected final AtomicReference<Session> session = new AtomicReference<>(null);
+
+ @OnScheduled
+ public void onScheduled(ProcessContext context) {
+ connectToIoTDB(context);
+ }
+
+ void connectToIoTDB(ProcessContext context) {
+ if (session.get() == null) {
+ final String host, username, password;
+ final int port;
+
+ host = context.getProperty(IOTDB_HOST).getValue();
+ port = Integer.parseInt(context.getProperty(IOTDB_PORT).getValue());
+ username = context.getProperty(USERNAME).getValue();
+ password = context.getProperty(PASSWORD).getValue();
+
+ session.set(
+ new Session.Builder()
+ .host(host)
+ .port(port)
+ .username(username)
+ .password(password)
+ .build());
+ try {
+ session.get().open();
+ } catch (IoTDBConnectionException e) {
+ getLogger().error(e.getMessage());
+ e.printStackTrace();
+ }
+ }
+ }
+
+ public void stop(ProcessContext context) {
+ if (session.get() != null) {
+ try {
+ session.get().close();
+ } catch (IoTDBConnectionException e) {
+ e.printStackTrace();
+ }
+ session.set(null);
+ }
+ }
+
+ protected TSDataType getType(RecordFieldType type) {
+ return typeMap.get(type);
+ }
+
+ protected Tuple<Boolean, String> validateSchemaAttribute(String schemaAttribute) {
+ ObjectMapper mapper = new ObjectMapper();
+ JsonNode schema = null;
+ try {
+ schema = mapper.readTree(schemaAttribute);
+ } catch (JsonProcessingException e) {
+ return new Tuple<>(false, e.getMessage());
+ }
+ Set<String> keySet = new HashSet<>();
+ schema.fieldNames().forEachRemaining(field -> keySet.add(field));
+
+ if (!keySet.contains("timeType") || !keySet.contains("fields")) {
Review Comment:
It would be helpful to declare `timeType` and `fields` as static variables so that they can be reused in multiple locations.
##########
nifi-nar-bundles/nifi-iotdb-bundle/nifi-iotdb-processors/src/main/java/org/apache/nifi/processors/AbstractIoTDB.java:
##########
@@ -0,0 +1,442 @@
+/*
+ * 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;
+
+import java.text.SimpleDateFormat;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicReference;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.nifi.processors.model.Field;
+import org.apache.nifi.processors.model.Schema;
+import org.apache.iotdb.rpc.IoTDBConnectionException;
+import org.apache.iotdb.session.Session;
+import org.apache.iotdb.tsfile.file.metadata.enums.CompressionType;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSEncoding;
+import org.apache.iotdb.tsfile.utils.Binary;
+import org.apache.iotdb.tsfile.write.record.Tablet;
+import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.serialization.record.DataType;
+import org.apache.nifi.serialization.record.RecordFieldType;
+import org.apache.nifi.serialization.record.RecordSchema;
+import org.apache.nifi.util.Tuple;
+
+public abstract class AbstractIoTDB extends AbstractProcessor {
+ private static final int DEFAULT_IOTDB_PORT = 6667;
+
+ private static Map<RecordFieldType, TSDataType> typeMap =
+ new HashMap<>();
+
+ static final Set<RecordFieldType> supportedType =
+ new HashSet<>();
+
+ static final Set<RecordFieldType> supportedTimeType =
+ new HashSet<>();
+
+ private static final String[] STRING_TIME_FORMAT =
+ new String[]{
+ "yyyy-MM-dd HH:mm:ss.SSSX",
+ "yyyy/MM/dd HH:mm:ss.SSSX",
+ "yyyy.MM.dd HH:mm:ss.SSSX",
+ "yyyy-MM-dd HH:mm:ssX",
+ "yyyy/MM/dd HH:mm:ssX",
+ "yyyy.MM.dd HH:mm:ssX",
+ "yyyy-MM-dd HH:mm:ss.SSSz",
+ "yyyy/MM/dd HH:mm:ss.SSSz",
+ "yyyy.MM.dd HH:mm:ss.SSSz",
+ "yyyy-MM-dd HH:mm:ssz",
+ "yyyy/MM/dd HH:mm:ssz",
+ "yyyy.MM.dd HH:mm:ssz",
+ "yyyy-MM-dd HH:mm:ss.SSS",
+ "yyyy/MM/dd HH:mm:ss.SSS",
+ "yyyy.MM.dd HH:mm:ss.SSS",
+ "yyyy-MM-dd HH:mm:ss",
+ "yyyy/MM/dd HH:mm:ss",
+ "yyyy.MM.dd HH:mm:ss",
+ "yyyy-MM-dd'T'HH:mm:ss.SSSX",
+ "yyyy/MM/dd'T'HH:mm:ss.SSSX",
+ "yyyy.MM.dd'T'HH:mm:ss.SSSX",
+ "yyyy-MM-dd'T'HH:mm:ssX",
+ "yyyy/MM/dd'T'HH:mm:ssX",
+ "yyyy.MM.dd'T'HH:mm:ssX",
+ "yyyy-MM-dd'T'HH:mm:ss.SSSz",
+ "yyyy/MM/dd'T'HH:mm:ss.SSSz",
+ "yyyy.MM.dd'T'HH:mm:ss.SSSz",
+ "yyyy-MM-dd'T'HH:mm:ssz",
+ "yyyy/MM/dd'T'HH:mm:ssz",
+ "yyyy.MM.dd'T'HH:mm:ssz",
+ "yyyy-MM-dd'T'HH:mm:ss.SSS",
+ "yyyy/MM/dd'T'HH:mm:ss.SSS",
+ "yyyy.MM.dd'T'HH:mm:ss.SSS",
+ "yyyy-MM-dd'T'HH:mm:ss",
+ "yyyy/MM/dd'T'HH:mm:ss",
+ "yyyy.MM.dd'T'HH:mm:ss"
+ };
+
+ static final PropertyDescriptor IOTDB_HOST =
+ new PropertyDescriptor.Builder()
+ .name("Host")
+ .description("The host of IoTDB.")
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .required(true)
+ .build();
+
+ static final PropertyDescriptor IOTDB_PORT =
+ new PropertyDescriptor.Builder()
+ .name("Port")
+ .description("The port of IoTDB.")
+ .defaultValue(String.valueOf(DEFAULT_IOTDB_PORT))
+ .addValidator(StandardValidators.PORT_VALIDATOR)
+ .required(true)
+ .build();
+
+ static final PropertyDescriptor USERNAME =
+ new PropertyDescriptor.Builder()
+ .name("Username")
+ .description("Username to access the IoTDB.")
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .required(true)
+ .build();
+
+ static final PropertyDescriptor PASSWORD =
+ new PropertyDescriptor.Builder()
+ .name("Password")
+ .description("Password to access the IoTDB.")
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .required(true)
+ .sensitive(true)
+ .build();
+
+ protected static List<PropertyDescriptor> descriptors = new ArrayList<>();
+
+ protected static final Relationship REL_SUCCESS =
+ new Relationship.Builder()
+ .name("success")
+ .description("files that were successfully processed")
+ .build();
+ protected static final Relationship REL_FAILURE =
+ new Relationship.Builder()
+ .name("failure")
+ .description("files that were not successfully processed")
+ .build();
+
+ protected static Set<Relationship> relationships = new HashSet<>();
+
+ static {
+ descriptors.add(IOTDB_HOST);
+ descriptors.add(IOTDB_PORT);
+ descriptors.add(USERNAME);
+ descriptors.add(PASSWORD);
+
+ relationships.add(REL_SUCCESS);
+ relationships.add(REL_FAILURE);
+
+ typeMap.put(RecordFieldType.STRING, TSDataType.TEXT);
+ typeMap.put(RecordFieldType.BOOLEAN, TSDataType.BOOLEAN);
+ typeMap.put(RecordFieldType.INT, TSDataType.INT32);
+ typeMap.put(RecordFieldType.LONG, TSDataType.INT64);
+ typeMap.put(RecordFieldType.FLOAT, TSDataType.FLOAT);
+ typeMap.put(RecordFieldType.DOUBLE, TSDataType.DOUBLE);
+
+ supportedType.add(RecordFieldType.BOOLEAN);
+ supportedType.add(RecordFieldType.STRING);
+ supportedType.add(RecordFieldType.INT);
+ supportedType.add(RecordFieldType.LONG);
+ supportedType.add(RecordFieldType.FLOAT);
+ supportedType.add(RecordFieldType.DOUBLE);
+
+ supportedTimeType.add(RecordFieldType.STRING);
+ supportedTimeType.add(RecordFieldType.LONG);
+ }
+
+ protected final AtomicReference<Session> session = new AtomicReference<>(null);
+
+ @OnScheduled
+ public void onScheduled(ProcessContext context) {
+ connectToIoTDB(context);
+ }
+
+ void connectToIoTDB(ProcessContext context) {
+ if (session.get() == null) {
+ final String host, username, password;
+ final int port;
+
+ host = context.getProperty(IOTDB_HOST).getValue();
+ port = Integer.parseInt(context.getProperty(IOTDB_PORT).getValue());
+ username = context.getProperty(USERNAME).getValue();
+ password = context.getProperty(PASSWORD).getValue();
+
+ session.set(
+ new Session.Builder()
+ .host(host)
+ .port(port)
+ .username(username)
+ .password(password)
+ .build());
+ try {
+ session.get().open();
+ } catch (IoTDBConnectionException e) {
+ getLogger().error(e.getMessage());
+ e.printStackTrace();
+ }
+ }
+ }
+
+ public void stop(ProcessContext context) {
+ if (session.get() != null) {
+ try {
+ session.get().close();
+ } catch (IoTDBConnectionException e) {
+ e.printStackTrace();
+ }
+ session.set(null);
+ }
+ }
+
+ protected TSDataType getType(RecordFieldType type) {
+ return typeMap.get(type);
+ }
+
+ protected Tuple<Boolean, String> validateSchemaAttribute(String schemaAttribute) {
+ ObjectMapper mapper = new ObjectMapper();
Review Comment:
An instance of `ObjectMapper` is thread-safe, so this can be declared as a static member and accessed.
##########
nifi-nar-bundles/nifi-iotdb-bundle/nifi-iotdb-processors/src/main/java/org/apache/nifi/processors/AbstractIoTDB.java:
##########
@@ -0,0 +1,442 @@
+/*
+ * 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;
+
+import java.text.SimpleDateFormat;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicReference;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.nifi.processors.model.Field;
+import org.apache.nifi.processors.model.Schema;
+import org.apache.iotdb.rpc.IoTDBConnectionException;
+import org.apache.iotdb.session.Session;
+import org.apache.iotdb.tsfile.file.metadata.enums.CompressionType;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSEncoding;
+import org.apache.iotdb.tsfile.utils.Binary;
+import org.apache.iotdb.tsfile.write.record.Tablet;
+import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.serialization.record.DataType;
+import org.apache.nifi.serialization.record.RecordFieldType;
+import org.apache.nifi.serialization.record.RecordSchema;
+import org.apache.nifi.util.Tuple;
+
+public abstract class AbstractIoTDB extends AbstractProcessor {
+ private static final int DEFAULT_IOTDB_PORT = 6667;
+
+ private static Map<RecordFieldType, TSDataType> typeMap =
+ new HashMap<>();
+
+ static final Set<RecordFieldType> supportedType =
+ new HashSet<>();
+
+ static final Set<RecordFieldType> supportedTimeType =
+ new HashSet<>();
+
+ private static final String[] STRING_TIME_FORMAT =
+ new String[]{
+ "yyyy-MM-dd HH:mm:ss.SSSX",
+ "yyyy/MM/dd HH:mm:ss.SSSX",
+ "yyyy.MM.dd HH:mm:ss.SSSX",
+ "yyyy-MM-dd HH:mm:ssX",
+ "yyyy/MM/dd HH:mm:ssX",
+ "yyyy.MM.dd HH:mm:ssX",
+ "yyyy-MM-dd HH:mm:ss.SSSz",
+ "yyyy/MM/dd HH:mm:ss.SSSz",
+ "yyyy.MM.dd HH:mm:ss.SSSz",
+ "yyyy-MM-dd HH:mm:ssz",
+ "yyyy/MM/dd HH:mm:ssz",
+ "yyyy.MM.dd HH:mm:ssz",
+ "yyyy-MM-dd HH:mm:ss.SSS",
+ "yyyy/MM/dd HH:mm:ss.SSS",
+ "yyyy.MM.dd HH:mm:ss.SSS",
+ "yyyy-MM-dd HH:mm:ss",
+ "yyyy/MM/dd HH:mm:ss",
+ "yyyy.MM.dd HH:mm:ss",
+ "yyyy-MM-dd'T'HH:mm:ss.SSSX",
+ "yyyy/MM/dd'T'HH:mm:ss.SSSX",
+ "yyyy.MM.dd'T'HH:mm:ss.SSSX",
+ "yyyy-MM-dd'T'HH:mm:ssX",
+ "yyyy/MM/dd'T'HH:mm:ssX",
+ "yyyy.MM.dd'T'HH:mm:ssX",
+ "yyyy-MM-dd'T'HH:mm:ss.SSSz",
+ "yyyy/MM/dd'T'HH:mm:ss.SSSz",
+ "yyyy.MM.dd'T'HH:mm:ss.SSSz",
+ "yyyy-MM-dd'T'HH:mm:ssz",
+ "yyyy/MM/dd'T'HH:mm:ssz",
+ "yyyy.MM.dd'T'HH:mm:ssz",
+ "yyyy-MM-dd'T'HH:mm:ss.SSS",
+ "yyyy/MM/dd'T'HH:mm:ss.SSS",
+ "yyyy.MM.dd'T'HH:mm:ss.SSS",
+ "yyyy-MM-dd'T'HH:mm:ss",
+ "yyyy/MM/dd'T'HH:mm:ss",
+ "yyyy.MM.dd'T'HH:mm:ss"
+ };
+
+ static final PropertyDescriptor IOTDB_HOST =
+ new PropertyDescriptor.Builder()
+ .name("Host")
+ .description("The host of IoTDB.")
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .required(true)
+ .build();
+
+ static final PropertyDescriptor IOTDB_PORT =
+ new PropertyDescriptor.Builder()
+ .name("Port")
+ .description("The port of IoTDB.")
+ .defaultValue(String.valueOf(DEFAULT_IOTDB_PORT))
+ .addValidator(StandardValidators.PORT_VALIDATOR)
+ .required(true)
+ .build();
+
+ static final PropertyDescriptor USERNAME =
+ new PropertyDescriptor.Builder()
+ .name("Username")
+ .description("Username to access the IoTDB.")
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .required(true)
+ .build();
+
+ static final PropertyDescriptor PASSWORD =
+ new PropertyDescriptor.Builder()
+ .name("Password")
+ .description("Password to access the IoTDB.")
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .required(true)
+ .sensitive(true)
+ .build();
+
+ protected static List<PropertyDescriptor> descriptors = new ArrayList<>();
+
+ protected static final Relationship REL_SUCCESS =
+ new Relationship.Builder()
+ .name("success")
+ .description("files that were successfully processed")
+ .build();
+ protected static final Relationship REL_FAILURE =
+ new Relationship.Builder()
+ .name("failure")
+ .description("files that were not successfully processed")
+ .build();
+
+ protected static Set<Relationship> relationships = new HashSet<>();
+
+ static {
+ descriptors.add(IOTDB_HOST);
+ descriptors.add(IOTDB_PORT);
+ descriptors.add(USERNAME);
+ descriptors.add(PASSWORD);
+
+ relationships.add(REL_SUCCESS);
+ relationships.add(REL_FAILURE);
+
+ typeMap.put(RecordFieldType.STRING, TSDataType.TEXT);
+ typeMap.put(RecordFieldType.BOOLEAN, TSDataType.BOOLEAN);
+ typeMap.put(RecordFieldType.INT, TSDataType.INT32);
+ typeMap.put(RecordFieldType.LONG, TSDataType.INT64);
+ typeMap.put(RecordFieldType.FLOAT, TSDataType.FLOAT);
+ typeMap.put(RecordFieldType.DOUBLE, TSDataType.DOUBLE);
+
+ supportedType.add(RecordFieldType.BOOLEAN);
+ supportedType.add(RecordFieldType.STRING);
+ supportedType.add(RecordFieldType.INT);
+ supportedType.add(RecordFieldType.LONG);
+ supportedType.add(RecordFieldType.FLOAT);
+ supportedType.add(RecordFieldType.DOUBLE);
+
+ supportedTimeType.add(RecordFieldType.STRING);
+ supportedTimeType.add(RecordFieldType.LONG);
+ }
+
+ protected final AtomicReference<Session> session = new AtomicReference<>(null);
+
+ @OnScheduled
+ public void onScheduled(ProcessContext context) {
+ connectToIoTDB(context);
+ }
+
+ void connectToIoTDB(ProcessContext context) {
+ if (session.get() == null) {
+ final String host, username, password;
+ final int port;
+
+ host = context.getProperty(IOTDB_HOST).getValue();
+ port = Integer.parseInt(context.getProperty(IOTDB_PORT).getValue());
+ username = context.getProperty(USERNAME).getValue();
+ password = context.getProperty(PASSWORD).getValue();
+
+ session.set(
+ new Session.Builder()
+ .host(host)
+ .port(port)
+ .username(username)
+ .password(password)
+ .build());
+ try {
+ session.get().open();
+ } catch (IoTDBConnectionException e) {
+ getLogger().error(e.getMessage());
+ e.printStackTrace();
+ }
+ }
+ }
+
+ public void stop(ProcessContext context) {
+ if (session.get() != null) {
+ try {
+ session.get().close();
+ } catch (IoTDBConnectionException e) {
+ e.printStackTrace();
+ }
+ session.set(null);
+ }
+ }
+
+ protected TSDataType getType(RecordFieldType type) {
+ return typeMap.get(type);
+ }
+
+ protected Tuple<Boolean, String> validateSchemaAttribute(String schemaAttribute) {
Review Comment:
Although the `Tuple` is a useful utility class, it is not very useful for conveying other details. It would be better to create a custom object and reuse that instead of using `Tuple`.
##########
nifi-nar-bundles/nifi-iotdb-bundle/nifi-iotdb-processors/src/main/java/org/apache/nifi/processors/PutIoTDB.java:
##########
@@ -0,0 +1,286 @@
+/*
+ * 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;
+
+import java.io.InputStream;
+import java.text.SimpleDateFormat;
+import java.util.Set;
+import java.util.List;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.nifi.processors.model.Schema;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.write.record.Tablet;
+import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnUnscheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.expression.ExpressionLanguageScope;
+import org.apache.nifi.flowfile.FlowFile;
+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.serialization.RecordReader;
+import org.apache.nifi.serialization.RecordReaderFactory;
+import org.apache.nifi.serialization.record.Record;
+import org.apache.nifi.util.Tuple;
+
+@Tags({"iotdb", "insert", "tablet"})
+@InputRequirement(InputRequirement.Requirement.INPUT_REQUIRED)
+@CapabilityDescription(
+ "This is a record aware processor that reads the content of the incoming FlowFile as individual records using the "
+ + "configured 'Record Reader' and writes them to Apache IoTDB using native interface.")
+public class PutIoTDB extends AbstractIoTDB {
+
+ static final PropertyDescriptor RECORD_READER_FACTORY =
+ new PropertyDescriptor.Builder()
+ .name("Record Reader")
+ .description(
+ "Specifies the type of Record Reader controller service to use for parsing the incoming data "
+ + "and determining the schema")
+ .identifiesControllerService(RecordReaderFactory.class)
+ .required(true)
+ .build();
+
+ static final PropertyDescriptor TIME_FIELD =
+ new PropertyDescriptor.Builder()
+ .name("Time Field")
+ .description(
+ "The field name which represents time. It can be updated by expression language.")
Review Comment:
The second sentence can be removed since support for FlowFile attributes and expression language will be indicated automatically in generated documentation.
```suggestion
"The field name which represents time")
```
##########
nifi-nar-bundles/nifi-iotdb-bundle/nifi-iotdb-processors/src/main/java/org/apache/nifi/processors/PutIoTDB.java:
##########
@@ -0,0 +1,286 @@
+/*
+ * 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;
+
+import java.io.InputStream;
+import java.text.SimpleDateFormat;
+import java.util.Set;
+import java.util.List;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.nifi.processors.model.Schema;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.write.record.Tablet;
+import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnUnscheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.expression.ExpressionLanguageScope;
+import org.apache.nifi.flowfile.FlowFile;
+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.serialization.RecordReader;
+import org.apache.nifi.serialization.RecordReaderFactory;
+import org.apache.nifi.serialization.record.Record;
+import org.apache.nifi.util.Tuple;
+
+@Tags({"iotdb", "insert", "tablet"})
+@InputRequirement(InputRequirement.Requirement.INPUT_REQUIRED)
+@CapabilityDescription(
+ "This is a record aware processor that reads the content of the incoming FlowFile as individual records using the "
+ + "configured 'Record Reader' and writes them to Apache IoTDB using native interface.")
+public class PutIoTDB extends AbstractIoTDB {
+
+ static final PropertyDescriptor RECORD_READER_FACTORY =
+ new PropertyDescriptor.Builder()
+ .name("Record Reader")
+ .description(
+ "Specifies the type of Record Reader controller service to use for parsing the incoming data "
+ + "and determining the schema")
+ .identifiesControllerService(RecordReaderFactory.class)
+ .required(true)
+ .build();
+
+ static final PropertyDescriptor TIME_FIELD =
+ new PropertyDescriptor.Builder()
+ .name("Time Field")
+ .description(
+ "The field name which represents time. It can be updated by expression language.")
+ .defaultValue("Time")
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .required(false)
+ .build();
+
+ static final PropertyDescriptor SCHEMA =
+ new PropertyDescriptor.Builder()
+ .name("Schema")
+ .description(
+ "The schema that IoTDB needs doesn't support good by NiFi.\n"
+ + "Therefore, you can define the schema here.\n"
+ + "Besides, you can set encoding type and compression type by this method.\n"
+ + "If you don't set this property, the inferred schema will be used.\n"
+ + "It can be updated by expression language.")
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .required(false)
+ .build();
+
+ static final PropertyDescriptor ALIGNED =
+ new PropertyDescriptor.Builder()
+ .name("Aligned")
+ .description("Whether using aligned interface? It can be updated by expression language.")
+ .allowableValues(new String[] {"true", "false"})
Review Comment:
```suggestion
.allowableValues("true", "false")
```
##########
nifi-nar-bundles/nifi-iotdb-bundle/nifi-iotdb-processors/src/main/java/org/apache/nifi/processors/AbstractIoTDB.java:
##########
@@ -0,0 +1,442 @@
+/*
+ * 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;
+
+import java.text.SimpleDateFormat;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicReference;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.nifi.processors.model.Field;
+import org.apache.nifi.processors.model.Schema;
+import org.apache.iotdb.rpc.IoTDBConnectionException;
+import org.apache.iotdb.session.Session;
+import org.apache.iotdb.tsfile.file.metadata.enums.CompressionType;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSEncoding;
+import org.apache.iotdb.tsfile.utils.Binary;
+import org.apache.iotdb.tsfile.write.record.Tablet;
+import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.serialization.record.DataType;
+import org.apache.nifi.serialization.record.RecordFieldType;
+import org.apache.nifi.serialization.record.RecordSchema;
+import org.apache.nifi.util.Tuple;
+
+public abstract class AbstractIoTDB extends AbstractProcessor {
+ private static final int DEFAULT_IOTDB_PORT = 6667;
+
+ private static Map<RecordFieldType, TSDataType> typeMap =
+ new HashMap<>();
+
+ static final Set<RecordFieldType> supportedType =
+ new HashSet<>();
+
+ static final Set<RecordFieldType> supportedTimeType =
+ new HashSet<>();
+
+ private static final String[] STRING_TIME_FORMAT =
+ new String[]{
+ "yyyy-MM-dd HH:mm:ss.SSSX",
+ "yyyy/MM/dd HH:mm:ss.SSSX",
+ "yyyy.MM.dd HH:mm:ss.SSSX",
+ "yyyy-MM-dd HH:mm:ssX",
+ "yyyy/MM/dd HH:mm:ssX",
+ "yyyy.MM.dd HH:mm:ssX",
+ "yyyy-MM-dd HH:mm:ss.SSSz",
+ "yyyy/MM/dd HH:mm:ss.SSSz",
+ "yyyy.MM.dd HH:mm:ss.SSSz",
+ "yyyy-MM-dd HH:mm:ssz",
+ "yyyy/MM/dd HH:mm:ssz",
+ "yyyy.MM.dd HH:mm:ssz",
+ "yyyy-MM-dd HH:mm:ss.SSS",
+ "yyyy/MM/dd HH:mm:ss.SSS",
+ "yyyy.MM.dd HH:mm:ss.SSS",
+ "yyyy-MM-dd HH:mm:ss",
+ "yyyy/MM/dd HH:mm:ss",
+ "yyyy.MM.dd HH:mm:ss",
+ "yyyy-MM-dd'T'HH:mm:ss.SSSX",
+ "yyyy/MM/dd'T'HH:mm:ss.SSSX",
+ "yyyy.MM.dd'T'HH:mm:ss.SSSX",
+ "yyyy-MM-dd'T'HH:mm:ssX",
+ "yyyy/MM/dd'T'HH:mm:ssX",
+ "yyyy.MM.dd'T'HH:mm:ssX",
+ "yyyy-MM-dd'T'HH:mm:ss.SSSz",
+ "yyyy/MM/dd'T'HH:mm:ss.SSSz",
+ "yyyy.MM.dd'T'HH:mm:ss.SSSz",
+ "yyyy-MM-dd'T'HH:mm:ssz",
+ "yyyy/MM/dd'T'HH:mm:ssz",
+ "yyyy.MM.dd'T'HH:mm:ssz",
+ "yyyy-MM-dd'T'HH:mm:ss.SSS",
+ "yyyy/MM/dd'T'HH:mm:ss.SSS",
+ "yyyy.MM.dd'T'HH:mm:ss.SSS",
+ "yyyy-MM-dd'T'HH:mm:ss",
+ "yyyy/MM/dd'T'HH:mm:ss",
+ "yyyy.MM.dd'T'HH:mm:ss"
+ };
+
+ static final PropertyDescriptor IOTDB_HOST =
+ new PropertyDescriptor.Builder()
+ .name("Host")
+ .description("The host of IoTDB.")
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .required(true)
+ .build();
+
+ static final PropertyDescriptor IOTDB_PORT =
+ new PropertyDescriptor.Builder()
+ .name("Port")
+ .description("The port of IoTDB.")
+ .defaultValue(String.valueOf(DEFAULT_IOTDB_PORT))
+ .addValidator(StandardValidators.PORT_VALIDATOR)
+ .required(true)
+ .build();
+
+ static final PropertyDescriptor USERNAME =
+ new PropertyDescriptor.Builder()
+ .name("Username")
+ .description("Username to access the IoTDB.")
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .required(true)
+ .build();
+
+ static final PropertyDescriptor PASSWORD =
+ new PropertyDescriptor.Builder()
+ .name("Password")
+ .description("Password to access the IoTDB.")
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .required(true)
+ .sensitive(true)
+ .build();
+
+ protected static List<PropertyDescriptor> descriptors = new ArrayList<>();
+
+ protected static final Relationship REL_SUCCESS =
+ new Relationship.Builder()
+ .name("success")
+ .description("files that were successfully processed")
+ .build();
+ protected static final Relationship REL_FAILURE =
+ new Relationship.Builder()
+ .name("failure")
+ .description("files that were not successfully processed")
+ .build();
+
+ protected static Set<Relationship> relationships = new HashSet<>();
+
+ static {
+ descriptors.add(IOTDB_HOST);
+ descriptors.add(IOTDB_PORT);
+ descriptors.add(USERNAME);
+ descriptors.add(PASSWORD);
+
+ relationships.add(REL_SUCCESS);
+ relationships.add(REL_FAILURE);
+
+ typeMap.put(RecordFieldType.STRING, TSDataType.TEXT);
+ typeMap.put(RecordFieldType.BOOLEAN, TSDataType.BOOLEAN);
+ typeMap.put(RecordFieldType.INT, TSDataType.INT32);
+ typeMap.put(RecordFieldType.LONG, TSDataType.INT64);
+ typeMap.put(RecordFieldType.FLOAT, TSDataType.FLOAT);
+ typeMap.put(RecordFieldType.DOUBLE, TSDataType.DOUBLE);
+
+ supportedType.add(RecordFieldType.BOOLEAN);
+ supportedType.add(RecordFieldType.STRING);
+ supportedType.add(RecordFieldType.INT);
+ supportedType.add(RecordFieldType.LONG);
+ supportedType.add(RecordFieldType.FLOAT);
+ supportedType.add(RecordFieldType.DOUBLE);
+
+ supportedTimeType.add(RecordFieldType.STRING);
+ supportedTimeType.add(RecordFieldType.LONG);
+ }
+
+ protected final AtomicReference<Session> session = new AtomicReference<>(null);
+
+ @OnScheduled
+ public void onScheduled(ProcessContext context) {
+ connectToIoTDB(context);
+ }
+
+ void connectToIoTDB(ProcessContext context) {
+ if (session.get() == null) {
+ final String host, username, password;
+ final int port;
+
+ host = context.getProperty(IOTDB_HOST).getValue();
+ port = Integer.parseInt(context.getProperty(IOTDB_PORT).getValue());
+ username = context.getProperty(USERNAME).getValue();
+ password = context.getProperty(PASSWORD).getValue();
+
+ session.set(
+ new Session.Builder()
+ .host(host)
+ .port(port)
+ .username(username)
+ .password(password)
+ .build());
+ try {
+ session.get().open();
+ } catch (IoTDBConnectionException e) {
+ getLogger().error(e.getMessage());
+ e.printStackTrace();
+ }
+ }
+ }
+
+ public void stop(ProcessContext context) {
+ if (session.get() != null) {
+ try {
+ session.get().close();
+ } catch (IoTDBConnectionException e) {
+ e.printStackTrace();
+ }
+ session.set(null);
+ }
+ }
+
+ protected TSDataType getType(RecordFieldType type) {
+ return typeMap.get(type);
+ }
+
+ protected Tuple<Boolean, String> validateSchemaAttribute(String schemaAttribute) {
+ ObjectMapper mapper = new ObjectMapper();
+ JsonNode schema = null;
+ try {
+ schema = mapper.readTree(schemaAttribute);
+ } catch (JsonProcessingException e) {
+ return new Tuple<>(false, e.getMessage());
+ }
+ Set<String> keySet = new HashSet<>();
+ schema.fieldNames().forEachRemaining(field -> keySet.add(field));
+
+ if (!keySet.contains("timeType") || !keySet.contains("fields")) {
+ String msg = "The JSON of schema must contain `timeType` and `fields`.";
+ return new Tuple<>(false, msg);
+ }
+
+ if (!Schema.getSupportedTimeType().contains(schema.get("timeType").asText().toLowerCase())) {
+ String msg =
+ String.format(
+ "Unknown `timeType`: %s, there are only two options `long` and `string` for this property.",
+ schema.get("timeType").asText());
+ return new Tuple<>(false, msg);
+ }
+
+ for (int i = 0; i < schema.get("fields").size(); i++) {
+ JsonNode field = schema.get("fields").get(i);
+ Set<String> fieldKeySet = new HashSet<>();
+
+ field.fieldNames().forEachRemaining(fieldName -> fieldKeySet.add(fieldName));
+ if (!fieldKeySet.contains("tsName") || !fieldKeySet.contains("dataType")) {
+ String msg = "`tsName` or `dataType` has not been set.";
+ return new Tuple<>(false, msg);
+ }
+
+ if (!field.get("tsName").asText().startsWith("root.")) {
+ String msg =
+ String.format("The tsName `%s` is not start with 'root.'.", field.get("tsName").asText());
+ return new Tuple<>(false, msg);
+ }
+
+ if (!Field.getSupportedDataType().contains(field.get("dataType").asText())) {
+ String msg =
+ String.format(
+ "Unknown `dataType`: %s. The supported dataTypes are %s",
+ field.get("dataType").asText(), Field.getSupportedDataType());
+ return new Tuple<>(false, msg);
+ }
+
+ Set<String> supportedKeySet = new HashSet<>();
+ supportedKeySet.add("tsName");
+ supportedKeySet.add("dataType");
+ supportedKeySet.add("encoding");
+ supportedKeySet.add("compressionType");
+
+ HashSet<String> tmpKetSet = new HashSet<>();
+ tmpKetSet.addAll(supportedKeySet);
+ tmpKetSet.addAll(fieldKeySet);
+ tmpKetSet.removeAll(supportedKeySet);
+ if (!tmpKetSet.isEmpty()) {
+ String msg = "Unknown property or properties: " + tmpKetSet;
+ return new Tuple<>(false, msg);
+ }
+
+ if (fieldKeySet.contains("compressionType") && !fieldKeySet.contains("encoding")) {
+ String msg =
+ "The `compressionType` has been set, but the `encoding` has not. The property `compressionType` will not take effect.";
+ return new Tuple<>(true, msg);
+ }
+
+ if (field.get("encoding") != null
+ && !Field.getSupportedEncoding().contains(field.get("encoding").asText())) {
+ String msg =
+ String.format(
+ "Unknown `encoding`: %s, The supported encoding types are %s",
+ field.get("encoding").asText(), Field.getSupportedEncoding());
+ return new Tuple<>(false, msg);
+ }
+
+ if (field.get("compressionType") != null
+ && !Field.getSupportedCompressionType().contains(field.get("compressionType"))) {
+ String msg =
+ String.format(
+ "Unknown `compressionType`: %s, The supported compressionType are %s",
+ field.get("compressionType").asText(), Field.getSupportedCompressionType());
+ return new Tuple<>(false, msg);
+ }
+ }
+
+ return new Tuple<>(true, null);
+ }
+
+ protected Tuple<Boolean, String> validateSchema(RecordSchema recordSchema) {
+ List<String> fieldNames = recordSchema.getFieldNames();
+ List<DataType> dataTypes = recordSchema.getDataTypes();
+ if (!fieldNames.contains("Time")) {
+ return new Tuple<>(false, "The fields must contain `Time`.");
+ }
+ if (!supportedTimeType.contains(recordSchema.getDataType("Time").get().getFieldType())) {
+ return new Tuple<>(false, "The dataType of `Time` must be `STRING` or `LONG`.");
+ }
+ fieldNames.remove("Time");
+ for (String fieldName : fieldNames) {
+ if (!fieldName.startsWith("root.")) {
+ String msg = String.format("The tsName `%s` is not start with 'root.'.", fieldName);
+ return new Tuple<>(false, msg);
+ }
+ }
+ for (DataType type : dataTypes) {
+ RecordFieldType dataType = type.getFieldType();
+ if (!supportedType.contains(dataType)) {
+ String msg =
+ String.format(
+ "Unknown `dataType`: %s. The supported dataTypes are %s",
+ dataType.toString(), supportedType);
+ return new Tuple<>(false, msg);
+ }
+ }
+
+ return new Tuple<>(true, null);
+ }
+
+ protected Map<String, List<String>> parseSchema(List<String> filedNames) {
+ HashMap<String, List<String>> deviceMeasurementMap = new HashMap<>();
+ filedNames.stream()
+ .forEach(
+ filed -> {
+ String[] paths = filed.split("\\.");
+ String device = StringUtils.join(paths, ".", 0, paths.length - 1);
+
+ if (!deviceMeasurementMap.containsKey(device)) {
+ deviceMeasurementMap.put(device, new ArrayList<>());
+ }
+ deviceMeasurementMap.get(device).add(paths[paths.length - 1]);
+ });
+
+ return deviceMeasurementMap;
+ }
+
+ protected HashMap<String, Tablet> generateTablets(Schema schema, int maxRowNumber) {
+ Map<String, List<String>> deviceMeasurementMap = parseSchema(schema.getFieldNames());
+ HashMap<String, Tablet> tablets = new HashMap<>();
+ deviceMeasurementMap.forEach(
+ (device, measurements) -> {
+ ArrayList<MeasurementSchema> schemas = new ArrayList<>();
+ for (String measurement : measurements) {
+ String tsName = device + "." + measurement;
+ TSDataType dataType = schema.getDataType(tsName);
+ TSEncoding encoding = schema.getEncodingType(tsName);
+ CompressionType compressionType = schema.getCompressionType(tsName);
+ if (encoding == null) {
+ schemas.add(new MeasurementSchema(measurement, dataType));
+ } else if (compressionType == null) {
+ schemas.add(new MeasurementSchema(measurement, dataType, encoding));
+ } else {
+ schemas.add(new MeasurementSchema(measurement, dataType, encoding, compressionType));
+ }
+ }
+ Tablet tablet = new Tablet(device, schemas, maxRowNumber);
+ tablets.put(device, tablet);
+ });
+ return tablets;
+ }
+
+ protected SimpleDateFormat initFormatter(String time) {
Review Comment:
`SimpleDateFormat` is not thread-safe, and Java 8 provides better alternatives in `DateTimeFormatter`. Recommend replacing all references to `SimpleDateFormat` with `DateTimeFormatter`.
##########
nifi-nar-bundles/nifi-iotdb-bundle/nifi-iotdb-processors/src/main/java/org/apache/nifi/processors/PutIoTDB.java:
##########
@@ -0,0 +1,286 @@
+/*
+ * 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;
+
+import java.io.InputStream;
+import java.text.SimpleDateFormat;
+import java.util.Set;
+import java.util.List;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.nifi.processors.model.Schema;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.write.record.Tablet;
+import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnUnscheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.expression.ExpressionLanguageScope;
+import org.apache.nifi.flowfile.FlowFile;
+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.serialization.RecordReader;
+import org.apache.nifi.serialization.RecordReaderFactory;
+import org.apache.nifi.serialization.record.Record;
+import org.apache.nifi.util.Tuple;
+
+@Tags({"iotdb", "insert", "tablet"})
+@InputRequirement(InputRequirement.Requirement.INPUT_REQUIRED)
+@CapabilityDescription(
+ "This is a record aware processor that reads the content of the incoming FlowFile as individual records using the "
+ + "configured 'Record Reader' and writes them to Apache IoTDB using native interface.")
+public class PutIoTDB extends AbstractIoTDB {
+
+ static final PropertyDescriptor RECORD_READER_FACTORY =
+ new PropertyDescriptor.Builder()
+ .name("Record Reader")
+ .description(
+ "Specifies the type of Record Reader controller service to use for parsing the incoming data "
+ + "and determining the schema")
+ .identifiesControllerService(RecordReaderFactory.class)
+ .required(true)
+ .build();
+
+ static final PropertyDescriptor TIME_FIELD =
+ new PropertyDescriptor.Builder()
+ .name("Time Field")
+ .description(
+ "The field name which represents time. It can be updated by expression language.")
+ .defaultValue("Time")
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .required(false)
+ .build();
+
+ static final PropertyDescriptor SCHEMA =
+ new PropertyDescriptor.Builder()
+ .name("Schema")
+ .description(
+ "The schema that IoTDB needs doesn't support good by NiFi.\n"
+ + "Therefore, you can define the schema here.\n"
+ + "Besides, you can set encoding type and compression type by this method.\n"
+ + "If you don't set this property, the inferred schema will be used.\n"
+ + "It can be updated by expression language.")
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .required(false)
+ .build();
+
+ static final PropertyDescriptor ALIGNED =
+ new PropertyDescriptor.Builder()
+ .name("Aligned")
+ .description("Whether using aligned interface? It can be updated by expression language.")
+ .allowableValues(new String[] {"true", "false"})
+ .addValidator(StandardValidators.BOOLEAN_VALIDATOR)
+ .defaultValue("false")
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .required(false)
+ .build();
+
+ static final PropertyDescriptor MAX_ROW_NUMBER =
+ new PropertyDescriptor.Builder()
+ .name("Max Row Number")
+ .description(
+ "Specifies the max row number of each tablet. It can be updated by expression language.")
+ .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
+ .defaultValue("false")
Review Comment:
The default value should be removed if this property is not required.
```suggestion
```
##########
nifi-nar-bundles/nifi-iotdb-bundle/nifi-iotdb-processors/src/main/java/org/apache/nifi/processors/PutIoTDB.java:
##########
@@ -0,0 +1,286 @@
+/*
+ * 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;
+
+import java.io.InputStream;
+import java.text.SimpleDateFormat;
+import java.util.Set;
+import java.util.List;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.nifi.processors.model.Schema;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.write.record.Tablet;
+import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnUnscheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.expression.ExpressionLanguageScope;
+import org.apache.nifi.flowfile.FlowFile;
+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.serialization.RecordReader;
+import org.apache.nifi.serialization.RecordReaderFactory;
+import org.apache.nifi.serialization.record.Record;
+import org.apache.nifi.util.Tuple;
+
+@Tags({"iotdb", "insert", "tablet"})
+@InputRequirement(InputRequirement.Requirement.INPUT_REQUIRED)
+@CapabilityDescription(
+ "This is a record aware processor that reads the content of the incoming FlowFile as individual records using the "
+ + "configured 'Record Reader' and writes them to Apache IoTDB using native interface.")
+public class PutIoTDB extends AbstractIoTDB {
+
+ static final PropertyDescriptor RECORD_READER_FACTORY =
+ new PropertyDescriptor.Builder()
+ .name("Record Reader")
+ .description(
+ "Specifies the type of Record Reader controller service to use for parsing the incoming data "
+ + "and determining the schema")
+ .identifiesControllerService(RecordReaderFactory.class)
+ .required(true)
+ .build();
+
+ static final PropertyDescriptor TIME_FIELD =
+ new PropertyDescriptor.Builder()
+ .name("Time Field")
+ .description(
+ "The field name which represents time. It can be updated by expression language.")
+ .defaultValue("Time")
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .required(false)
+ .build();
+
+ static final PropertyDescriptor SCHEMA =
+ new PropertyDescriptor.Builder()
+ .name("Schema")
+ .description(
+ "The schema that IoTDB needs doesn't support good by NiFi.\n"
+ + "Therefore, you can define the schema here.\n"
+ + "Besides, you can set encoding type and compression type by this method.\n"
+ + "If you don't set this property, the inferred schema will be used.\n"
+ + "It can be updated by expression language.")
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .required(false)
+ .build();
+
+ static final PropertyDescriptor ALIGNED =
+ new PropertyDescriptor.Builder()
+ .name("Aligned")
+ .description("Whether using aligned interface? It can be updated by expression language.")
Review Comment:
```suggestion
.description("Whether to use the Apache IoTDB Aligned Timeseries interface")
```
##########
nifi-nar-bundles/nifi-iotdb-bundle/nifi-iotdb-processors/src/main/java/org/apache/nifi/processors/PutIoTDB.java:
##########
@@ -0,0 +1,286 @@
+/*
+ * 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;
+
+import java.io.InputStream;
+import java.text.SimpleDateFormat;
+import java.util.Set;
+import java.util.List;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.nifi.processors.model.Schema;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.write.record.Tablet;
+import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnUnscheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.expression.ExpressionLanguageScope;
+import org.apache.nifi.flowfile.FlowFile;
+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.serialization.RecordReader;
+import org.apache.nifi.serialization.RecordReaderFactory;
+import org.apache.nifi.serialization.record.Record;
+import org.apache.nifi.util.Tuple;
+
+@Tags({"iotdb", "insert", "tablet"})
+@InputRequirement(InputRequirement.Requirement.INPUT_REQUIRED)
+@CapabilityDescription(
+ "This is a record aware processor that reads the content of the incoming FlowFile as individual records using the "
+ + "configured 'Record Reader' and writes them to Apache IoTDB using native interface.")
+public class PutIoTDB extends AbstractIoTDB {
+
+ static final PropertyDescriptor RECORD_READER_FACTORY =
+ new PropertyDescriptor.Builder()
+ .name("Record Reader")
+ .description(
+ "Specifies the type of Record Reader controller service to use for parsing the incoming data "
+ + "and determining the schema")
+ .identifiesControllerService(RecordReaderFactory.class)
+ .required(true)
+ .build();
+
+ static final PropertyDescriptor TIME_FIELD =
+ new PropertyDescriptor.Builder()
+ .name("Time Field")
+ .description(
+ "The field name which represents time. It can be updated by expression language.")
+ .defaultValue("Time")
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .required(false)
+ .build();
+
+ static final PropertyDescriptor SCHEMA =
+ new PropertyDescriptor.Builder()
+ .name("Schema")
+ .description(
+ "The schema that IoTDB needs doesn't support good by NiFi.\n"
+ + "Therefore, you can define the schema here.\n"
+ + "Besides, you can set encoding type and compression type by this method.\n"
+ + "If you don't set this property, the inferred schema will be used.\n"
+ + "It can be updated by expression language.")
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .required(false)
+ .build();
+
+ static final PropertyDescriptor ALIGNED =
+ new PropertyDescriptor.Builder()
+ .name("Aligned")
+ .description("Whether using aligned interface? It can be updated by expression language.")
+ .allowableValues(new String[] {"true", "false"})
+ .addValidator(StandardValidators.BOOLEAN_VALIDATOR)
+ .defaultValue("false")
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .required(false)
+ .build();
+
+ static final PropertyDescriptor MAX_ROW_NUMBER =
+ new PropertyDescriptor.Builder()
+ .name("Max Row Number")
+ .description(
+ "Specifies the max row number of each tablet. It can be updated by expression language.")
+ .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
+ .defaultValue("false")
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .required(false)
+ .build();
+
+ static {
+ descriptors.add(RECORD_READER_FACTORY);
+ descriptors.add(TIME_FIELD);
+ descriptors.add(SCHEMA);
+ descriptors.add(ALIGNED);
+ descriptors.add(MAX_ROW_NUMBER);
+ }
+
+ @Override
+ protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ return descriptors;
+ }
+
+ @Override
+ public Set<Relationship> getRelationships() {
+ return relationships;
+ }
+
+ @Override
+ public void onTrigger(ProcessContext processContext, ProcessSession processSession)
+ throws ProcessException {
+ final RecordReaderFactory recordParserFactory =
+ processContext
+ .getProperty(RECORD_READER_FACTORY)
+ .asControllerService(RecordReaderFactory.class);
+
+ String timeFieldProperty = processContext.getProperty(TIME_FIELD).getValue();
+ String schemaProperty = processContext.getProperty(SCHEMA).getValue();
+ String alignedProperty = processContext.getProperty(ALIGNED).getValue();
+ String maxRowNumberProperty = processContext.getProperty(MAX_ROW_NUMBER).getValue();
+
+ FlowFile flowFile = processSession.get();
+
+ String timeField = timeFieldProperty != null ? timeFieldProperty : "Time";
+ Boolean aligned = alignedProperty != null ? Boolean.valueOf(alignedProperty) : false;
+ int maxRowNumber = maxRowNumberProperty != null ? Integer.valueOf(maxRowNumberProperty) : 1024;
+
+ try (final InputStream inputStream = processSession.read(flowFile);
+ final RecordReader recordReader =
+ recordParserFactory.createRecordReader(flowFile, inputStream, getLogger())) {
+ if (flowFile == null) {
Review Comment:
This null check should be moved right after `processSession.get()`, instead of inside this try-catch block.
##########
nifi-nar-bundles/nifi-iotdb-bundle/nifi-iotdb-processors/src/main/java/org/apache/nifi/processors/PutIoTDB.java:
##########
@@ -0,0 +1,286 @@
+/*
+ * 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;
+
+import java.io.InputStream;
+import java.text.SimpleDateFormat;
+import java.util.Set;
+import java.util.List;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.nifi.processors.model.Schema;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.write.record.Tablet;
+import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnUnscheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.expression.ExpressionLanguageScope;
+import org.apache.nifi.flowfile.FlowFile;
+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.serialization.RecordReader;
+import org.apache.nifi.serialization.RecordReaderFactory;
+import org.apache.nifi.serialization.record.Record;
+import org.apache.nifi.util.Tuple;
+
+@Tags({"iotdb", "insert", "tablet"})
+@InputRequirement(InputRequirement.Requirement.INPUT_REQUIRED)
+@CapabilityDescription(
+ "This is a record aware processor that reads the content of the incoming FlowFile as individual records using the "
+ + "configured 'Record Reader' and writes them to Apache IoTDB using native interface.")
+public class PutIoTDB extends AbstractIoTDB {
+
+ static final PropertyDescriptor RECORD_READER_FACTORY =
+ new PropertyDescriptor.Builder()
+ .name("Record Reader")
+ .description(
+ "Specifies the type of Record Reader controller service to use for parsing the incoming data "
+ + "and determining the schema")
+ .identifiesControllerService(RecordReaderFactory.class)
+ .required(true)
+ .build();
+
+ static final PropertyDescriptor TIME_FIELD =
+ new PropertyDescriptor.Builder()
+ .name("Time Field")
+ .description(
+ "The field name which represents time. It can be updated by expression language.")
+ .defaultValue("Time")
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .required(false)
+ .build();
+
+ static final PropertyDescriptor SCHEMA =
+ new PropertyDescriptor.Builder()
+ .name("Schema")
+ .description(
+ "The schema that IoTDB needs doesn't support good by NiFi.\n"
+ + "Therefore, you can define the schema here.\n"
+ + "Besides, you can set encoding type and compression type by this method.\n"
+ + "If you don't set this property, the inferred schema will be used.\n"
+ + "It can be updated by expression language.")
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .required(false)
+ .build();
+
+ static final PropertyDescriptor ALIGNED =
+ new PropertyDescriptor.Builder()
+ .name("Aligned")
+ .description("Whether using aligned interface? It can be updated by expression language.")
+ .allowableValues(new String[] {"true", "false"})
+ .addValidator(StandardValidators.BOOLEAN_VALIDATOR)
+ .defaultValue("false")
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .required(false)
+ .build();
+
+ static final PropertyDescriptor MAX_ROW_NUMBER =
+ new PropertyDescriptor.Builder()
+ .name("Max Row Number")
+ .description(
+ "Specifies the max row number of each tablet. It can be updated by expression language.")
+ .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
+ .defaultValue("false")
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .required(false)
+ .build();
+
+ static {
+ descriptors.add(RECORD_READER_FACTORY);
+ descriptors.add(TIME_FIELD);
+ descriptors.add(SCHEMA);
+ descriptors.add(ALIGNED);
+ descriptors.add(MAX_ROW_NUMBER);
+ }
+
+ @Override
+ protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ return descriptors;
+ }
+
+ @Override
+ public Set<Relationship> getRelationships() {
+ return relationships;
+ }
+
+ @Override
+ public void onTrigger(ProcessContext processContext, ProcessSession processSession)
+ throws ProcessException {
+ final RecordReaderFactory recordParserFactory =
+ processContext
+ .getProperty(RECORD_READER_FACTORY)
+ .asControllerService(RecordReaderFactory.class);
+
+ String timeFieldProperty = processContext.getProperty(TIME_FIELD).getValue();
+ String schemaProperty = processContext.getProperty(SCHEMA).getValue();
+ String alignedProperty = processContext.getProperty(ALIGNED).getValue();
+ String maxRowNumberProperty = processContext.getProperty(MAX_ROW_NUMBER).getValue();
Review Comment:
These `getProperty()` calls need to include `evaluateAttributeExpressions(flowFile)` in order to resolve FlowFile-based attributes.
##########
nifi-nar-bundles/nifi-iotdb-bundle/nifi-iotdb-processors/src/main/java/org/apache/nifi/processors/PutIoTDB.java:
##########
@@ -0,0 +1,286 @@
+/*
+ * 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;
+
+import java.io.InputStream;
+import java.text.SimpleDateFormat;
+import java.util.Set;
+import java.util.List;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.nifi.processors.model.Schema;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.write.record.Tablet;
+import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnUnscheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.expression.ExpressionLanguageScope;
+import org.apache.nifi.flowfile.FlowFile;
+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.serialization.RecordReader;
+import org.apache.nifi.serialization.RecordReaderFactory;
+import org.apache.nifi.serialization.record.Record;
+import org.apache.nifi.util.Tuple;
+
+@Tags({"iotdb", "insert", "tablet"})
+@InputRequirement(InputRequirement.Requirement.INPUT_REQUIRED)
+@CapabilityDescription(
+ "This is a record aware processor that reads the content of the incoming FlowFile as individual records using the "
+ + "configured 'Record Reader' and writes them to Apache IoTDB using native interface.")
+public class PutIoTDB extends AbstractIoTDB {
+
+ static final PropertyDescriptor RECORD_READER_FACTORY =
+ new PropertyDescriptor.Builder()
+ .name("Record Reader")
+ .description(
+ "Specifies the type of Record Reader controller service to use for parsing the incoming data "
+ + "and determining the schema")
+ .identifiesControllerService(RecordReaderFactory.class)
+ .required(true)
+ .build();
+
+ static final PropertyDescriptor TIME_FIELD =
+ new PropertyDescriptor.Builder()
+ .name("Time Field")
+ .description(
+ "The field name which represents time. It can be updated by expression language.")
+ .defaultValue("Time")
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .required(false)
+ .build();
+
+ static final PropertyDescriptor SCHEMA =
+ new PropertyDescriptor.Builder()
+ .name("Schema")
+ .description(
+ "The schema that IoTDB needs doesn't support good by NiFi.\n"
+ + "Therefore, you can define the schema here.\n"
+ + "Besides, you can set encoding type and compression type by this method.\n"
+ + "If you don't set this property, the inferred schema will be used.\n"
+ + "It can be updated by expression language.")
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .required(false)
+ .build();
+
+ static final PropertyDescriptor ALIGNED =
+ new PropertyDescriptor.Builder()
+ .name("Aligned")
+ .description("Whether using aligned interface? It can be updated by expression language.")
+ .allowableValues(new String[] {"true", "false"})
+ .addValidator(StandardValidators.BOOLEAN_VALIDATOR)
+ .defaultValue("false")
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .required(false)
+ .build();
+
+ static final PropertyDescriptor MAX_ROW_NUMBER =
+ new PropertyDescriptor.Builder()
+ .name("Max Row Number")
+ .description(
+ "Specifies the max row number of each tablet. It can be updated by expression language.")
+ .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
+ .defaultValue("false")
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .required(false)
+ .build();
+
+ static {
+ descriptors.add(RECORD_READER_FACTORY);
+ descriptors.add(TIME_FIELD);
+ descriptors.add(SCHEMA);
+ descriptors.add(ALIGNED);
+ descriptors.add(MAX_ROW_NUMBER);
+ }
+
+ @Override
+ protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ return descriptors;
+ }
+
+ @Override
+ public Set<Relationship> getRelationships() {
+ return relationships;
+ }
+
+ @Override
+ public void onTrigger(ProcessContext processContext, ProcessSession processSession)
+ throws ProcessException {
+ final RecordReaderFactory recordParserFactory =
+ processContext
+ .getProperty(RECORD_READER_FACTORY)
+ .asControllerService(RecordReaderFactory.class);
+
+ String timeFieldProperty = processContext.getProperty(TIME_FIELD).getValue();
+ String schemaProperty = processContext.getProperty(SCHEMA).getValue();
+ String alignedProperty = processContext.getProperty(ALIGNED).getValue();
+ String maxRowNumberProperty = processContext.getProperty(MAX_ROW_NUMBER).getValue();
+
+ FlowFile flowFile = processSession.get();
+
+ String timeField = timeFieldProperty != null ? timeFieldProperty : "Time";
+ Boolean aligned = alignedProperty != null ? Boolean.valueOf(alignedProperty) : false;
+ int maxRowNumber = maxRowNumberProperty != null ? Integer.valueOf(maxRowNumberProperty) : 1024;
+
+ try (final InputStream inputStream = processSession.read(flowFile);
+ final RecordReader recordReader =
+ recordParserFactory.createRecordReader(flowFile, inputStream, getLogger())) {
+ if (flowFile == null) {
+ inputStream.close();
+ recordReader.close();
+ processSession.transfer(flowFile, REL_SUCCESS);
+ return;
+ }
+
+ HashMap<String, Tablet> tablets;
+ boolean needInitFormatter;
+ Schema schema;
+ Tuple<Boolean, String> result;
+
+ result =
+ schemaProperty != null
+ ? validateSchemaAttribute(schemaProperty)
+ : validateSchema(recordReader.getSchema());
+
+ if (!result.getKey()) {
+ getLogger().error(result.getValue());
+ inputStream.close();
+ recordReader.close();
+ processSession.transfer(flowFile, REL_FAILURE);
+ return;
+ } else {
+ if (result.getValue() != null) {
+ getLogger().warn(result.getValue());
+ }
+ }
+
+ schema =
+ schemaProperty != null
+ ? new ObjectMapper().readValue(schemaProperty, Schema.class)
+ : convertSchema(recordReader.getSchema());
+
+ List<String> fieldNames = schema.getFieldNames();
+
+ needInitFormatter = schema.getTimeType() != Schema.TimeType.LONG;
+
+ tablets = generateTablets(schema, maxRowNumber);
+ SimpleDateFormat timeFormatter = null;
+
+ Record record;
+
+ while ((record = recordReader.nextRecord()) != null) {
+ Object[] values = record.getValues();
+ if (timeFormatter == null && needInitFormatter) {
+ timeFormatter = initFormatter((String) values[0]);
+ if (timeFormatter == null) {
+ getLogger().error("The format of time is not supported.");
Review Comment:
This message should include some additional details, such as the FlowFile and record number:
```suggestion
getLogger().error("{} Record [{}] time format not supported", flowFile, recordNumber);
```
##########
nifi-nar-bundles/nifi-iotdb-bundle/nifi-iotdb-processors/src/main/java/org/apache/nifi/processors/PutIoTDB.java:
##########
@@ -0,0 +1,286 @@
+/*
+ * 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;
+
+import java.io.InputStream;
+import java.text.SimpleDateFormat;
+import java.util.Set;
+import java.util.List;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.nifi.processors.model.Schema;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.write.record.Tablet;
+import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnUnscheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.expression.ExpressionLanguageScope;
+import org.apache.nifi.flowfile.FlowFile;
+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.serialization.RecordReader;
+import org.apache.nifi.serialization.RecordReaderFactory;
+import org.apache.nifi.serialization.record.Record;
+import org.apache.nifi.util.Tuple;
+
+@Tags({"iotdb", "insert", "tablet"})
+@InputRequirement(InputRequirement.Requirement.INPUT_REQUIRED)
+@CapabilityDescription(
+ "This is a record aware processor that reads the content of the incoming FlowFile as individual records using the "
+ + "configured 'Record Reader' and writes them to Apache IoTDB using native interface.")
+public class PutIoTDB extends AbstractIoTDB {
+
+ static final PropertyDescriptor RECORD_READER_FACTORY =
+ new PropertyDescriptor.Builder()
+ .name("Record Reader")
+ .description(
+ "Specifies the type of Record Reader controller service to use for parsing the incoming data "
+ + "and determining the schema")
+ .identifiesControllerService(RecordReaderFactory.class)
+ .required(true)
+ .build();
+
+ static final PropertyDescriptor TIME_FIELD =
+ new PropertyDescriptor.Builder()
+ .name("Time Field")
+ .description(
+ "The field name which represents time. It can be updated by expression language.")
+ .defaultValue("Time")
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .required(false)
+ .build();
+
+ static final PropertyDescriptor SCHEMA =
+ new PropertyDescriptor.Builder()
+ .name("Schema")
+ .description(
+ "The schema that IoTDB needs doesn't support good by NiFi.\n"
+ + "Therefore, you can define the schema here.\n"
+ + "Besides, you can set encoding type and compression type by this method.\n"
+ + "If you don't set this property, the inferred schema will be used.\n"
+ + "It can be updated by expression language.")
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .required(false)
+ .build();
+
+ static final PropertyDescriptor ALIGNED =
+ new PropertyDescriptor.Builder()
+ .name("Aligned")
+ .description("Whether using aligned interface? It can be updated by expression language.")
+ .allowableValues(new String[] {"true", "false"})
+ .addValidator(StandardValidators.BOOLEAN_VALIDATOR)
+ .defaultValue("false")
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .required(false)
+ .build();
+
+ static final PropertyDescriptor MAX_ROW_NUMBER =
+ new PropertyDescriptor.Builder()
+ .name("Max Row Number")
+ .description(
+ "Specifies the max row number of each tablet. It can be updated by expression language.")
+ .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
+ .defaultValue("false")
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .required(false)
+ .build();
+
+ static {
+ descriptors.add(RECORD_READER_FACTORY);
+ descriptors.add(TIME_FIELD);
+ descriptors.add(SCHEMA);
+ descriptors.add(ALIGNED);
+ descriptors.add(MAX_ROW_NUMBER);
+ }
+
+ @Override
+ protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ return descriptors;
+ }
+
+ @Override
+ public Set<Relationship> getRelationships() {
+ return relationships;
+ }
+
+ @Override
+ public void onTrigger(ProcessContext processContext, ProcessSession processSession)
+ throws ProcessException {
+ final RecordReaderFactory recordParserFactory =
+ processContext
+ .getProperty(RECORD_READER_FACTORY)
+ .asControllerService(RecordReaderFactory.class);
+
+ String timeFieldProperty = processContext.getProperty(TIME_FIELD).getValue();
+ String schemaProperty = processContext.getProperty(SCHEMA).getValue();
+ String alignedProperty = processContext.getProperty(ALIGNED).getValue();
+ String maxRowNumberProperty = processContext.getProperty(MAX_ROW_NUMBER).getValue();
+
+ FlowFile flowFile = processSession.get();
+
+ String timeField = timeFieldProperty != null ? timeFieldProperty : "Time";
+ Boolean aligned = alignedProperty != null ? Boolean.valueOf(alignedProperty) : false;
+ int maxRowNumber = maxRowNumberProperty != null ? Integer.valueOf(maxRowNumberProperty) : 1024;
+
+ try (final InputStream inputStream = processSession.read(flowFile);
+ final RecordReader recordReader =
+ recordParserFactory.createRecordReader(flowFile, inputStream, getLogger())) {
+ if (flowFile == null) {
+ inputStream.close();
+ recordReader.close();
+ processSession.transfer(flowFile, REL_SUCCESS);
+ return;
+ }
+
+ HashMap<String, Tablet> tablets;
+ boolean needInitFormatter;
+ Schema schema;
+ Tuple<Boolean, String> result;
+
+ result =
+ schemaProperty != null
+ ? validateSchemaAttribute(schemaProperty)
+ : validateSchema(recordReader.getSchema());
+
+ if (!result.getKey()) {
+ getLogger().error(result.getValue());
+ inputStream.close();
+ recordReader.close();
+ processSession.transfer(flowFile, REL_FAILURE);
+ return;
+ } else {
+ if (result.getValue() != null) {
+ getLogger().warn(result.getValue());
+ }
+ }
+
+ schema =
+ schemaProperty != null
+ ? new ObjectMapper().readValue(schemaProperty, Schema.class)
+ : convertSchema(recordReader.getSchema());
+
+ List<String> fieldNames = schema.getFieldNames();
+
+ needInitFormatter = schema.getTimeType() != Schema.TimeType.LONG;
+
+ tablets = generateTablets(schema, maxRowNumber);
+ SimpleDateFormat timeFormatter = null;
+
+ Record record;
+
+ while ((record = recordReader.nextRecord()) != null) {
+ Object[] values = record.getValues();
+ if (timeFormatter == null && needInitFormatter) {
+ timeFormatter = initFormatter((String) values[0]);
+ if (timeFormatter == null) {
+ getLogger().error("The format of time is not supported.");
+ inputStream.close();
+ recordReader.close();
+ processSession.transfer(flowFile, REL_FAILURE);
+ return;
+ }
+ }
+
+ long timestamp;
+ if (needInitFormatter) {
+ timestamp = timeFormatter.parse((String) values[0]).getTime();
+ } else {
+ timestamp = (Long) values[0];
+ }
+
+ boolean flag = false;
+
+ for (Map.Entry<String, Tablet> entry : tablets.entrySet()) {
+ String device = entry.getKey();
+ Tablet tablet = entry.getValue();
+ int rowIndex = tablet.rowSize++;
+
+ tablet.addTimestamp(rowIndex, timestamp);
+ List<MeasurementSchema> measurements = tablet.getSchemas();
+ for (MeasurementSchema measurement : measurements) {
+ String tsName =
+ new StringBuilder()
+ .append(device)
+ .append(".")
+ .append(measurement.getMeasurementId())
+ .toString();
+ int valueIndex = fieldNames.indexOf(tsName) + 1;
+ Object value;
+ try {
+ TSDataType type = measurement.getType();
+ value = convertType(values[valueIndex], type);
+ } catch (Exception e) {
Review Comment:
It looks like this exception should be logged, at least at the debug level.
##########
nifi-nar-bundles/nifi-iotdb-bundle/nifi-iotdb-processors/src/test/java/org/apache/nifi/processors/AbstractIoTDBUT.java:
##########
@@ -0,0 +1,322 @@
+/*
+ * 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;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.nifi.processors.model.Schema;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSEncoding;
+import org.apache.iotdb.tsfile.write.record.Tablet;
+import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.util.Tuple;
+import org.junit.Assert;
Review Comment:
The JUnit 5 `Assertions` static methods should be used instead of JUnit 4.
##########
nifi-nar-bundles/nifi-iotdb-bundle/nifi-iotdb-processors/src/main/java/org/apache/nifi/processors/PutIoTDB.java:
##########
@@ -0,0 +1,286 @@
+/*
+ * 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;
+
+import java.io.InputStream;
+import java.text.SimpleDateFormat;
+import java.util.Set;
+import java.util.List;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.nifi.processors.model.Schema;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.write.record.Tablet;
+import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnUnscheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.expression.ExpressionLanguageScope;
+import org.apache.nifi.flowfile.FlowFile;
+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.serialization.RecordReader;
+import org.apache.nifi.serialization.RecordReaderFactory;
+import org.apache.nifi.serialization.record.Record;
+import org.apache.nifi.util.Tuple;
+
+@Tags({"iotdb", "insert", "tablet"})
+@InputRequirement(InputRequirement.Requirement.INPUT_REQUIRED)
+@CapabilityDescription(
+ "This is a record aware processor that reads the content of the incoming FlowFile as individual records using the "
+ + "configured 'Record Reader' and writes them to Apache IoTDB using native interface.")
+public class PutIoTDB extends AbstractIoTDB {
+
+ static final PropertyDescriptor RECORD_READER_FACTORY =
+ new PropertyDescriptor.Builder()
+ .name("Record Reader")
+ .description(
+ "Specifies the type of Record Reader controller service to use for parsing the incoming data "
+ + "and determining the schema")
+ .identifiesControllerService(RecordReaderFactory.class)
+ .required(true)
+ .build();
+
+ static final PropertyDescriptor TIME_FIELD =
+ new PropertyDescriptor.Builder()
+ .name("Time Field")
+ .description(
+ "The field name which represents time. It can be updated by expression language.")
+ .defaultValue("Time")
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .required(false)
+ .build();
+
+ static final PropertyDescriptor SCHEMA =
+ new PropertyDescriptor.Builder()
+ .name("Schema")
+ .description(
+ "The schema that IoTDB needs doesn't support good by NiFi.\n"
+ + "Therefore, you can define the schema here.\n"
+ + "Besides, you can set encoding type and compression type by this method.\n"
+ + "If you don't set this property, the inferred schema will be used.\n"
+ + "It can be updated by expression language.")
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .required(false)
+ .build();
+
+ static final PropertyDescriptor ALIGNED =
+ new PropertyDescriptor.Builder()
+ .name("Aligned")
+ .description("Whether using aligned interface? It can be updated by expression language.")
+ .allowableValues(new String[] {"true", "false"})
+ .addValidator(StandardValidators.BOOLEAN_VALIDATOR)
+ .defaultValue("false")
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .required(false)
+ .build();
+
+ static final PropertyDescriptor MAX_ROW_NUMBER =
+ new PropertyDescriptor.Builder()
+ .name("Max Row Number")
+ .description(
+ "Specifies the max row number of each tablet. It can be updated by expression language.")
+ .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
+ .defaultValue("false")
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .required(false)
+ .build();
+
+ static {
+ descriptors.add(RECORD_READER_FACTORY);
+ descriptors.add(TIME_FIELD);
+ descriptors.add(SCHEMA);
+ descriptors.add(ALIGNED);
+ descriptors.add(MAX_ROW_NUMBER);
+ }
+
+ @Override
+ protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ return descriptors;
+ }
+
+ @Override
+ public Set<Relationship> getRelationships() {
+ return relationships;
+ }
+
+ @Override
+ public void onTrigger(ProcessContext processContext, ProcessSession processSession)
+ throws ProcessException {
+ final RecordReaderFactory recordParserFactory =
+ processContext
+ .getProperty(RECORD_READER_FACTORY)
+ .asControllerService(RecordReaderFactory.class);
+
+ String timeFieldProperty = processContext.getProperty(TIME_FIELD).getValue();
+ String schemaProperty = processContext.getProperty(SCHEMA).getValue();
+ String alignedProperty = processContext.getProperty(ALIGNED).getValue();
+ String maxRowNumberProperty = processContext.getProperty(MAX_ROW_NUMBER).getValue();
+
+ FlowFile flowFile = processSession.get();
+
+ String timeField = timeFieldProperty != null ? timeFieldProperty : "Time";
+ Boolean aligned = alignedProperty != null ? Boolean.valueOf(alignedProperty) : false;
Review Comment:
```suggestion
final boolean aligned = alignedProperty != null ? Boolean.valueOf(alignedProperty) : false;
```
##########
nifi-nar-bundles/nifi-iotdb-bundle/nifi-iotdb-processors/src/main/java/org/apache/nifi/processors/PutIoTDB.java:
##########
@@ -0,0 +1,286 @@
+/*
+ * 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;
+
+import java.io.InputStream;
+import java.text.SimpleDateFormat;
+import java.util.Set;
+import java.util.List;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.nifi.processors.model.Schema;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.write.record.Tablet;
+import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnUnscheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.expression.ExpressionLanguageScope;
+import org.apache.nifi.flowfile.FlowFile;
+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.serialization.RecordReader;
+import org.apache.nifi.serialization.RecordReaderFactory;
+import org.apache.nifi.serialization.record.Record;
+import org.apache.nifi.util.Tuple;
+
+@Tags({"iotdb", "insert", "tablet"})
+@InputRequirement(InputRequirement.Requirement.INPUT_REQUIRED)
+@CapabilityDescription(
+ "This is a record aware processor that reads the content of the incoming FlowFile as individual records using the "
+ + "configured 'Record Reader' and writes them to Apache IoTDB using native interface.")
+public class PutIoTDB extends AbstractIoTDB {
+
+ static final PropertyDescriptor RECORD_READER_FACTORY =
+ new PropertyDescriptor.Builder()
+ .name("Record Reader")
+ .description(
+ "Specifies the type of Record Reader controller service to use for parsing the incoming data "
+ + "and determining the schema")
+ .identifiesControllerService(RecordReaderFactory.class)
+ .required(true)
+ .build();
+
+ static final PropertyDescriptor TIME_FIELD =
+ new PropertyDescriptor.Builder()
+ .name("Time Field")
+ .description(
+ "The field name which represents time. It can be updated by expression language.")
+ .defaultValue("Time")
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .required(false)
+ .build();
+
+ static final PropertyDescriptor SCHEMA =
+ new PropertyDescriptor.Builder()
+ .name("Schema")
+ .description(
+ "The schema that IoTDB needs doesn't support good by NiFi.\n"
+ + "Therefore, you can define the schema here.\n"
+ + "Besides, you can set encoding type and compression type by this method.\n"
+ + "If you don't set this property, the inferred schema will be used.\n"
+ + "It can be updated by expression language.")
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .required(false)
+ .build();
+
+ static final PropertyDescriptor ALIGNED =
+ new PropertyDescriptor.Builder()
+ .name("Aligned")
+ .description("Whether using aligned interface? It can be updated by expression language.")
+ .allowableValues(new String[] {"true", "false"})
+ .addValidator(StandardValidators.BOOLEAN_VALIDATOR)
+ .defaultValue("false")
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .required(false)
+ .build();
+
+ static final PropertyDescriptor MAX_ROW_NUMBER =
+ new PropertyDescriptor.Builder()
+ .name("Max Row Number")
+ .description(
+ "Specifies the max row number of each tablet. It can be updated by expression language.")
+ .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
+ .defaultValue("false")
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .required(false)
+ .build();
+
+ static {
+ descriptors.add(RECORD_READER_FACTORY);
+ descriptors.add(TIME_FIELD);
+ descriptors.add(SCHEMA);
+ descriptors.add(ALIGNED);
+ descriptors.add(MAX_ROW_NUMBER);
+ }
+
+ @Override
+ protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ return descriptors;
+ }
+
+ @Override
+ public Set<Relationship> getRelationships() {
+ return relationships;
+ }
+
+ @Override
+ public void onTrigger(ProcessContext processContext, ProcessSession processSession)
+ throws ProcessException {
+ final RecordReaderFactory recordParserFactory =
+ processContext
+ .getProperty(RECORD_READER_FACTORY)
+ .asControllerService(RecordReaderFactory.class);
+
+ String timeFieldProperty = processContext.getProperty(TIME_FIELD).getValue();
+ String schemaProperty = processContext.getProperty(SCHEMA).getValue();
+ String alignedProperty = processContext.getProperty(ALIGNED).getValue();
+ String maxRowNumberProperty = processContext.getProperty(MAX_ROW_NUMBER).getValue();
+
+ FlowFile flowFile = processSession.get();
+
+ String timeField = timeFieldProperty != null ? timeFieldProperty : "Time";
+ Boolean aligned = alignedProperty != null ? Boolean.valueOf(alignedProperty) : false;
+ int maxRowNumber = maxRowNumberProperty != null ? Integer.valueOf(maxRowNumberProperty) : 1024;
+
+ try (final InputStream inputStream = processSession.read(flowFile);
+ final RecordReader recordReader =
+ recordParserFactory.createRecordReader(flowFile, inputStream, getLogger())) {
+ if (flowFile == null) {
+ inputStream.close();
+ recordReader.close();
+ processSession.transfer(flowFile, REL_SUCCESS);
+ return;
+ }
+
+ HashMap<String, Tablet> tablets;
+ boolean needInitFormatter;
+ Schema schema;
+ Tuple<Boolean, String> result;
+
+ result =
+ schemaProperty != null
+ ? validateSchemaAttribute(schemaProperty)
+ : validateSchema(recordReader.getSchema());
+
+ if (!result.getKey()) {
+ getLogger().error(result.getValue());
+ inputStream.close();
+ recordReader.close();
+ processSession.transfer(flowFile, REL_FAILURE);
+ return;
+ } else {
+ if (result.getValue() != null) {
+ getLogger().warn(result.getValue());
+ }
+ }
+
+ schema =
+ schemaProperty != null
+ ? new ObjectMapper().readValue(schemaProperty, Schema.class)
+ : convertSchema(recordReader.getSchema());
+
+ List<String> fieldNames = schema.getFieldNames();
+
+ needInitFormatter = schema.getTimeType() != Schema.TimeType.LONG;
+
+ tablets = generateTablets(schema, maxRowNumber);
+ SimpleDateFormat timeFormatter = null;
+
+ Record record;
+
+ while ((record = recordReader.nextRecord()) != null) {
+ Object[] values = record.getValues();
+ if (timeFormatter == null && needInitFormatter) {
+ timeFormatter = initFormatter((String) values[0]);
+ if (timeFormatter == null) {
+ getLogger().error("The format of time is not supported.");
+ inputStream.close();
+ recordReader.close();
+ processSession.transfer(flowFile, REL_FAILURE);
+ return;
+ }
+ }
+
+ long timestamp;
+ if (needInitFormatter) {
+ timestamp = timeFormatter.parse((String) values[0]).getTime();
+ } else {
+ timestamp = (Long) values[0];
+ }
+
+ boolean flag = false;
+
+ for (Map.Entry<String, Tablet> entry : tablets.entrySet()) {
+ String device = entry.getKey();
+ Tablet tablet = entry.getValue();
+ int rowIndex = tablet.rowSize++;
+
+ tablet.addTimestamp(rowIndex, timestamp);
+ List<MeasurementSchema> measurements = tablet.getSchemas();
+ for (MeasurementSchema measurement : measurements) {
+ String tsName =
+ new StringBuilder()
+ .append(device)
+ .append(".")
+ .append(measurement.getMeasurementId())
+ .toString();
+ int valueIndex = fieldNames.indexOf(tsName) + 1;
+ Object value;
+ try {
+ TSDataType type = measurement.getType();
+ value = convertType(values[valueIndex], type);
+ } catch (Exception e) {
+ value = null;
+ }
+ tablet.addValue(measurement.getMeasurementId(), rowIndex, value);
+ }
+ flag = tablet.rowSize == tablet.getMaxRowNumber();
+ }
Review Comment:
It would be helpful to refactor this method into smaller component parts to avoid deeply-nested loops.
##########
nifi-nar-bundles/nifi-iotdb-bundle/nifi-iotdb-processors/src/main/java/org/apache/nifi/processors/PutIoTDB.java:
##########
@@ -0,0 +1,286 @@
+/*
+ * 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;
+
+import java.io.InputStream;
+import java.text.SimpleDateFormat;
+import java.util.Set;
+import java.util.List;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.nifi.processors.model.Schema;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.write.record.Tablet;
+import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnUnscheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.expression.ExpressionLanguageScope;
+import org.apache.nifi.flowfile.FlowFile;
+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.serialization.RecordReader;
+import org.apache.nifi.serialization.RecordReaderFactory;
+import org.apache.nifi.serialization.record.Record;
+import org.apache.nifi.util.Tuple;
+
+@Tags({"iotdb", "insert", "tablet"})
+@InputRequirement(InputRequirement.Requirement.INPUT_REQUIRED)
+@CapabilityDescription(
+ "This is a record aware processor that reads the content of the incoming FlowFile as individual records using the "
+ + "configured 'Record Reader' and writes them to Apache IoTDB using native interface.")
+public class PutIoTDB extends AbstractIoTDB {
+
+ static final PropertyDescriptor RECORD_READER_FACTORY =
+ new PropertyDescriptor.Builder()
+ .name("Record Reader")
+ .description(
+ "Specifies the type of Record Reader controller service to use for parsing the incoming data "
+ + "and determining the schema")
+ .identifiesControllerService(RecordReaderFactory.class)
+ .required(true)
+ .build();
+
+ static final PropertyDescriptor TIME_FIELD =
+ new PropertyDescriptor.Builder()
+ .name("Time Field")
+ .description(
+ "The field name which represents time. It can be updated by expression language.")
+ .defaultValue("Time")
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .required(false)
+ .build();
+
+ static final PropertyDescriptor SCHEMA =
+ new PropertyDescriptor.Builder()
+ .name("Schema")
+ .description(
+ "The schema that IoTDB needs doesn't support good by NiFi.\n"
+ + "Therefore, you can define the schema here.\n"
+ + "Besides, you can set encoding type and compression type by this method.\n"
+ + "If you don't set this property, the inferred schema will be used.\n"
+ + "It can be updated by expression language.")
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .required(false)
+ .build();
+
+ static final PropertyDescriptor ALIGNED =
+ new PropertyDescriptor.Builder()
+ .name("Aligned")
+ .description("Whether using aligned interface? It can be updated by expression language.")
+ .allowableValues(new String[] {"true", "false"})
+ .addValidator(StandardValidators.BOOLEAN_VALIDATOR)
+ .defaultValue("false")
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .required(false)
+ .build();
+
+ static final PropertyDescriptor MAX_ROW_NUMBER =
+ new PropertyDescriptor.Builder()
+ .name("Max Row Number")
+ .description(
+ "Specifies the max row number of each tablet. It can be updated by expression language.")
+ .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
+ .defaultValue("false")
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .required(false)
+ .build();
+
+ static {
+ descriptors.add(RECORD_READER_FACTORY);
+ descriptors.add(TIME_FIELD);
+ descriptors.add(SCHEMA);
+ descriptors.add(ALIGNED);
+ descriptors.add(MAX_ROW_NUMBER);
+ }
+
+ @Override
+ protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ return descriptors;
+ }
+
+ @Override
+ public Set<Relationship> getRelationships() {
+ return relationships;
+ }
+
+ @Override
+ public void onTrigger(ProcessContext processContext, ProcessSession processSession)
+ throws ProcessException {
+ final RecordReaderFactory recordParserFactory =
+ processContext
+ .getProperty(RECORD_READER_FACTORY)
+ .asControllerService(RecordReaderFactory.class);
+
+ String timeFieldProperty = processContext.getProperty(TIME_FIELD).getValue();
+ String schemaProperty = processContext.getProperty(SCHEMA).getValue();
+ String alignedProperty = processContext.getProperty(ALIGNED).getValue();
+ String maxRowNumberProperty = processContext.getProperty(MAX_ROW_NUMBER).getValue();
+
+ FlowFile flowFile = processSession.get();
+
+ String timeField = timeFieldProperty != null ? timeFieldProperty : "Time";
+ Boolean aligned = alignedProperty != null ? Boolean.valueOf(alignedProperty) : false;
+ int maxRowNumber = maxRowNumberProperty != null ? Integer.valueOf(maxRowNumberProperty) : 1024;
+
+ try (final InputStream inputStream = processSession.read(flowFile);
+ final RecordReader recordReader =
+ recordParserFactory.createRecordReader(flowFile, inputStream, getLogger())) {
+ if (flowFile == null) {
+ inputStream.close();
+ recordReader.close();
+ processSession.transfer(flowFile, REL_SUCCESS);
+ return;
+ }
+
+ HashMap<String, Tablet> tablets;
+ boolean needInitFormatter;
+ Schema schema;
+ Tuple<Boolean, String> result;
+
+ result =
+ schemaProperty != null
+ ? validateSchemaAttribute(schemaProperty)
+ : validateSchema(recordReader.getSchema());
+
+ if (!result.getKey()) {
+ getLogger().error(result.getValue());
+ inputStream.close();
+ recordReader.close();
+ processSession.transfer(flowFile, REL_FAILURE);
+ return;
+ } else {
+ if (result.getValue() != null) {
+ getLogger().warn(result.getValue());
+ }
+ }
+
+ schema =
+ schemaProperty != null
+ ? new ObjectMapper().readValue(schemaProperty, Schema.class)
+ : convertSchema(recordReader.getSchema());
+
+ List<String> fieldNames = schema.getFieldNames();
+
+ needInitFormatter = schema.getTimeType() != Schema.TimeType.LONG;
+
+ tablets = generateTablets(schema, maxRowNumber);
+ SimpleDateFormat timeFormatter = null;
+
+ Record record;
+
+ while ((record = recordReader.nextRecord()) != null) {
+ Object[] values = record.getValues();
+ if (timeFormatter == null && needInitFormatter) {
+ timeFormatter = initFormatter((String) values[0]);
+ if (timeFormatter == null) {
+ getLogger().error("The format of time is not supported.");
+ inputStream.close();
+ recordReader.close();
+ processSession.transfer(flowFile, REL_FAILURE);
+ return;
+ }
+ }
+
+ long timestamp;
+ if (needInitFormatter) {
+ timestamp = timeFormatter.parse((String) values[0]).getTime();
+ } else {
+ timestamp = (Long) values[0];
+ }
+
+ boolean flag = false;
+
+ for (Map.Entry<String, Tablet> entry : tablets.entrySet()) {
+ String device = entry.getKey();
+ Tablet tablet = entry.getValue();
+ int rowIndex = tablet.rowSize++;
+
+ tablet.addTimestamp(rowIndex, timestamp);
+ List<MeasurementSchema> measurements = tablet.getSchemas();
+ for (MeasurementSchema measurement : measurements) {
+ String tsName =
+ new StringBuilder()
+ .append(device)
+ .append(".")
+ .append(measurement.getMeasurementId())
+ .toString();
+ int valueIndex = fieldNames.indexOf(tsName) + 1;
+ Object value;
+ try {
+ TSDataType type = measurement.getType();
+ value = convertType(values[valueIndex], type);
+ } catch (Exception e) {
+ value = null;
+ }
+ tablet.addValue(measurement.getMeasurementId(), rowIndex, value);
+ }
+ flag = tablet.rowSize == tablet.getMaxRowNumber();
+ }
+ if (flag) {
+ if (aligned) {
+ session.get().insertAlignedTablets(tablets);
+ } else {
+ session.get().insertTablets(tablets);
+ }
+ tablets.values().forEach(tablet -> tablet.reset());
+ }
+ }
+
+ AtomicBoolean hasRest = new AtomicBoolean(false);
+ tablets.forEach(
+ (device, tablet) -> {
+ if (hasRest.get() == false && tablet.rowSize != 0) {
+ hasRest.set(true);
+ }
+ });
+ if (hasRest.get()) {
+ if (aligned) {
+ session.get().insertAlignedTablets(tablets);
+ } else {
+ session.get().insertTablets(tablets);
+ }
+ }
+
+ inputStream.close();
+ recordReader.close();
+ processSession.transfer(flowFile, REL_SUCCESS);
+ } catch (Exception e) {
+ getLogger().error(e.getMessage());
Review Comment:
The exception should be logged with a message to ensure the full stack trace is available:
```suggestion
getLogger().error("Processing failed {}", flowFile, e);
```
##########
nifi-nar-bundles/nifi-iotdb-bundle/nifi-iotdb-processors/src/main/java/org/apache/nifi/processors/PutIoTDB.java:
##########
@@ -0,0 +1,286 @@
+/*
+ * 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;
+
+import java.io.InputStream;
+import java.text.SimpleDateFormat;
+import java.util.Set;
+import java.util.List;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.nifi.processors.model.Schema;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.write.record.Tablet;
+import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnUnscheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.expression.ExpressionLanguageScope;
+import org.apache.nifi.flowfile.FlowFile;
+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.serialization.RecordReader;
+import org.apache.nifi.serialization.RecordReaderFactory;
+import org.apache.nifi.serialization.record.Record;
+import org.apache.nifi.util.Tuple;
+
+@Tags({"iotdb", "insert", "tablet"})
+@InputRequirement(InputRequirement.Requirement.INPUT_REQUIRED)
+@CapabilityDescription(
+ "This is a record aware processor that reads the content of the incoming FlowFile as individual records using the "
+ + "configured 'Record Reader' and writes them to Apache IoTDB using native interface.")
+public class PutIoTDB extends AbstractIoTDB {
+
+ static final PropertyDescriptor RECORD_READER_FACTORY =
+ new PropertyDescriptor.Builder()
+ .name("Record Reader")
+ .description(
+ "Specifies the type of Record Reader controller service to use for parsing the incoming data "
+ + "and determining the schema")
+ .identifiesControllerService(RecordReaderFactory.class)
+ .required(true)
+ .build();
+
+ static final PropertyDescriptor TIME_FIELD =
+ new PropertyDescriptor.Builder()
+ .name("Time Field")
+ .description(
+ "The field name which represents time. It can be updated by expression language.")
+ .defaultValue("Time")
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .required(false)
+ .build();
+
+ static final PropertyDescriptor SCHEMA =
+ new PropertyDescriptor.Builder()
+ .name("Schema")
+ .description(
+ "The schema that IoTDB needs doesn't support good by NiFi.\n"
+ + "Therefore, you can define the schema here.\n"
+ + "Besides, you can set encoding type and compression type by this method.\n"
+ + "If you don't set this property, the inferred schema will be used.\n"
+ + "It can be updated by expression language.")
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .required(false)
+ .build();
+
+ static final PropertyDescriptor ALIGNED =
+ new PropertyDescriptor.Builder()
+ .name("Aligned")
+ .description("Whether using aligned interface? It can be updated by expression language.")
+ .allowableValues(new String[] {"true", "false"})
+ .addValidator(StandardValidators.BOOLEAN_VALIDATOR)
+ .defaultValue("false")
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .required(false)
+ .build();
+
+ static final PropertyDescriptor MAX_ROW_NUMBER =
+ new PropertyDescriptor.Builder()
+ .name("Max Row Number")
+ .description(
+ "Specifies the max row number of each tablet. It can be updated by expression language.")
+ .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
+ .defaultValue("false")
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .required(false)
+ .build();
+
+ static {
+ descriptors.add(RECORD_READER_FACTORY);
+ descriptors.add(TIME_FIELD);
+ descriptors.add(SCHEMA);
+ descriptors.add(ALIGNED);
+ descriptors.add(MAX_ROW_NUMBER);
+ }
+
+ @Override
+ protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ return descriptors;
+ }
+
+ @Override
+ public Set<Relationship> getRelationships() {
+ return relationships;
+ }
+
+ @Override
+ public void onTrigger(ProcessContext processContext, ProcessSession processSession)
+ throws ProcessException {
+ final RecordReaderFactory recordParserFactory =
+ processContext
+ .getProperty(RECORD_READER_FACTORY)
+ .asControllerService(RecordReaderFactory.class);
+
+ String timeFieldProperty = processContext.getProperty(TIME_FIELD).getValue();
+ String schemaProperty = processContext.getProperty(SCHEMA).getValue();
+ String alignedProperty = processContext.getProperty(ALIGNED).getValue();
+ String maxRowNumberProperty = processContext.getProperty(MAX_ROW_NUMBER).getValue();
+
+ FlowFile flowFile = processSession.get();
+
+ String timeField = timeFieldProperty != null ? timeFieldProperty : "Time";
+ Boolean aligned = alignedProperty != null ? Boolean.valueOf(alignedProperty) : false;
+ int maxRowNumber = maxRowNumberProperty != null ? Integer.valueOf(maxRowNumberProperty) : 1024;
+
+ try (final InputStream inputStream = processSession.read(flowFile);
+ final RecordReader recordReader =
+ recordParserFactory.createRecordReader(flowFile, inputStream, getLogger())) {
+ if (flowFile == null) {
+ inputStream.close();
+ recordReader.close();
+ processSession.transfer(flowFile, REL_SUCCESS);
+ return;
+ }
+
+ HashMap<String, Tablet> tablets;
+ boolean needInitFormatter;
+ Schema schema;
+ Tuple<Boolean, String> result;
+
+ result =
+ schemaProperty != null
+ ? validateSchemaAttribute(schemaProperty)
+ : validateSchema(recordReader.getSchema());
+
+ if (!result.getKey()) {
+ getLogger().error(result.getValue());
+ inputStream.close();
+ recordReader.close();
Review Comment:
Closing these objects should be unnecessary with the use of try-with-resources.
##########
nifi-nar-bundles/nifi-iotdb-bundle/nifi-iotdb-processors/src/main/java/org/apache/nifi/processors/PutIoTDB.java:
##########
@@ -0,0 +1,286 @@
+/*
+ * 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;
+
+import java.io.InputStream;
+import java.text.SimpleDateFormat;
+import java.util.Set;
+import java.util.List;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.nifi.processors.model.Schema;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.write.record.Tablet;
+import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnUnscheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.expression.ExpressionLanguageScope;
+import org.apache.nifi.flowfile.FlowFile;
+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.serialization.RecordReader;
+import org.apache.nifi.serialization.RecordReaderFactory;
+import org.apache.nifi.serialization.record.Record;
+import org.apache.nifi.util.Tuple;
+
+@Tags({"iotdb", "insert", "tablet"})
+@InputRequirement(InputRequirement.Requirement.INPUT_REQUIRED)
+@CapabilityDescription(
+ "This is a record aware processor that reads the content of the incoming FlowFile as individual records using the "
+ + "configured 'Record Reader' and writes them to Apache IoTDB using native interface.")
+public class PutIoTDB extends AbstractIoTDB {
+
+ static final PropertyDescriptor RECORD_READER_FACTORY =
+ new PropertyDescriptor.Builder()
+ .name("Record Reader")
+ .description(
+ "Specifies the type of Record Reader controller service to use for parsing the incoming data "
+ + "and determining the schema")
+ .identifiesControllerService(RecordReaderFactory.class)
+ .required(true)
+ .build();
+
+ static final PropertyDescriptor TIME_FIELD =
+ new PropertyDescriptor.Builder()
+ .name("Time Field")
+ .description(
+ "The field name which represents time. It can be updated by expression language.")
+ .defaultValue("Time")
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .required(false)
+ .build();
+
+ static final PropertyDescriptor SCHEMA =
+ new PropertyDescriptor.Builder()
+ .name("Schema")
+ .description(
+ "The schema that IoTDB needs doesn't support good by NiFi.\n"
+ + "Therefore, you can define the schema here.\n"
+ + "Besides, you can set encoding type and compression type by this method.\n"
+ + "If you don't set this property, the inferred schema will be used.\n"
+ + "It can be updated by expression language.")
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .required(false)
+ .build();
+
+ static final PropertyDescriptor ALIGNED =
+ new PropertyDescriptor.Builder()
+ .name("Aligned")
+ .description("Whether using aligned interface? It can be updated by expression language.")
+ .allowableValues(new String[] {"true", "false"})
+ .addValidator(StandardValidators.BOOLEAN_VALIDATOR)
+ .defaultValue("false")
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .required(false)
+ .build();
+
+ static final PropertyDescriptor MAX_ROW_NUMBER =
+ new PropertyDescriptor.Builder()
+ .name("Max Row Number")
+ .description(
+ "Specifies the max row number of each tablet. It can be updated by expression language.")
+ .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
+ .defaultValue("false")
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .required(false)
+ .build();
+
+ static {
+ descriptors.add(RECORD_READER_FACTORY);
+ descriptors.add(TIME_FIELD);
+ descriptors.add(SCHEMA);
+ descriptors.add(ALIGNED);
+ descriptors.add(MAX_ROW_NUMBER);
+ }
+
+ @Override
+ protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ return descriptors;
+ }
+
+ @Override
+ public Set<Relationship> getRelationships() {
+ return relationships;
+ }
+
+ @Override
+ public void onTrigger(ProcessContext processContext, ProcessSession processSession)
+ throws ProcessException {
+ final RecordReaderFactory recordParserFactory =
+ processContext
+ .getProperty(RECORD_READER_FACTORY)
+ .asControllerService(RecordReaderFactory.class);
+
+ String timeFieldProperty = processContext.getProperty(TIME_FIELD).getValue();
+ String schemaProperty = processContext.getProperty(SCHEMA).getValue();
+ String alignedProperty = processContext.getProperty(ALIGNED).getValue();
+ String maxRowNumberProperty = processContext.getProperty(MAX_ROW_NUMBER).getValue();
+
+ FlowFile flowFile = processSession.get();
+
+ String timeField = timeFieldProperty != null ? timeFieldProperty : "Time";
+ Boolean aligned = alignedProperty != null ? Boolean.valueOf(alignedProperty) : false;
+ int maxRowNumber = maxRowNumberProperty != null ? Integer.valueOf(maxRowNumberProperty) : 1024;
+
+ try (final InputStream inputStream = processSession.read(flowFile);
+ final RecordReader recordReader =
+ recordParserFactory.createRecordReader(flowFile, inputStream, getLogger())) {
+ if (flowFile == null) {
+ inputStream.close();
+ recordReader.close();
+ processSession.transfer(flowFile, REL_SUCCESS);
+ return;
+ }
+
+ HashMap<String, Tablet> tablets;
Review Comment:
This variable, and perhaps others, should be declared and assigned in the same location for better readability.
##########
nifi-nar-bundles/nifi-iotdb-bundle/nifi-iotdb-processors/src/test/java/org/apache/nifi/processors/PutIoTDBIT.java:
##########
@@ -0,0 +1,297 @@
+/*
+ * 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;
+
+import org.apache.iotdb.db.utils.EnvironmentUtils;
+import org.apache.iotdb.rpc.IoTDBConnectionException;
+import org.apache.iotdb.rpc.StatementExecutionException;
+import org.apache.iotdb.session.Session;
+import org.apache.iotdb.session.SessionDataSet;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.read.common.Field;
+import org.apache.iotdb.tsfile.read.common.RowRecord;
+import org.apache.nifi.reporting.InitializationException;
+import org.apache.nifi.serialization.record.MockRecordParser;
+import org.apache.nifi.serialization.record.RecordFieldType;
+import org.apache.nifi.util.TestRunner;
+import org.apache.nifi.util.TestRunners;
+import org.junit.Assert;
Review Comment:
Should be changed to JUnit 5 `Assertions` and it would be better to use static imports for assert methods.
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] xuanronaldo commented on pull request #6265: NIFI-10234 Implement PutIoTDB
Posted by GitBox <gi...@apache.org>.
xuanronaldo commented on PR #6265:
URL: https://github.com/apache/nifi/pull/6265#issuecomment-1237879594
Hi, @exceptionfactory . Sorry about committing the code so late. I got lots of work to do, recently. Besides, I have updated a document about `IoTDB Schema Template` in our website as you requested. The link of it was added into the description of `Schema Template`.
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] exceptionfactory commented on pull request #6265: NIFI-10234 Implement PutIoTDB
Posted by GitBox <gi...@apache.org>.
exceptionfactory commented on PR #6265:
URL: https://github.com/apache/nifi/pull/6265#issuecomment-1240662579
> I have reverted these changes. However, several files were still marked for modification. I have no idea what to do. @exceptionfactory
I see the files are still marked as changed. If it is necessary to squash all current changes in order to remove the merge commits and avoid including these files marked as changed, that would be fine too. Either way, the latest version should just reflect the new components.
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] exceptionfactory commented on a diff in pull request #6265: NIFI-10234 Implement PutIoTDB
Posted by GitBox <gi...@apache.org>.
exceptionfactory commented on code in PR #6265:
URL: https://github.com/apache/nifi/pull/6265#discussion_r946753075
##########
nifi-nar-bundles/nifi-iotdb-bundle/nifi-iotdb-processors/src/main/java/org/apache/nifi/processors/PutIoTDB.java:
##########
@@ -0,0 +1,267 @@
+/*
+ * 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;
+
+import java.io.InputStream;
+import java.sql.Timestamp;
+import java.time.LocalDateTime;
+import java.time.format.DateTimeFormatter;
+import java.util.Set;
+import java.util.List;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.apache.nifi.processors.model.IoTDBSchema;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.write.record.Tablet;
+import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnUnscheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.expression.ExpressionLanguageScope;
+import org.apache.nifi.flowfile.FlowFile;
+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.model.ValidationResult;
+import org.apache.nifi.serialization.RecordReader;
+import org.apache.nifi.serialization.RecordReaderFactory;
+import org.apache.nifi.serialization.record.Record;
+
+@Tags({"iotdb", "insert", "tablet"})
+@InputRequirement(InputRequirement.Requirement.INPUT_REQUIRED)
+@CapabilityDescription(
+ "This is a record aware processor that reads the content of the incoming FlowFile as individual records using the "
+ + "configured 'Record Reader' and writes them to Apache IoTDB using native interface.")
+public class PutIoTDB extends AbstractIoTDB {
+
+ static final PropertyDescriptor RECORD_READER_FACTORY =
+ new PropertyDescriptor.Builder()
+ .name("Record Reader")
+ .description(
+ "Specifies the type of Record Reader controller service to use for parsing the incoming data "
+ + "and determining the schema")
+ .identifiesControllerService(RecordReaderFactory.class)
+ .required(true)
+ .build();
+
+ static final PropertyDescriptor SCHEMA =
+ new PropertyDescriptor.Builder()
+ .name("Schema")
+ .description(
+ "The schema that IoTDB needs doesn't support good by NiFi.\n"
+ + "Therefore, you can define the schema here.\n"
+ + "Besides, you can set encoding type and compression type by this method.\n"
+ + "If you don't set this property, the inferred schema will be used.\n")
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .required(false)
+ .build();
+
+ static final PropertyDescriptor ALIGNED =
+ new PropertyDescriptor.Builder()
+ .name("Aligned")
+ .description("Whether to use the Apache IoTDB Aligned Timeseries interface")
+ .allowableValues("true", "false")
+ .addValidator(StandardValidators.BOOLEAN_VALIDATOR)
+ .defaultValue("false")
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .required(false)
+ .build();
+
+ static final PropertyDescriptor MAX_ROW_NUMBER =
+ new PropertyDescriptor.Builder()
+ .name("Max Row Number")
+ .description(
+ "Specifies the max row number of each Apache IoTDB Tablet")
+ .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .required(false)
+ .build();
+
+ static {
+ descriptors.add(RECORD_READER_FACTORY);
+ descriptors.add(SCHEMA);
+ descriptors.add(ALIGNED);
+ descriptors.add(MAX_ROW_NUMBER);
+ }
+
+ @Override
+ protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ return descriptors;
+ }
+
+ @Override
+ public Set<Relationship> getRelationships() {
+ return relationships;
+ }
+
+ @Override
+ public void onTrigger(ProcessContext processContext, ProcessSession processSession)
+ throws ProcessException {
+ final RecordReaderFactory recordParserFactory =
+ processContext
+ .getProperty(RECORD_READER_FACTORY)
+ .asControllerService(RecordReaderFactory.class);
+
+ FlowFile flowFile = processSession.get();
+
+ if (flowFile == null) {
+ processSession.transfer(flowFile, REL_SUCCESS);
+ return;
+ }
+
+ String schemaProperty = processContext.getProperty(SCHEMA).evaluateAttributeExpressions(flowFile).getValue();
+ String alignedProperty = processContext.getProperty(ALIGNED).evaluateAttributeExpressions(flowFile).getValue();
+ String maxRowNumberProperty = processContext.getProperty(MAX_ROW_NUMBER).evaluateAttributeExpressions(flowFile).getValue();
+
+ final boolean aligned = alignedProperty != null ? Boolean.valueOf(alignedProperty) : false;
+ int maxRowNumber = maxRowNumberProperty != null ? Integer.valueOf(maxRowNumberProperty) : 1024;
+
+ try (final InputStream inputStream = processSession.read(flowFile);
+ final RecordReader recordReader =
+ recordParserFactory.createRecordReader(flowFile, inputStream, getLogger())) {
+ boolean needInitFormatter;
+ IoTDBSchema schema;
+ ValidationResult result;
+
+ result =
+ schemaProperty != null
+ ? validateSchemaAttribute(schemaProperty)
+ : validateSchema(recordReader.getSchema());
+
+ if (!result.getKey()) {
+ getLogger().error(String.format("The property `schema` has an error: %s", result.getValue()));
+ inputStream.close();
+ recordReader.close();
+ processSession.transfer(flowFile, REL_FAILURE);
+ return;
+ } else {
+ if (result.getValue() != null) {
+ getLogger().warn(String.format("The property `schema` has a warn: %s", result.getValue()));
+ }
+ }
+
+ schema =
+ schemaProperty != null
+ ? mapper.readValue(schemaProperty, IoTDBSchema.class)
+ : convertSchema(recordReader.getSchema());
+
+ List<String> fieldNames = schema.getFieldNames();
+
+ needInitFormatter = schema.getTimeType() != IoTDBSchema.TimeType.LONG;
+
+ HashMap<String, Tablet> tablets = generateTablets(schema, maxRowNumber);
+ DateTimeFormatter format = null;
+
+ Record record;
+
+ while ((record = recordReader.nextRecord()) != null) {
+ Object[] values = record.getValues();
+ if (format == null && needInitFormatter) {
+ format = initFormatter((String) values[0]);
+ if (format == null) {
+ getLogger().error("{} Record [{}] time format not supported\", flowFile, recordNumber");
Review Comment:
The trailing slash character should be removed:
```suggestion
getLogger().error("{} Record [{}] time format not supported", flowFile, recordNumber");
```
##########
nifi-nar-bundles/nifi-iotdb-bundle/nifi-iotdb-processors/src/main/java/org/apache/nifi/processors/PutIoTDB.java:
##########
@@ -0,0 +1,267 @@
+/*
+ * 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;
+
+import java.io.InputStream;
+import java.sql.Timestamp;
+import java.time.LocalDateTime;
+import java.time.format.DateTimeFormatter;
+import java.util.Set;
+import java.util.List;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.apache.nifi.processors.model.IoTDBSchema;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.write.record.Tablet;
+import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnUnscheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.expression.ExpressionLanguageScope;
+import org.apache.nifi.flowfile.FlowFile;
+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.model.ValidationResult;
+import org.apache.nifi.serialization.RecordReader;
+import org.apache.nifi.serialization.RecordReaderFactory;
+import org.apache.nifi.serialization.record.Record;
+
+@Tags({"iotdb", "insert", "tablet"})
+@InputRequirement(InputRequirement.Requirement.INPUT_REQUIRED)
+@CapabilityDescription(
+ "This is a record aware processor that reads the content of the incoming FlowFile as individual records using the "
+ + "configured 'Record Reader' and writes them to Apache IoTDB using native interface.")
+public class PutIoTDB extends AbstractIoTDB {
+
+ static final PropertyDescriptor RECORD_READER_FACTORY =
+ new PropertyDescriptor.Builder()
+ .name("Record Reader")
+ .description(
+ "Specifies the type of Record Reader controller service to use for parsing the incoming data "
+ + "and determining the schema")
+ .identifiesControllerService(RecordReaderFactory.class)
+ .required(true)
+ .build();
+
+ static final PropertyDescriptor SCHEMA =
+ new PropertyDescriptor.Builder()
+ .name("Schema")
+ .description(
+ "The schema that IoTDB needs doesn't support good by NiFi.\n"
+ + "Therefore, you can define the schema here.\n"
+ + "Besides, you can set encoding type and compression type by this method.\n"
+ + "If you don't set this property, the inferred schema will be used.\n")
Review Comment:
The wording of this description is not quite clear, and does not need to state what NiFi does not support. Recommend adjusting the wording to include more details about the expected format, along the following lines:
The Apache IoTDB Schema Template defined using JSON. The Processor will infer the IoTDB Schema when this property is not configured.
The [IoTDB Schema Template](https://iotdb.apache.org/UserGuide/Master/Data-Concept/Schema-Template.html) documentation provides some details, but is their an official format specification? It would be helpful to link to that documentation if available, otherwise it will be unclear how to define this property.
##########
nifi-nar-bundles/nifi-iotdb-bundle/nifi-iotdb-processors/pom.xml:
##########
@@ -0,0 +1,102 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ 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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <parent>
+ <artifactId>nifi-iotdb-bundle</artifactId>
+ <groupId>org.apache.nifi</groupId>
+ <version>1.18.0-SNAPSHOT</version>
+ </parent>
+ <modelVersion>4.0.0</modelVersion>
+
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-iotdb-processors</artifactId>
+ <packaging>jar</packaging>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.iotdb</groupId>
+ <artifactId>iotdb-session</artifactId>
+ <version>${iotdb.sdk.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.iotdb</groupId>
+ <artifactId>iotdb-server</artifactId>
+ <version>${iotdb.sdk.version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.iotdb</groupId>
+ <artifactId>iotdb-server</artifactId>
+ <version>${iotdb.sdk.version}</version>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-api</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-mock</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-record-serialization-service-api</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-record</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-mock-record-utils</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-dbcp-service-api</artifactId>
+ <version>${project.version}</version>
+ </dependency>
Review Comment:
This dependency is not necessary and should be removed
##########
nifi-nar-bundles/nifi-iotdb-bundle/nifi-iotdb-processors/src/main/java/org/apache/nifi/processors/PutIoTDB.java:
##########
@@ -0,0 +1,267 @@
+/*
+ * 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;
+
+import java.io.InputStream;
+import java.sql.Timestamp;
+import java.time.LocalDateTime;
+import java.time.format.DateTimeFormatter;
+import java.util.Set;
+import java.util.List;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.apache.nifi.processors.model.IoTDBSchema;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.write.record.Tablet;
+import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnUnscheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.expression.ExpressionLanguageScope;
+import org.apache.nifi.flowfile.FlowFile;
+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.model.ValidationResult;
+import org.apache.nifi.serialization.RecordReader;
+import org.apache.nifi.serialization.RecordReaderFactory;
+import org.apache.nifi.serialization.record.Record;
+
+@Tags({"iotdb", "insert", "tablet"})
+@InputRequirement(InputRequirement.Requirement.INPUT_REQUIRED)
+@CapabilityDescription(
+ "This is a record aware processor that reads the content of the incoming FlowFile as individual records using the "
+ + "configured 'Record Reader' and writes them to Apache IoTDB using native interface.")
+public class PutIoTDB extends AbstractIoTDB {
+
+ static final PropertyDescriptor RECORD_READER_FACTORY =
+ new PropertyDescriptor.Builder()
+ .name("Record Reader")
+ .description(
+ "Specifies the type of Record Reader controller service to use for parsing the incoming data "
+ + "and determining the schema")
+ .identifiesControllerService(RecordReaderFactory.class)
+ .required(true)
+ .build();
+
+ static final PropertyDescriptor SCHEMA =
+ new PropertyDescriptor.Builder()
+ .name("Schema")
Review Comment:
Recommend naming this property `Schema Template` to align with the IoTDB terminology.
##########
nifi-nar-bundles/nifi-iotdb-bundle/nifi-iotdb-processors/src/main/java/org/apache/nifi/processors/PutIoTDB.java:
##########
@@ -0,0 +1,267 @@
+/*
+ * 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;
+
+import java.io.InputStream;
+import java.sql.Timestamp;
+import java.time.LocalDateTime;
+import java.time.format.DateTimeFormatter;
+import java.util.Set;
+import java.util.List;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.apache.nifi.processors.model.IoTDBSchema;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.write.record.Tablet;
+import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnUnscheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.expression.ExpressionLanguageScope;
+import org.apache.nifi.flowfile.FlowFile;
+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.model.ValidationResult;
+import org.apache.nifi.serialization.RecordReader;
+import org.apache.nifi.serialization.RecordReaderFactory;
+import org.apache.nifi.serialization.record.Record;
+
+@Tags({"iotdb", "insert", "tablet"})
+@InputRequirement(InputRequirement.Requirement.INPUT_REQUIRED)
+@CapabilityDescription(
+ "This is a record aware processor that reads the content of the incoming FlowFile as individual records using the "
+ + "configured 'Record Reader' and writes them to Apache IoTDB using native interface.")
+public class PutIoTDB extends AbstractIoTDB {
+
+ static final PropertyDescriptor RECORD_READER_FACTORY =
+ new PropertyDescriptor.Builder()
+ .name("Record Reader")
+ .description(
+ "Specifies the type of Record Reader controller service to use for parsing the incoming data "
+ + "and determining the schema")
+ .identifiesControllerService(RecordReaderFactory.class)
+ .required(true)
+ .build();
+
+ static final PropertyDescriptor SCHEMA =
+ new PropertyDescriptor.Builder()
+ .name("Schema")
+ .description(
+ "The schema that IoTDB needs doesn't support good by NiFi.\n"
+ + "Therefore, you can define the schema here.\n"
+ + "Besides, you can set encoding type and compression type by this method.\n"
+ + "If you don't set this property, the inferred schema will be used.\n")
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .required(false)
+ .build();
+
+ static final PropertyDescriptor ALIGNED =
+ new PropertyDescriptor.Builder()
+ .name("Aligned")
+ .description("Whether to use the Apache IoTDB Aligned Timeseries interface")
+ .allowableValues("true", "false")
+ .addValidator(StandardValidators.BOOLEAN_VALIDATOR)
+ .defaultValue("false")
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .required(false)
+ .build();
+
+ static final PropertyDescriptor MAX_ROW_NUMBER =
+ new PropertyDescriptor.Builder()
+ .name("Max Row Number")
+ .description(
+ "Specifies the max row number of each Apache IoTDB Tablet")
+ .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .required(false)
+ .build();
+
+ static {
+ descriptors.add(RECORD_READER_FACTORY);
+ descriptors.add(SCHEMA);
+ descriptors.add(ALIGNED);
+ descriptors.add(MAX_ROW_NUMBER);
+ }
+
+ @Override
+ protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ return descriptors;
+ }
+
+ @Override
+ public Set<Relationship> getRelationships() {
+ return relationships;
+ }
+
+ @Override
+ public void onTrigger(ProcessContext processContext, ProcessSession processSession)
+ throws ProcessException {
+ final RecordReaderFactory recordParserFactory =
+ processContext
+ .getProperty(RECORD_READER_FACTORY)
+ .asControllerService(RecordReaderFactory.class);
+
+ FlowFile flowFile = processSession.get();
+
+ if (flowFile == null) {
+ processSession.transfer(flowFile, REL_SUCCESS);
Review Comment:
This transfer should be removed because `flowFile` is null.
##########
nifi-nar-bundles/nifi-iotdb-bundle/nifi-iotdb-processors/src/main/java/org/apache/nifi/processors/PutIoTDB.java:
##########
@@ -0,0 +1,267 @@
+/*
+ * 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;
+
+import java.io.InputStream;
+import java.sql.Timestamp;
+import java.time.LocalDateTime;
+import java.time.format.DateTimeFormatter;
+import java.util.Set;
+import java.util.List;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.apache.nifi.processors.model.IoTDBSchema;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.write.record.Tablet;
+import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnUnscheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.expression.ExpressionLanguageScope;
+import org.apache.nifi.flowfile.FlowFile;
+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.model.ValidationResult;
+import org.apache.nifi.serialization.RecordReader;
+import org.apache.nifi.serialization.RecordReaderFactory;
+import org.apache.nifi.serialization.record.Record;
+
+@Tags({"iotdb", "insert", "tablet"})
+@InputRequirement(InputRequirement.Requirement.INPUT_REQUIRED)
+@CapabilityDescription(
+ "This is a record aware processor that reads the content of the incoming FlowFile as individual records using the "
+ + "configured 'Record Reader' and writes them to Apache IoTDB using native interface.")
+public class PutIoTDB extends AbstractIoTDB {
+
+ static final PropertyDescriptor RECORD_READER_FACTORY =
+ new PropertyDescriptor.Builder()
+ .name("Record Reader")
+ .description(
+ "Specifies the type of Record Reader controller service to use for parsing the incoming data "
+ + "and determining the schema")
+ .identifiesControllerService(RecordReaderFactory.class)
+ .required(true)
+ .build();
+
+ static final PropertyDescriptor SCHEMA =
+ new PropertyDescriptor.Builder()
+ .name("Schema")
+ .description(
+ "The schema that IoTDB needs doesn't support good by NiFi.\n"
+ + "Therefore, you can define the schema here.\n"
+ + "Besides, you can set encoding type and compression type by this method.\n"
+ + "If you don't set this property, the inferred schema will be used.\n")
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .required(false)
+ .build();
+
+ static final PropertyDescriptor ALIGNED =
+ new PropertyDescriptor.Builder()
+ .name("Aligned")
+ .description("Whether to use the Apache IoTDB Aligned Timeseries interface")
+ .allowableValues("true", "false")
+ .addValidator(StandardValidators.BOOLEAN_VALIDATOR)
+ .defaultValue("false")
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .required(false)
+ .build();
+
+ static final PropertyDescriptor MAX_ROW_NUMBER =
+ new PropertyDescriptor.Builder()
+ .name("Max Row Number")
+ .description(
+ "Specifies the max row number of each Apache IoTDB Tablet")
+ .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .required(false)
+ .build();
+
+ static {
+ descriptors.add(RECORD_READER_FACTORY);
+ descriptors.add(SCHEMA);
+ descriptors.add(ALIGNED);
+ descriptors.add(MAX_ROW_NUMBER);
+ }
+
+ @Override
+ protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ return descriptors;
+ }
+
+ @Override
+ public Set<Relationship> getRelationships() {
+ return relationships;
+ }
+
+ @Override
+ public void onTrigger(ProcessContext processContext, ProcessSession processSession)
+ throws ProcessException {
+ final RecordReaderFactory recordParserFactory =
+ processContext
+ .getProperty(RECORD_READER_FACTORY)
+ .asControllerService(RecordReaderFactory.class);
+
+ FlowFile flowFile = processSession.get();
+
+ if (flowFile == null) {
+ processSession.transfer(flowFile, REL_SUCCESS);
+ return;
+ }
+
+ String schemaProperty = processContext.getProperty(SCHEMA).evaluateAttributeExpressions(flowFile).getValue();
+ String alignedProperty = processContext.getProperty(ALIGNED).evaluateAttributeExpressions(flowFile).getValue();
+ String maxRowNumberProperty = processContext.getProperty(MAX_ROW_NUMBER).evaluateAttributeExpressions(flowFile).getValue();
+
+ final boolean aligned = alignedProperty != null ? Boolean.valueOf(alignedProperty) : false;
+ int maxRowNumber = maxRowNumberProperty != null ? Integer.valueOf(maxRowNumberProperty) : 1024;
+
+ try (final InputStream inputStream = processSession.read(flowFile);
+ final RecordReader recordReader =
+ recordParserFactory.createRecordReader(flowFile, inputStream, getLogger())) {
+ boolean needInitFormatter;
+ IoTDBSchema schema;
+ ValidationResult result;
+
+ result =
+ schemaProperty != null
+ ? validateSchemaAttribute(schemaProperty)
+ : validateSchema(recordReader.getSchema());
+
+ if (!result.getKey()) {
+ getLogger().error(String.format("The property `schema` has an error: %s", result.getValue()));
+ inputStream.close();
+ recordReader.close();
+ processSession.transfer(flowFile, REL_FAILURE);
+ return;
+ } else {
+ if (result.getValue() != null) {
+ getLogger().warn(String.format("The property `schema` has a warn: %s", result.getValue()));
+ }
+ }
+
+ schema =
+ schemaProperty != null
+ ? mapper.readValue(schemaProperty, IoTDBSchema.class)
+ : convertSchema(recordReader.getSchema());
+
+ List<String> fieldNames = schema.getFieldNames();
+
+ needInitFormatter = schema.getTimeType() != IoTDBSchema.TimeType.LONG;
+
+ HashMap<String, Tablet> tablets = generateTablets(schema, maxRowNumber);
+ DateTimeFormatter format = null;
+
+ Record record;
+
+ while ((record = recordReader.nextRecord()) != null) {
+ Object[] values = record.getValues();
+ if (format == null && needInitFormatter) {
+ format = initFormatter((String) values[0]);
+ if (format == null) {
+ getLogger().error("{} Record [{}] time format not supported\", flowFile, recordNumber");
+ inputStream.close();
+ recordReader.close();
+ processSession.transfer(flowFile, REL_FAILURE);
+ return;
+ }
+ }
+
+ long timestamp;
+ if (needInitFormatter) {
+ timestamp = Timestamp.valueOf(LocalDateTime.parse((String) values[0], format)).getTime();
Review Comment:
Recommend breaking this into multiple lines for better readability.
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] xuanronaldo commented on a diff in pull request #6265: NIFI-10234 Implement PutIoTDB
Posted by GitBox <gi...@apache.org>.
xuanronaldo commented on code in PR #6265:
URL: https://github.com/apache/nifi/pull/6265#discussion_r940939151
##########
nifi-nar-bundles/nifi-iotdb-bundle/nifi-iotdb-processors/src/main/java/org/apache/nifi/processors/PutIoTDB.java:
##########
@@ -0,0 +1,286 @@
+/*
+ * 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;
+
+import java.io.InputStream;
+import java.text.SimpleDateFormat;
+import java.util.Set;
+import java.util.List;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.nifi.processors.model.Schema;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.write.record.Tablet;
+import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnUnscheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.expression.ExpressionLanguageScope;
+import org.apache.nifi.flowfile.FlowFile;
+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.serialization.RecordReader;
+import org.apache.nifi.serialization.RecordReaderFactory;
+import org.apache.nifi.serialization.record.Record;
+import org.apache.nifi.util.Tuple;
+
+@Tags({"iotdb", "insert", "tablet"})
+@InputRequirement(InputRequirement.Requirement.INPUT_REQUIRED)
+@CapabilityDescription(
+ "This is a record aware processor that reads the content of the incoming FlowFile as individual records using the "
+ + "configured 'Record Reader' and writes them to Apache IoTDB using native interface.")
+public class PutIoTDB extends AbstractIoTDB {
+
+ static final PropertyDescriptor RECORD_READER_FACTORY =
+ new PropertyDescriptor.Builder()
+ .name("Record Reader")
+ .description(
+ "Specifies the type of Record Reader controller service to use for parsing the incoming data "
+ + "and determining the schema")
+ .identifiesControllerService(RecordReaderFactory.class)
+ .required(true)
+ .build();
+
+ static final PropertyDescriptor TIME_FIELD =
+ new PropertyDescriptor.Builder()
+ .name("Time Field")
+ .description(
+ "The field name which represents time. It can be updated by expression language.")
+ .defaultValue("Time")
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .required(false)
+ .build();
+
+ static final PropertyDescriptor SCHEMA =
+ new PropertyDescriptor.Builder()
+ .name("Schema")
+ .description(
+ "The schema that IoTDB needs doesn't support good by NiFi.\n"
+ + "Therefore, you can define the schema here.\n"
+ + "Besides, you can set encoding type and compression type by this method.\n"
+ + "If you don't set this property, the inferred schema will be used.\n"
+ + "It can be updated by expression language.")
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .required(false)
+ .build();
+
+ static final PropertyDescriptor ALIGNED =
+ new PropertyDescriptor.Builder()
+ .name("Aligned")
+ .description("Whether using aligned interface? It can be updated by expression language.")
+ .allowableValues(new String[] {"true", "false"})
+ .addValidator(StandardValidators.BOOLEAN_VALIDATOR)
+ .defaultValue("false")
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .required(false)
+ .build();
+
+ static final PropertyDescriptor MAX_ROW_NUMBER =
+ new PropertyDescriptor.Builder()
+ .name("Max Row Number")
+ .description(
+ "Specifies the max row number of each tablet. It can be updated by expression language.")
+ .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
+ .defaultValue("false")
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .required(false)
+ .build();
+
+ static {
+ descriptors.add(RECORD_READER_FACTORY);
+ descriptors.add(TIME_FIELD);
+ descriptors.add(SCHEMA);
+ descriptors.add(ALIGNED);
+ descriptors.add(MAX_ROW_NUMBER);
+ }
+
+ @Override
+ protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ return descriptors;
+ }
+
+ @Override
+ public Set<Relationship> getRelationships() {
+ return relationships;
+ }
+
+ @Override
+ public void onTrigger(ProcessContext processContext, ProcessSession processSession)
+ throws ProcessException {
+ final RecordReaderFactory recordParserFactory =
+ processContext
+ .getProperty(RECORD_READER_FACTORY)
+ .asControllerService(RecordReaderFactory.class);
+
+ String timeFieldProperty = processContext.getProperty(TIME_FIELD).getValue();
+ String schemaProperty = processContext.getProperty(SCHEMA).getValue();
+ String alignedProperty = processContext.getProperty(ALIGNED).getValue();
+ String maxRowNumberProperty = processContext.getProperty(MAX_ROW_NUMBER).getValue();
+
+ FlowFile flowFile = processSession.get();
+
+ String timeField = timeFieldProperty != null ? timeFieldProperty : "Time";
+ Boolean aligned = alignedProperty != null ? Boolean.valueOf(alignedProperty) : false;
+ int maxRowNumber = maxRowNumberProperty != null ? Integer.valueOf(maxRowNumberProperty) : 1024;
+
+ try (final InputStream inputStream = processSession.read(flowFile);
+ final RecordReader recordReader =
+ recordParserFactory.createRecordReader(flowFile, inputStream, getLogger())) {
+ if (flowFile == null) {
+ inputStream.close();
+ recordReader.close();
+ processSession.transfer(flowFile, REL_SUCCESS);
+ return;
+ }
+
+ HashMap<String, Tablet> tablets;
+ boolean needInitFormatter;
+ Schema schema;
+ Tuple<Boolean, String> result;
+
+ result =
+ schemaProperty != null
+ ? validateSchemaAttribute(schemaProperty)
+ : validateSchema(recordReader.getSchema());
+
+ if (!result.getKey()) {
+ getLogger().error(result.getValue());
+ inputStream.close();
+ recordReader.close();
Review Comment:
I have tried to delete these, but it will throw an exception `java.lang.IllegalStateException: FlowFile[0,16972544312800.mockFlowFile,0B] already in use for an active callback or InputStream created by ProcessSession.read(FlowFile) has not been closed`.
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] xuanronaldo commented on pull request #6265: NIFI-10234 Implement PutIoTDB
Posted by GitBox <gi...@apache.org>.
xuanronaldo commented on PR #6265:
URL: https://github.com/apache/nifi/pull/6265#issuecomment-1244990982
> > I have reverted these changes. However, several files were still marked for modification. I have no idea what to do. @exceptionfactory
>
> I see the files are still marked as changed. If it is necessary to squash all current changes in order to remove the merge commits and avoid including these files marked as changed, that would be fine too. Either way, the latest version should just reflect the new components.
I have no idea about squash all current changes. Or I can create a new branch and submit another PR?
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] exceptionfactory commented on pull request #6265: NIFI-10234 Implement PutIoTDB
Posted by GitBox <gi...@apache.org>.
exceptionfactory commented on PR #6265:
URL: https://github.com/apache/nifi/pull/6265#issuecomment-1202928805
One additional note, Apache NiFi just released version 1.17.0, so the current main branch is now on 1.18.0-SNAPSHOT. Please rebase and update to the current version.
Thanks!
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] exceptionfactory commented on a diff in pull request #6265: NIFI-10234 Implement PutIoTDB
Posted by GitBox <gi...@apache.org>.
exceptionfactory commented on code in PR #6265:
URL: https://github.com/apache/nifi/pull/6265#discussion_r946762657
##########
nifi-nar-bundles/nifi-iotdb-bundle/nifi-iotdb-processors/src/main/java/org/apache/nifi/processors/PutIoTDB.java:
##########
@@ -0,0 +1,267 @@
+/*
+ * 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;
+
+import java.io.InputStream;
+import java.sql.Timestamp;
+import java.time.LocalDateTime;
+import java.time.format.DateTimeFormatter;
+import java.util.Set;
+import java.util.List;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.apache.nifi.processors.model.IoTDBSchema;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.write.record.Tablet;
+import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnUnscheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.expression.ExpressionLanguageScope;
+import org.apache.nifi.flowfile.FlowFile;
+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.model.ValidationResult;
+import org.apache.nifi.serialization.RecordReader;
+import org.apache.nifi.serialization.RecordReaderFactory;
+import org.apache.nifi.serialization.record.Record;
+
+@Tags({"iotdb", "insert", "tablet"})
+@InputRequirement(InputRequirement.Requirement.INPUT_REQUIRED)
+@CapabilityDescription(
+ "This is a record aware processor that reads the content of the incoming FlowFile as individual records using the "
+ + "configured 'Record Reader' and writes them to Apache IoTDB using native interface.")
+public class PutIoTDB extends AbstractIoTDB {
+
+ static final PropertyDescriptor RECORD_READER_FACTORY =
+ new PropertyDescriptor.Builder()
+ .name("Record Reader")
+ .description(
+ "Specifies the type of Record Reader controller service to use for parsing the incoming data "
+ + "and determining the schema")
+ .identifiesControllerService(RecordReaderFactory.class)
+ .required(true)
+ .build();
+
+ static final PropertyDescriptor SCHEMA =
+ new PropertyDescriptor.Builder()
+ .name("Schema")
+ .description(
+ "The schema that IoTDB needs doesn't support good by NiFi.\n"
+ + "Therefore, you can define the schema here.\n"
+ + "Besides, you can set encoding type and compression type by this method.\n"
+ + "If you don't set this property, the inferred schema will be used.\n")
Review Comment:
The wording of this description is not quite clear, and does not need to state what NiFi does not support. Recommend adjusting the wording to include more details about the expected format, along the following lines:
The Apache IoTDB Schema Template defined using JSON. The Processor will infer the IoTDB Schema when this property is not configured.
The [IoTDB Schema Template](https://iotdb.apache.org/UserGuide/Master/Data-Concept/Schema-Template.html) documentation provides some details, but is there an official format specification? It would be helpful to link to that documentation if available, otherwise it will be unclear how to define this property.
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] exceptionfactory commented on a diff in pull request #6265: NIFI-10234 Implement PutIoTDB
Posted by GitBox <gi...@apache.org>.
exceptionfactory commented on code in PR #6265:
URL: https://github.com/apache/nifi/pull/6265#discussion_r941316372
##########
nifi-nar-bundles/nifi-iotdb-bundle/nifi-iotdb-processors/src/main/java/org/apache/nifi/processors/PutIoTDB.java:
##########
@@ -0,0 +1,286 @@
+/*
+ * 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;
+
+import java.io.InputStream;
+import java.text.SimpleDateFormat;
+import java.util.Set;
+import java.util.List;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.nifi.processors.model.Schema;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.write.record.Tablet;
+import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnUnscheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.expression.ExpressionLanguageScope;
+import org.apache.nifi.flowfile.FlowFile;
+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.serialization.RecordReader;
+import org.apache.nifi.serialization.RecordReaderFactory;
+import org.apache.nifi.serialization.record.Record;
+import org.apache.nifi.util.Tuple;
+
+@Tags({"iotdb", "insert", "tablet"})
+@InputRequirement(InputRequirement.Requirement.INPUT_REQUIRED)
+@CapabilityDescription(
+ "This is a record aware processor that reads the content of the incoming FlowFile as individual records using the "
+ + "configured 'Record Reader' and writes them to Apache IoTDB using native interface.")
+public class PutIoTDB extends AbstractIoTDB {
+
+ static final PropertyDescriptor RECORD_READER_FACTORY =
+ new PropertyDescriptor.Builder()
+ .name("Record Reader")
+ .description(
+ "Specifies the type of Record Reader controller service to use for parsing the incoming data "
+ + "and determining the schema")
+ .identifiesControllerService(RecordReaderFactory.class)
+ .required(true)
+ .build();
+
+ static final PropertyDescriptor TIME_FIELD =
+ new PropertyDescriptor.Builder()
+ .name("Time Field")
+ .description(
+ "The field name which represents time. It can be updated by expression language.")
+ .defaultValue("Time")
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .required(false)
+ .build();
+
+ static final PropertyDescriptor SCHEMA =
+ new PropertyDescriptor.Builder()
+ .name("Schema")
+ .description(
+ "The schema that IoTDB needs doesn't support good by NiFi.\n"
+ + "Therefore, you can define the schema here.\n"
+ + "Besides, you can set encoding type and compression type by this method.\n"
+ + "If you don't set this property, the inferred schema will be used.\n"
+ + "It can be updated by expression language.")
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .required(false)
+ .build();
+
+ static final PropertyDescriptor ALIGNED =
+ new PropertyDescriptor.Builder()
+ .name("Aligned")
+ .description("Whether using aligned interface? It can be updated by expression language.")
+ .allowableValues(new String[] {"true", "false"})
+ .addValidator(StandardValidators.BOOLEAN_VALIDATOR)
+ .defaultValue("false")
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .required(false)
+ .build();
+
+ static final PropertyDescriptor MAX_ROW_NUMBER =
+ new PropertyDescriptor.Builder()
+ .name("Max Row Number")
+ .description(
+ "Specifies the max row number of each tablet. It can be updated by expression language.")
+ .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
+ .defaultValue("false")
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .required(false)
+ .build();
+
+ static {
+ descriptors.add(RECORD_READER_FACTORY);
+ descriptors.add(TIME_FIELD);
+ descriptors.add(SCHEMA);
+ descriptors.add(ALIGNED);
+ descriptors.add(MAX_ROW_NUMBER);
+ }
+
+ @Override
+ protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ return descriptors;
+ }
+
+ @Override
+ public Set<Relationship> getRelationships() {
+ return relationships;
+ }
+
+ @Override
+ public void onTrigger(ProcessContext processContext, ProcessSession processSession)
+ throws ProcessException {
+ final RecordReaderFactory recordParserFactory =
+ processContext
+ .getProperty(RECORD_READER_FACTORY)
+ .asControllerService(RecordReaderFactory.class);
+
+ String timeFieldProperty = processContext.getProperty(TIME_FIELD).getValue();
+ String schemaProperty = processContext.getProperty(SCHEMA).getValue();
+ String alignedProperty = processContext.getProperty(ALIGNED).getValue();
+ String maxRowNumberProperty = processContext.getProperty(MAX_ROW_NUMBER).getValue();
+
+ FlowFile flowFile = processSession.get();
+
+ String timeField = timeFieldProperty != null ? timeFieldProperty : "Time";
+ Boolean aligned = alignedProperty != null ? Boolean.valueOf(alignedProperty) : false;
+ int maxRowNumber = maxRowNumberProperty != null ? Integer.valueOf(maxRowNumberProperty) : 1024;
+
+ try (final InputStream inputStream = processSession.read(flowFile);
+ final RecordReader recordReader =
+ recordParserFactory.createRecordReader(flowFile, inputStream, getLogger())) {
+ if (flowFile == null) {
+ inputStream.close();
+ recordReader.close();
+ processSession.transfer(flowFile, REL_SUCCESS);
+ return;
+ }
+
+ HashMap<String, Tablet> tablets;
+ boolean needInitFormatter;
+ Schema schema;
+ Tuple<Boolean, String> result;
+
+ result =
+ schemaProperty != null
+ ? validateSchemaAttribute(schemaProperty)
+ : validateSchema(recordReader.getSchema());
+
+ if (!result.getKey()) {
+ getLogger().error(result.getValue());
+ inputStream.close();
+ recordReader.close();
Review Comment:
Thanks for the reply @xuanronaldo, does this occur when calling `ProcessSession.transfer()`? If you move it outside of the try-catch block, or change the `read()` signature to use a callback, that should resolve this error.
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] exceptionfactory commented on a diff in pull request #6265: NIFI-10234 Implement PutIoTDB
Posted by GitBox <gi...@apache.org>.
exceptionfactory commented on code in PR #6265:
URL: https://github.com/apache/nifi/pull/6265#discussion_r946167414
##########
nifi-nar-bundles/nifi-iotdb-bundle/nifi-iotdb-processors/src/test/java/org/apache/nifi/processors/PutIoTDBUT.java:
##########
@@ -0,0 +1,85 @@
+/*
+ * 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;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.nifi.processors.model.IoTDBSchema;
+import org.apache.iotdb.tsfile.file.metadata.enums.CompressionType;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSEncoding;
+import org.junit.Assert;
Review Comment:
See the other note about changing to use JUnit 5 `Assertions`.
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] xuanronaldo commented on pull request #6265: NIFI-10234 Implement PutIoTDB
Posted by GitBox <gi...@apache.org>.
xuanronaldo commented on PR #6265:
URL: https://github.com/apache/nifi/pull/6265#issuecomment-1214668955
All checks have passed, could you please take a look at my latest commit? @exceptionfactory
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] xuanronaldo commented on pull request #6265: NIFI-10234 Implement PutIoTDB
Posted by GitBox <gi...@apache.org>.
xuanronaldo commented on PR #6265:
URL: https://github.com/apache/nifi/pull/6265#issuecomment-1238886451
Yep, you are rigth @exceptionfactory . The `template` in IoTDB is a structure to reduce the memory usage. However, the `Schema Template` here is just an adapter to define the schema in IoTDB. If you want to create schema by native way, the DDL or coding is necessary. Therefore, I designed the `Schema Template` to make it.
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] xuanronaldo closed pull request #6265: NIFI-10234 Implement PutIoTDB
Posted by GitBox <gi...@apache.org>.
xuanronaldo closed pull request #6265: NIFI-10234 Implement PutIoTDB
URL: https://github.com/apache/nifi/pull/6265
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org