You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by ed...@apache.org on 2019/01/09 04:33:36 UTC
[nifi] branch master updated: NIFI-5937 use processor-configured
encoding instead of the system default
This is an automated email from the ASF dual-hosted git repository.
edbe pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/nifi.git
The following commit(s) were added to refs/heads/master by this push:
new 1a443c7 NIFI-5937 use processor-configured encoding instead of the system default
1a443c7 is described below
commit 1a443c73ecb0ed00233b66d282031aa5b37afb24
Author: Alex Savitsky <al...@scotiabank.com>
AuthorDate: Tue Jan 8 10:15:45 2019 -0500
NIFI-5937 use processor-configured encoding instead of the system default
NIFI-5937 added tests to verify that accented characters are preserved correctly
NIFI-5937 unfolding starred imports
NIFI-5937 unfolding starred imports (now with statics)
Signed-off-by: Ed <ed...@gmail.com>
This closes #3250
---
.../elasticsearch/PutElasticsearchHttpRecord.java | 5 +-
.../TestPutElasticsearchHttpRecord.java | 94 ++++++++++++++++------
2 files changed, 73 insertions(+), 26 deletions(-)
diff --git a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/main/java/org/apache/nifi/processors/elasticsearch/PutElasticsearchHttpRecord.java b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/main/java/org/apache/nifi/processors/elasticsearch/PutElasticsearchHttpRecord.java
index ac36604..52de424 100644
--- a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/main/java/org/apache/nifi/processors/elasticsearch/PutElasticsearchHttpRecord.java
+++ b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/main/java/org/apache/nifi/processors/elasticsearch/PutElasticsearchHttpRecord.java
@@ -73,6 +73,7 @@ import java.io.IOException;
import java.io.InputStream;
import java.math.BigInteger;
import java.net.URL;
+import java.nio.charset.Charset;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
@@ -198,6 +199,7 @@ public class PutElasticsearchHttpRecord extends AbstractElasticsearchHttpProcess
descriptors.add(ID_RECORD_PATH);
descriptors.add(INDEX);
descriptors.add(TYPE);
+ descriptors.add(CHARSET);
descriptors.add(INDEX_OP);
descriptors.add(SUPPRESS_NULLS);
@@ -313,6 +315,7 @@ public class PutElasticsearchHttpRecord extends AbstractElasticsearchHttpProcess
final String id_path = context.getProperty(ID_RECORD_PATH).evaluateAttributeExpressions(flowFile).getValue();
final RecordPath recordPath = StringUtils.isEmpty(id_path) ? null : recordPathCache.getCompiled(id_path);
final StringBuilder sb = new StringBuilder();
+ final Charset charset = Charset.forName(context.getProperty(CHARSET).evaluateAttributeExpressions(flowFile).getValue());
int recordCount = 0;
try (final InputStream in = session.read(flowFile);
@@ -345,7 +348,7 @@ public class PutElasticsearchHttpRecord extends AbstractElasticsearchHttpProcess
writeRecord(record, record.getSchema(), generator);
generator.flush();
generator.close();
- json.append(out.toString());
+ json.append(out.toString(charset.name()));
buildBulkCommand(sb, index, docType, indexOp, id, json.toString());
recordCount++;
diff --git a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/test/java/org/apache/nifi/processors/elasticsearch/TestPutElasticsearchHttpRecord.java b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/test/java/org/apache/nifi/processors/elasticsearch/TestPutElasticsearchHttpRecord.java
index 862e177..2cc16c1 100644
--- a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/test/java/org/apache/nifi/processors/elasticsearch/TestPutElasticsearchHttpRecord.java
+++ b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/test/java/org/apache/nifi/processors/elasticsearch/TestPutElasticsearchHttpRecord.java
@@ -16,18 +16,15 @@
*/
package org.apache.nifi.processors.elasticsearch;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertTrue;
-import static org.mockito.Matchers.any;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.when;
-
-import java.io.IOException;
-import java.net.ConnectException;
-import java.util.HashMap;
-import java.util.List;
-
+import com.fasterxml.jackson.databind.ObjectMapper;
+import okhttp3.Call;
+import okhttp3.MediaType;
+import okhttp3.OkHttpClient;
+import okhttp3.Protocol;
+import okhttp3.Request;
+import okhttp3.Response;
+import okhttp3.ResponseBody;
+import okio.Buffer;
import org.apache.nifi.processor.ProcessContext;
import org.apache.nifi.processor.exception.ProcessException;
import org.apache.nifi.provenance.ProvenanceEventRecord;
@@ -42,16 +39,21 @@ import org.junit.After;
import org.junit.Ignore;
import org.junit.Test;
-import okhttp3.Call;
-import okhttp3.MediaType;
-import okhttp3.OkHttpClient;
-import okhttp3.Protocol;
-import okhttp3.Request;
-import okhttp3.Response;
-import okhttp3.ResponseBody;
+import java.io.IOException;
+import java.net.ConnectException;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Consumer;
-public class TestPutElasticsearchHttpRecord {
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Matchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+public class TestPutElasticsearchHttpRecord {
private TestRunner runner;
@After
@@ -61,7 +63,25 @@ public class TestPutElasticsearchHttpRecord {
@Test
public void testPutElasticSearchOnTriggerIndex() throws IOException {
- runner = TestRunners.newTestRunner(new PutElasticsearchHttpRecordTestProcessor(false)); // no failures
+ PutElasticsearchHttpRecordTestProcessor processor = new PutElasticsearchHttpRecordTestProcessor(false);
+ processor.setRecordChecks(record -> {
+ assertEquals(1, record.get("id"));
+ assertEquals("reç1", record.get("name"));
+ assertEquals(101, record.get("code"));
+ }, record -> {
+ assertEquals(2, record.get("id"));
+ assertEquals("ræc2", record.get("name"));
+ assertEquals(102, record.get("code"));
+ }, record -> {
+ assertEquals(3, record.get("id"));
+ assertEquals("rèc3", record.get("name"));
+ assertEquals(103, record.get("code"));
+ }, record -> {
+ assertEquals(4, record.get("id"));
+ assertEquals("rëc4", record.get("name"));
+ assertEquals(104, record.get("code"));
+ });
+ runner = TestRunners.newTestRunner(processor); // no failures
generateTestData();
runner.setProperty(AbstractElasticsearchHttpProcessor.ES_URL, "http://127.0.0.1:9200");
@@ -368,6 +388,7 @@ public class TestPutElasticsearchHttpRecord {
int statusCode = 200;
String statusMessage = "OK";
String expectedUrl = null;
+ Consumer<Map>[] recordChecks;
PutElasticsearchHttpRecordTestProcessor(boolean responseHasFailures) {
this.responseHasFailures = responseHasFailures;
@@ -382,6 +403,11 @@ public class TestPutElasticsearchHttpRecord {
expectedUrl = url;
}
+ @SafeVarargs
+ final void setRecordChecks(Consumer<Map>... checks) {
+ recordChecks = checks;
+ }
+
@Override
protected void createElasticsearchClient(ProcessContext context) throws ProcessException {
client = mock(OkHttpClient.class);
@@ -391,6 +417,24 @@ public class TestPutElasticsearchHttpRecord {
if (statusCode != -1) {
Request realRequest = (Request) invocationOnMock.getArguments()[0];
assertTrue((expectedUrl == null) || (expectedUrl.equals(realRequest.url().toString())));
+ if (recordChecks != null) {
+ final ObjectMapper mapper = new ObjectMapper();
+ Buffer sink = new Buffer();
+ realRequest.body().writeTo(sink);
+ String line;
+ int recordIndex = 0;
+ boolean content = false;
+ while ((line = sink.readUtf8Line()) != null) {
+ if (content) {
+ content = false;
+ if (recordIndex < recordChecks.length) {
+ recordChecks[recordIndex++].accept(mapper.readValue(line, Map.class));
+ }
+ } else {
+ content = true;
+ }
+ }
+ }
StringBuilder sb = new StringBuilder("{\"took\": 1, \"errors\": \"");
sb.append(responseHasFailures);
sb.append("\", \"items\": [");
@@ -521,9 +565,9 @@ public class TestPutElasticsearchHttpRecord {
parser.addSchemaField("name", RecordFieldType.STRING);
parser.addSchemaField("code", RecordFieldType.INT);
- parser.addRecord(1, "rec1", 101);
- parser.addRecord(2, "rec2", 102);
- parser.addRecord(3, "rec3", 103);
- parser.addRecord(4, "rec4", 104);
+ parser.addRecord(1, "reç1", 101);
+ parser.addRecord(2, "ræc2", 102);
+ parser.addRecord(3, "rèc3", 103);
+ parser.addRecord(4, "rëc4", 104);
}
}