This is an automated email from the ASF dual-hosted git repository.
tallison pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/tika.git
The following commit(s) were added to refs/heads/main by this push:
new 0abebdc27 TIKA-3855 -- enable upsert for OpenSearchEmitter
new 6e1e6b724 Merge remote-tracking branch 'origin/main'
0abebdc27 is described below
commit 0abebdc27dbfc9eb34abc3619e13ef816af9e331
Author: tallison <[email protected]>
AuthorDate: Thu Sep 15 12:22:10 2022 -0400
TIKA-3855 -- enable upsert for OpenSearchEmitter
---
CHANGES.txt | 13 +-
.../opensearch/tests/TikaPipesOpenSearchTest.java | 3 +-
.../pipes/xsearch/tests/TikaPipesXSearchBase.java | 122 +++++++++--
.../pipes/xsearch/tests/XSearchTestClient.java | 11 +-
.../opensearch/tika-config-opensearch.xml | 2 -
.../pipes/emitter/opensearch/OpenSearchClient.java | 240 +++++++++++++++------
.../emitter/opensearch/OpenSearchEmitter.java | 27 ++-
.../emitter/opensearch/OpenSearchClientTest.java | 4 +-
8 files changed, 318 insertions(+), 104 deletions(-)
diff --git a/CHANGES.txt b/CHANGES.txt
index cda84e564..9583c56cd 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,4 @@
-Release 2.4.2 - ???
+Release 2.5.0 - ???
* Improved extraction of PDF subset info for PDF/UA, PDF/VT, and PDF/X.
NOTE: we no longer append PDF/A information, e.g. 'version="A-1b"'
@@ -6,6 +6,17 @@ Release 2.4.2 - ???
'pdfa:PDFVersion' key or from 'pdfaid:conformance'
and 'pdfaid:part' (TIKA-3844).
+ * Avoid infinite loop in bookmark extraction from PDFs (TIKA-3832).
+
+ * Added upsert option for the OpenSearch emitter (TIKA-3855).
+
+ * Extract PDF signature information at the document level
+ into the metadata (TIKA-3852).
+
+ * Enable configuration of digests via AutoDetectParserConfig (TIKA-3853).
+
+ * Use commons-io byte array streams via PJ Fanning (TIKA-3843).
+
* Add extraction of ODF version from ODF files (TIKA-3840).
* tika-parser-html-commons (BoilerPipeHandler) is no longer a
diff --git
a/tika-integration-tests/tika-pipes-opensearch-integration-tests/src/test/java/org/apache/tika/pipes/opensearch/tests/TikaPipesOpenSearchTest.java
b/tika-integration-tests/tika-pipes-opensearch-integration-tests/src/test/java/org/apache/tika/pipes/opensearch/tests/TikaPipesOpenSearchTest.java
index 3bb2d5c4a..cb2646ba8 100644
---
a/tika-integration-tests/tika-pipes-opensearch-integration-tests/src/test/java/org/apache/tika/pipes/opensearch/tests/TikaPipesOpenSearchTest.java
+++
b/tika-integration-tests/tika-pipes-opensearch-integration-tests/src/test/java/org/apache/tika/pipes/opensearch/tests/TikaPipesOpenSearchTest.java
@@ -16,6 +16,7 @@
*/
package org.apache.tika.pipes.opensearch.tests;
+
import java.time.Duration;
import java.time.temporal.ChronoUnit;
@@ -28,7 +29,7 @@ import
org.apache.tika.pipes.xsearch.tests.TikaPipesXSearchBase;
public class TikaPipesOpenSearchTest extends TikaPipesXSearchBase {
- private static final String DOCKER_IMAGE_NAME =
"opensearchproject/opensearch:2.1.0";
+ private static final String DOCKER_IMAGE_NAME =
"opensearchproject/opensearch:2.3.0";
@ClassRule
public static GenericContainer<?> OPEN_SEARCH_CONTAINER =
diff --git
a/tika-integration-tests/tika-pipes-opensearch-integration-tests/src/test/java/org/apache/tika/pipes/xsearch/tests/TikaPipesXSearchBase.java
b/tika-integration-tests/tika-pipes-opensearch-integration-tests/src/test/java/org/apache/tika/pipes/xsearch/tests/TikaPipesXSearchBase.java
index 92cddbe6c..1f8378157 100644
---
a/tika-integration-tests/tika-pipes-opensearch-integration-tests/src/test/java/org/apache/tika/pipes/xsearch/tests/TikaPipesXSearchBase.java
+++
b/tika-integration-tests/tika-pipes-opensearch-integration-tests/src/test/java/org/apache/tika/pipes/xsearch/tests/TikaPipesXSearchBase.java
@@ -21,11 +21,13 @@ import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import java.io.File;
+import java.io.IOException;
import java.io.InputStream;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
+import java.util.Collections;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
@@ -33,31 +35,47 @@ import com.fasterxml.jackson.databind.JsonNode;
import org.apache.commons.io.IOUtils;
import org.jetbrains.annotations.NotNull;
import org.junit.After;
+import org.junit.Before;
import org.junit.Test;
+import org.junit.jupiter.api.Assertions;
import org.testcontainers.containers.GenericContainer;
import org.testcontainers.shaded.org.apache.commons.io.FileUtils;
import org.apache.tika.cli.TikaCLI;
import org.apache.tika.client.HttpClientFactory;
+import org.apache.tika.metadata.Metadata;
import org.apache.tika.pipes.HandlerConfig;
+import org.apache.tika.pipes.emitter.Emitter;
+import org.apache.tika.pipes.emitter.EmitterManager;
import org.apache.tika.pipes.emitter.opensearch.JsonResponse;
import org.apache.tika.pipes.emitter.opensearch.OpenSearchEmitter;
public abstract class TikaPipesXSearchBase {
- private static final String TEST_INDEX = "tika-pipes-index";
- private static final File TEST_FILE_FOLDER = new File("target",
"test-files");
+ //@TempDir -- we can't use this yet because of junit4+junit5
+ private Path pipesDirectory;
+
+ private Path testDocDirectory;
+
+ protected static final String TEST_INDEX = "tika-pipes-index";
private int numTestDocs = 0;
private static String OPEN_SEARCH_HOST;
private static int OPEN_SEARCH_PORT;
//this includes only the base, not the collection, e.g.
https://localhost:49213
- private static String OPEN_SEARCH_ENDPOINT_BASE;
+ protected static String OPEN_SEARCH_ENDPOINT_BASE;
private static XSearchTestClient client;
+ @Before
+ public void setUp() throws IOException {
+ pipesDirectory =
Files.createTempDirectory("tika-opensearch-integration-");
+ testDocDirectory = pipesDirectory.resolve("docs");
+ }
@After
public void tearDown() throws Exception {
- FileUtils.deleteDirectory(TEST_FILE_FOLDER);
+ //we shouldn't have to do this because of @TempDir
+ //there are some timing/order issues because of the joint junit 4 and 5
+ FileUtils.deleteDirectory(pipesDirectory.toFile());
JsonResponse response = client.deleteIndex(OPEN_SEARCH_ENDPOINT_BASE +
TEST_INDEX);
assertEquals(200, response.getStatus());
assertTrue(response.getJson().get("acknowledged").asBoolean());
@@ -72,6 +90,7 @@ public abstract class TikaPipesXSearchBase {
sendMappings(endpoint, TEST_INDEX, "opensearch-mappings.json");
runPipes(OpenSearchEmitter.AttachmentStrategy.SEPARATE_DOCUMENTS,
+ OpenSearchEmitter.UpdateStrategy.OVERWRITE,
HandlerConfig.PARSE_MODE.CONCATENATE, endpoint);
String query = "{ \"track_total_hits\": true, \"query\": { \"match\":
{ \"content\": { " +
@@ -98,6 +117,7 @@ public abstract class TikaPipesXSearchBase {
sendMappings(endpoint, TEST_INDEX,
"opensearch-parent-child-mappings.json");
runPipes(OpenSearchEmitter.AttachmentStrategy.PARENT_CHILD,
+ OpenSearchEmitter.UpdateStrategy.OVERWRITE,
HandlerConfig.PARSE_MODE.RMETA, endpoint);
String query = "{ \"track_total_hits\": true, \"query\": { \"match\":
{ \"content\": { " +
@@ -161,6 +181,7 @@ public abstract class TikaPipesXSearchBase {
sendMappings(endpoint, TEST_INDEX, "opensearch-mappings.json");
runPipes(OpenSearchEmitter.AttachmentStrategy.SEPARATE_DOCUMENTS,
+ OpenSearchEmitter.UpdateStrategy.OVERWRITE,
HandlerConfig.PARSE_MODE.RMETA, endpoint);
String query = "{ \"track_total_hits\": true, \"query\": { \"match\":
{ \"content\": { " +
@@ -214,8 +235,42 @@ public abstract class TikaPipesXSearchBase {
assertEquals(400, results.getStatus());
}
+ @Test
+ public void testUpsert() throws Exception {
+ String endpoint = OPEN_SEARCH_ENDPOINT_BASE + TEST_INDEX;
+ sendMappings(endpoint, TEST_INDEX, "opensearch-mappings.json");
+ Path tikaConfigFile =
+
getTikaConfigFile(OpenSearchEmitter.AttachmentStrategy.SEPARATE_DOCUMENTS,
+ OpenSearchEmitter.UpdateStrategy.UPSERT,
HandlerConfig.PARSE_MODE.RMETA,
+ endpoint);
+ Emitter emitter = EmitterManager.load(tikaConfigFile).getEmitter();
+ Metadata metadata = new Metadata();
+ metadata.set("mime", "mimeA");
+ metadata.set("title", "titleA");
+ emitter.emit("1", Collections.singletonList(metadata));
+ JsonResponse refresh = client.getJson(endpoint + "/_refresh");
+ metadata.set("title", "titleB");
+ emitter.emit("1", Collections.singletonList(metadata));
+ refresh = client.getJson(endpoint + "/_refresh");
+
+ Metadata metadata2 = new Metadata();
+ metadata2.set("content", "the quick brown fox");
+ emitter.emit("1", Collections.singletonList(metadata2));
+ refresh = client.getJson(endpoint + "/_refresh");
- private void sendMappings(String endpoint, String index, String
mappingsFile) throws Exception {
+ String query = "{ " +
+ //"\"from\":0, \"size\":1000," +
+ "\"track_total_hits\": true, \"query\": { " +
+ "\"match_all\": {} } }";
+ JsonResponse response = client.postJson(endpoint + "/_search", query);
+ JsonNode doc1 = response.getJson().get("hits").get("hits").get(0).get(
+ "_source");
+ Assertions.assertEquals("mimeA", doc1.get("mime").asText());
+ Assertions.assertEquals("titleB", doc1.get("title").asText());
+ Assertions.assertEquals("the quick brown fox",
doc1.get("content").asText());
+ }
+
+ protected void sendMappings(String endpoint, String index, String
mappingsFile) throws Exception {
//create the collection with mappings
String mappings =
IOUtils.toString(TikaPipesXSearchBase.class.getResourceAsStream(
"/opensearch/" + mappingsFile), StandardCharsets.UTF_8);
@@ -240,14 +295,30 @@ public abstract class TikaPipesXSearchBase {
}
private void runPipes(OpenSearchEmitter.AttachmentStrategy
attachmentStrategy,
+ OpenSearchEmitter.UpdateStrategy updateStrategy,
HandlerConfig.PARSE_MODE parseMode, String endpoint)
throws Exception {
- File tikaConfigFile = new File("target", "ta-opensearch.xml");
- File log4jPropFile = new File("target", "tmp-log4j2.xml");
+ Path tikaConfigFile = getTikaConfigFile(attachmentStrategy,
updateStrategy, parseMode,
+ endpoint);
+
+ TikaCLI.main(new String[]{"-a", "--config=" +
tikaConfigFile.toAbsolutePath().toString()});
+
+ //refresh to make sure the content is searchable
+ JsonResponse refresh = client.getJson(endpoint + "/_refresh");
+
+ }
+
+ private Path getTikaConfigFile(OpenSearchEmitter.AttachmentStrategy
attachmentStrategy,
+ OpenSearchEmitter.UpdateStrategy
updateStrategy,
+ HandlerConfig.PARSE_MODE parseMode, String
endpoint) throws
+ IOException {
+ Path tikaConfigFile = pipesDirectory.resolve("ta-opensearch.xml");
+ Path log4jPropFile = pipesDirectory.resolve("tmp-log4j2.xml");
try (InputStream is = TikaPipesXSearchBase.class
.getResourceAsStream("/pipes-fork-server-custom-log4j2.xml")) {
- FileUtils.copyInputStreamToFile(is, log4jPropFile);
+ Files.copy(is, log4jPropFile);
}
+
String tikaConfigTemplateXml;
try (InputStream is = TikaPipesXSearchBase.class
.getResourceAsStream("/opensearch/tika-config-opensearch.xml")) {
@@ -256,26 +327,25 @@ public abstract class TikaPipesXSearchBase {
String tikaConfigXml =
createTikaConfigXml(tikaConfigFile, log4jPropFile,
tikaConfigTemplateXml,
- attachmentStrategy, parseMode, endpoint);
- FileUtils.writeStringToFile(tikaConfigFile, tikaConfigXml,
StandardCharsets.UTF_8);
-
- TikaCLI.main(new String[]{"-a", "--config=" +
tikaConfigFile.getAbsolutePath()});
-
- //refresh to make sure the content is searchable
- JsonResponse refresh = client.getJson(endpoint + "/_refresh");
+ attachmentStrategy, updateStrategy, parseMode,
endpoint);
+ writeStringToPath(tikaConfigFile, tikaConfigXml);
+ return tikaConfigFile;
}
@NotNull
- private String createTikaConfigXml(File tikaConfigFile, File
log4jPropFile, String tikaConfigTemplateXml,
+ private String createTikaConfigXml(Path tikaConfigFile, Path log4jPropFile,
+ String tikaConfigTemplateXml,
OpenSearchEmitter.AttachmentStrategy
attachmentStrategy,
+ OpenSearchEmitter.UpdateStrategy
updateStrategy,
HandlerConfig.PARSE_MODE parseMode,
String endpoint) {
String res =
- tikaConfigTemplateXml.replace("{TIKA_CONFIG}",
tikaConfigFile.getAbsolutePath())
+ tikaConfigTemplateXml.replace("{TIKA_CONFIG}",
tikaConfigFile.toAbsolutePath().toString())
.replace("{ATTACHMENT_STRATEGY}",
attachmentStrategy.toString())
- .replace("{LOG4J_PROPERTIES_FILE}",
log4jPropFile.getAbsolutePath())
+ .replace("{LOG4J_PROPERTIES_FILE}",
log4jPropFile.toAbsolutePath().toString())
+ .replace("{UPDATE_STRATEGY}",
updateStrategy.toString())
.replaceAll("\\{PATH_TO_DOCS\\}",
-
Matcher.quoteReplacement(TEST_FILE_FOLDER.getAbsolutePath()))
+
Matcher.quoteReplacement(testDocDirectory.toAbsolutePath().toString()))
.replace("{PARSE_MODE}", parseMode.name());
res = res.replace("{OPENSEARCH_CONNECTION}", endpoint);
@@ -295,23 +365,29 @@ public abstract class TikaPipesXSearchBase {
client = new XSearchTestClient(OPEN_SEARCH_ENDPOINT_BASE,
httpClientFactory.build(),
OpenSearchEmitter.AttachmentStrategy.SEPARATE_DOCUMENTS,
+ OpenSearchEmitter.UpdateStrategy.OVERWRITE,
OpenSearchEmitter.DEFAULT_EMBEDDED_FILE_FIELD_NAME);
}
private void createTestHtmlFiles(String bodyContent, int numHtmlDocs)
throws Exception {
- TEST_FILE_FOLDER.mkdirs();
+ Files.createDirectories(testDocDirectory);
for (int i = 0; i < numHtmlDocs; ++i) {
- FileUtils.writeStringToFile(new File(TEST_FILE_FOLDER, "test-" + i
+ ".html"),
- "<html><body>" + bodyContent + "</body></html>",
StandardCharsets.UTF_8);
+ String html = "<html><body>" + bodyContent + "</body></html>";
+ Path p = testDocDirectory.resolve( "test-" + i + ".html");
+ writeStringToPath(p, html);
}
File testDocuments =
Paths.get(TikaPipesXSearchBase.class.getResource("/test-documents").toURI()).toFile();
for (File f : testDocuments.listFiles()) {
- Path targ = TEST_FILE_FOLDER.toPath().resolve(f.getName());
+ Path targ = testDocDirectory.resolve(f.getName());
Files.copy(f.toPath(), targ);
numTestDocs++;
}
}
+ private static void writeStringToPath(Path path, String string) throws
IOException {
+ Files.write(path, string.getBytes(StandardCharsets.UTF_8));
+ }
+
}
diff --git
a/tika-integration-tests/tika-pipes-opensearch-integration-tests/src/test/java/org/apache/tika/pipes/xsearch/tests/XSearchTestClient.java
b/tika-integration-tests/tika-pipes-opensearch-integration-tests/src/test/java/org/apache/tika/pipes/xsearch/tests/XSearchTestClient.java
index ec8191f3c..8f2c580f1 100644
---
a/tika-integration-tests/tika-pipes-opensearch-integration-tests/src/test/java/org/apache/tika/pipes/xsearch/tests/XSearchTestClient.java
+++
b/tika-integration-tests/tika-pipes-opensearch-integration-tests/src/test/java/org/apache/tika/pipes/xsearch/tests/XSearchTestClient.java
@@ -43,10 +43,11 @@ import
org.apache.tika.pipes.emitter.opensearch.OpenSearchEmitter;
*/
public class XSearchTestClient extends OpenSearchClient {
- protected XSearchTestClient(String openSearchUrl, HttpClient httpClient,
- OpenSearchEmitter.AttachmentStrategy
attachmentStrategy,
- String embeddedFileFieldName) {
- super(openSearchUrl, httpClient, attachmentStrategy,
embeddedFileFieldName);
+ public XSearchTestClient(String openSearchUrl, HttpClient httpClient,
+ OpenSearchEmitter.AttachmentStrategy
attachmentStrategy,
+ OpenSearchEmitter.UpdateStrategy updateStrategy,
+ String embeddedFileFieldName) {
+ super(openSearchUrl, httpClient, attachmentStrategy, updateStrategy,
embeddedFileFieldName);
}
protected JsonResponse putJson(String url, String json) throws IOException
{
@@ -86,7 +87,7 @@ public class XSearchTestClient extends OpenSearchClient {
}
}
- protected JsonResponse getJson(String url) throws IOException {
+ public JsonResponse getJson(String url) throws IOException {
HttpGet httpRequest = new HttpGet(url);
httpRequest.setHeader("Accept", "application/json");
httpRequest.setHeader("Content-type", "application/json;
charset=utf-8");
diff --git
a/tika-integration-tests/tika-pipes-opensearch-integration-tests/src/test/resources/opensearch/tika-config-opensearch.xml
b/tika-integration-tests/tika-pipes-opensearch-integration-tests/src/test/resources/opensearch/tika-config-opensearch.xml
index df9452a2f..ee727b1ad 100644
---
a/tika-integration-tests/tika-pipes-opensearch-integration-tests/src/test/resources/opensearch/tika-config-opensearch.xml
+++
b/tika-integration-tests/tika-pipes-opensearch-integration-tests/src/test/resources/opensearch/tika-config-opensearch.xml
@@ -93,9 +93,7 @@
<params>
<name>ose</name>
<openSearchUrl>{OPENSEARCH_CONNECTION}</openSearchUrl>
-<!-- TODO: implement this
<updateStrategy>{UPDATE_STRATEGY}</updateStrategy>
- -->
<attachmentStrategy>{ATTACHMENT_STRATEGY}</attachmentStrategy>
<commitWithin>10</commitWithin>
<idField>_id</idField>
diff --git
a/tika-pipes/tika-emitters/tika-emitter-opensearch/src/main/java/org/apache/tika/pipes/emitter/opensearch/OpenSearchClient.java
b/tika-pipes/tika-emitters/tika-emitter-opensearch/src/main/java/org/apache/tika/pipes/emitter/opensearch/OpenSearchClient.java
index 3eff65997..3b8061a95 100644
---
a/tika-pipes/tika-emitters/tika-emitter-opensearch/src/main/java/org/apache/tika/pipes/emitter/opensearch/OpenSearchClient.java
+++
b/tika-pipes/tika-emitters/tika-emitter-opensearch/src/main/java/org/apache/tika/pipes/emitter/opensearch/OpenSearchClient.java
@@ -52,13 +52,18 @@ public class OpenSearchClient {
protected final String openSearchUrl;
protected final HttpClient httpClient;
private final OpenSearchEmitter.AttachmentStrategy attachmentStrategy;
+
+ private final MetadataToJsonWriter metadataToJsonWriter;
private final String embeddedFileFieldName;
protected OpenSearchClient(String openSearchUrl, HttpClient httpClient,
OpenSearchEmitter.AttachmentStrategy
attachmentStrategy,
+ OpenSearchEmitter.UpdateStrategy updateStrategy,
String embeddedFileFieldName) {
this.openSearchUrl = openSearchUrl;
this.httpClient = httpClient;
this.attachmentStrategy = attachmentStrategy;
+ this.metadataToJsonWriter = (updateStrategy ==
OpenSearchEmitter.UpdateStrategy.OVERWRITE) ?
+ new InsertMetadataToJsonWriter() : new
UpsertMetadataToJsonWriter();
this.embeddedFileFieldName = embeddedFileFieldName;
}
@@ -73,7 +78,6 @@ public class OpenSearchClient {
private void emitJson(StringBuilder json) throws IOException,
TikaClientException {
String requestUrl = openSearchUrl + "/_bulk";
-
JsonResponse response = postJson(requestUrl, json.toString());
if (response.getStatus() != 200) {
throw new TikaClientException(response.getMsg());
@@ -107,12 +111,12 @@ public class OpenSearchClient {
if (i > 0) {
id.append("-").append(UUID.randomUUID());
}
- String indexJson = getBulkIndexJson(id.toString(), routing);
+ String indexJson = metadataToJsonWriter.getBulkJson(id.toString(),
routing);
json.append(indexJson).append("\n");
if (i == 0) {
- json.append(metadataToJsonContainer(metadata,
attachmentStrategy));
+ json.append(metadataToJsonWriter.writeContainer(metadata,
attachmentStrategy));
} else {
- json.append(metadataToJsonEmbedded(metadata,
attachmentStrategy, emitKey,
+ json.append(metadataToJsonWriter.writeEmbedded(metadata,
attachmentStrategy, emitKey,
embeddedFileFieldName));
}
json.append("\n");
@@ -120,75 +124,20 @@ public class OpenSearchClient {
}
}
- protected static String metadataToJsonEmbedded(Metadata metadata,
-
OpenSearchEmitter.AttachmentStrategy attachmentStrategy,
- String emitKey, String
embeddedFileFieldName) throws IOException {
- StringWriter writer = new StringWriter();
- try (JsonGenerator jsonGenerator = new
JsonFactory().createGenerator(writer)) {
- jsonGenerator.writeStartObject();
-
- writeMetadata(metadata, jsonGenerator);
- if (attachmentStrategy ==
OpenSearchEmitter.AttachmentStrategy.PARENT_CHILD) {
- jsonGenerator.writeObjectFieldStart("relation_type");
- jsonGenerator.writeStringField("name", embeddedFileFieldName);
- jsonGenerator.writeStringField("parent", emitKey);
- //end the relation type object
- jsonGenerator.writeEndObject();
- } else if (attachmentStrategy ==
OpenSearchEmitter.AttachmentStrategy.SEPARATE_DOCUMENTS) {
- jsonGenerator.writeStringField("parent", emitKey);
- }
- //end the metadata object
- jsonGenerator.writeEndObject();
- }
- return writer.toString();
- }
-
- protected static String metadataToJsonContainer(Metadata metadata,
+ //Only here for testing. These may disappear without notice in the future.
+ protected static String metadataToJsonContainerInsert(Metadata metadata,
OpenSearchEmitter.AttachmentStrategy attachmentStrategy)
throws IOException {
- StringWriter writer = new StringWriter();
- try (JsonGenerator jsonGenerator = new
JsonFactory().createGenerator(writer)) {
- jsonGenerator.writeStartObject();
- writeMetadata(metadata, jsonGenerator);
- if (attachmentStrategy ==
OpenSearchEmitter.AttachmentStrategy.PARENT_CHILD) {
- jsonGenerator.writeStringField("relation_type", "container");
- }
- jsonGenerator.writeEndObject();
- }
- return writer.toString();
+ return new InsertMetadataToJsonWriter().writeContainer(metadata,
attachmentStrategy);
}
- private static void writeMetadata(Metadata metadata, JsonGenerator
jsonGenerator) throws IOException {
- //writes the metadata without the start { or the end }
- //to allow for other fields to be added
- for (String n : metadata.names()) {
- String[] vals = metadata.getValues(n);
- if (vals.length == 1) {
- jsonGenerator.writeStringField(n, vals[0]);
- } else {
- jsonGenerator.writeArrayFieldStart(n);
- for (String v : vals) {
- jsonGenerator.writeString(v);
- }
- jsonGenerator.writeEndArray();
- }
- }
- }
-
- private String getBulkIndexJson(String id, String routing) throws
IOException {
- StringWriter writer = new StringWriter();
- try (JsonGenerator jsonGenerator = new
JsonFactory().createGenerator(writer)) {
- jsonGenerator.writeStartObject();
- jsonGenerator.writeObjectFieldStart("index");
- jsonGenerator.writeStringField("_id", id);
- if (!StringUtils.isEmpty(routing)) {
- jsonGenerator.writeStringField("routing", routing);
- }
-
- jsonGenerator.writeEndObject();
- jsonGenerator.writeEndObject();
- }
- return writer.toString();
+ //Only here for testing. These may disappear without notice in the future.
+ protected static String metadataToJsonEmbeddedInsert(Metadata metadata,
+
OpenSearchEmitter.AttachmentStrategy attachmentStrategy,
+ String emitKey,
String embeddedFileFieldName)
+ throws IOException {
+ return new InsertMetadataToJsonWriter().writeEmbedded(metadata,
+ attachmentStrategy, emitKey, embeddedFileFieldName);
}
public JsonResponse postJson(String url, String json) throws IOException {
@@ -230,4 +179,157 @@ public class OpenSearchClient {
httpRequest.releaseConnection();
}
}
+
+ private interface MetadataToJsonWriter {
+ String writeContainer(Metadata metadata,
OpenSearchEmitter.AttachmentStrategy attachmentStrategy)
+ throws IOException;
+
+ String writeEmbedded(Metadata metadata,
OpenSearchEmitter.AttachmentStrategy attachmentStrategy,
+ String emitKey, String embeddedFileFieldName)
throws IOException;
+
+ String getBulkJson(String id, String routing) throws IOException;
+ }
+
+ private static class InsertMetadataToJsonWriter implements
MetadataToJsonWriter {
+
+ @Override
+ public String writeContainer(Metadata metadata,
+ OpenSearchEmitter.AttachmentStrategy
attachmentStrategy)
+ throws IOException {
+ StringWriter writer = new StringWriter();
+ try (JsonGenerator jsonGenerator = new
JsonFactory().createGenerator(writer)) {
+ jsonGenerator.writeStartObject();
+ writeMetadata(metadata, jsonGenerator);
+ if (attachmentStrategy ==
OpenSearchEmitter.AttachmentStrategy.PARENT_CHILD) {
+ jsonGenerator.writeStringField("relation_type",
"container");
+ }
+ jsonGenerator.writeEndObject();
+ }
+ return writer.toString();
+ }
+
+ @Override
+ public String writeEmbedded(Metadata metadata,
+ OpenSearchEmitter.AttachmentStrategy
attachmentStrategy,
+ String emitKey, String
embeddedFileFieldName)
+ throws IOException {
+ StringWriter writer = new StringWriter();
+ try (JsonGenerator jsonGenerator = new
JsonFactory().createGenerator(writer)) {
+ jsonGenerator.writeStartObject();
+
+ writeMetadata(metadata, jsonGenerator);
+ if (attachmentStrategy ==
OpenSearchEmitter.AttachmentStrategy.PARENT_CHILD) {
+ jsonGenerator.writeObjectFieldStart("relation_type");
+ jsonGenerator.writeStringField("name",
embeddedFileFieldName);
+ jsonGenerator.writeStringField("parent", emitKey);
+ //end the relation type object
+ jsonGenerator.writeEndObject();
+ } else if (attachmentStrategy ==
OpenSearchEmitter.AttachmentStrategy.SEPARATE_DOCUMENTS) {
+ jsonGenerator.writeStringField("parent", emitKey);
+ }
+ //end the metadata object
+ jsonGenerator.writeEndObject();
+ }
+ return writer.toString();
+ }
+
+ @Override
+ public String getBulkJson(String id, String routing) throws
IOException {
+ StringWriter writer = new StringWriter();
+ try (JsonGenerator jsonGenerator = new
JsonFactory().createGenerator(writer)) {
+ jsonGenerator.writeStartObject();
+ jsonGenerator.writeObjectFieldStart("index");
+ jsonGenerator.writeStringField("_id", id);
+ if (!StringUtils.isEmpty(routing)) {
+ jsonGenerator.writeStringField("routing", routing);
+ }
+ jsonGenerator.writeEndObject();
+ jsonGenerator.writeEndObject();
+ }
+ return writer.toString();
+ }
+ }
+
+ private static class UpsertMetadataToJsonWriter implements
MetadataToJsonWriter {
+
+ @Override
+ public String writeContainer(Metadata metadata,
+ OpenSearchEmitter.AttachmentStrategy
attachmentStrategy)
+ throws IOException {
+ StringWriter writer = new StringWriter();
+ try (JsonGenerator jsonGenerator = new
JsonFactory().createGenerator(writer)) {
+ jsonGenerator.writeStartObject();
+ jsonGenerator.writeObjectFieldStart("doc");
+ writeMetadata(metadata, jsonGenerator);
+ if (attachmentStrategy ==
OpenSearchEmitter.AttachmentStrategy.PARENT_CHILD) {
+ jsonGenerator.writeStringField("relation_type",
"container");
+ }
+ jsonGenerator.writeEndObject();
+ jsonGenerator.writeBooleanField("doc_as_upsert", true);
+ jsonGenerator.writeEndObject();
+ }
+ return writer.toString();
+ }
+
+ @Override
+ public String writeEmbedded(Metadata metadata,
+ OpenSearchEmitter.AttachmentStrategy
attachmentStrategy,
+ String emitKey, String
embeddedFileFieldName)
+ throws IOException {
+ StringWriter writer = new StringWriter();
+ try (JsonGenerator jsonGenerator = new
JsonFactory().createGenerator(writer)) {
+ jsonGenerator.writeStartObject();
+ jsonGenerator.writeObjectFieldStart("doc");
+ writeMetadata(metadata, jsonGenerator);
+ if (attachmentStrategy ==
OpenSearchEmitter.AttachmentStrategy.PARENT_CHILD) {
+ jsonGenerator.writeObjectFieldStart("relation_type");
+ jsonGenerator.writeStringField("name",
embeddedFileFieldName);
+ jsonGenerator.writeStringField("parent", emitKey);
+ //end the relation type object
+ jsonGenerator.writeEndObject();
+ } else if (attachmentStrategy ==
OpenSearchEmitter.AttachmentStrategy.SEPARATE_DOCUMENTS) {
+ jsonGenerator.writeStringField("parent", emitKey);
+ }
+ //end the "doc"
+ jsonGenerator.writeEndObject();
+ //end the metadata object
+ jsonGenerator.writeEndObject();
+ }
+ return writer.toString();
+ }
+
+ @Override
+ public String getBulkJson(String id, String routing) throws
IOException {
+ StringWriter writer = new StringWriter();
+ try (JsonGenerator jsonGenerator = new
JsonFactory().createGenerator(writer)) {
+ jsonGenerator.writeStartObject();
+ jsonGenerator.writeObjectFieldStart("update");
+ jsonGenerator.writeStringField("_id", id);
+ if (!StringUtils.isEmpty(routing)) {
+ jsonGenerator.writeStringField("routing", routing);
+ }
+ jsonGenerator.writeNumberField("retry_on_conflict", 3);
+ jsonGenerator.writeEndObject();
+ jsonGenerator.writeEndObject();
+ }
+ return writer.toString();
+ }
+ }
+
+ private static void writeMetadata(Metadata metadata, JsonGenerator
jsonGenerator) throws IOException {
+ //writes the metadata without the start { or the end }
+ //to allow for other fields to be added
+ for (String n : metadata.names()) {
+ String[] vals = metadata.getValues(n);
+ if (vals.length == 1) {
+ jsonGenerator.writeStringField(n, vals[0]);
+ } else {
+ jsonGenerator.writeArrayFieldStart(n);
+ for (String v : vals) {
+ jsonGenerator.writeString(v);
+ }
+ jsonGenerator.writeEndArray();
+ }
+ }
+ }
}
diff --git
a/tika-pipes/tika-emitters/tika-emitter-opensearch/src/main/java/org/apache/tika/pipes/emitter/opensearch/OpenSearchEmitter.java
b/tika-pipes/tika-emitters/tika-emitter-opensearch/src/main/java/org/apache/tika/pipes/emitter/opensearch/OpenSearchEmitter.java
index 2e94221f1..bc010cf46 100644
---
a/tika-pipes/tika-emitters/tika-emitter-opensearch/src/main/java/org/apache/tika/pipes/emitter/opensearch/OpenSearchEmitter.java
+++
b/tika-pipes/tika-emitters/tika-emitter-opensearch/src/main/java/org/apache/tika/pipes/emitter/opensearch/OpenSearchEmitter.java
@@ -20,6 +20,7 @@ import static
org.apache.tika.config.TikaConfig.mustNotBeEmpty;
import java.io.IOException;
import java.util.List;
+import java.util.Locale;
import java.util.Map;
import org.slf4j.Logger;
@@ -47,10 +48,16 @@ public class OpenSearchEmitter extends AbstractEmitter
implements Initializable
//anything else?
}
+ public enum UpdateStrategy {
+ OVERWRITE, UPSERT
+ //others?
+ }
+
public static String DEFAULT_EMBEDDED_FILE_FIELD_NAME = "embedded";
private static final Logger LOG =
LoggerFactory.getLogger(OpenSearchEmitter.class);
private AttachmentStrategy attachmentStrategy =
AttachmentStrategy.PARENT_CHILD;
+ private UpdateStrategy updateStrategy = UpdateStrategy.OVERWRITE;
private String openSearchUrl = null;
private String idField = "_id";
private int commitWithin = 1000;
@@ -179,6 +186,24 @@ public class OpenSearchEmitter extends AbstractEmitter
implements Initializable
httpClientFactory.setProxyPort(proxyPort);
}
+ public void setUpdateStrategy(UpdateStrategy updateStrategy) {
+ this.updateStrategy = updateStrategy;
+ }
+
+ public void setUpdateStrategy(String strategy) throws TikaConfigException {
+ switch (strategy.toLowerCase(Locale.US)) {
+ case "overwrite" :
+ setUpdateStrategy(UpdateStrategy.OVERWRITE);
+ break;
+ case "upsert" :
+ setUpdateStrategy(UpdateStrategy.UPSERT);
+ break;
+ default :
+ throw new TikaConfigException("'overwrite' and 'upsert' are
the two options so " +
+ "far. I regret I don't understand: " + strategy);
+ }
+ }
+
/**
* If using the {@link AttachmentStrategy#PARENT_CHILD}, this is the field
name
* used to store the child documents. Note that we artificially flatten
all embedded
@@ -200,7 +225,7 @@ public class OpenSearchEmitter extends AbstractEmitter
implements Initializable
} else {
openSearchClient =
new OpenSearchClient(openSearchUrl,
- httpClientFactory.build(), attachmentStrategy,
+ httpClientFactory.build(), attachmentStrategy,
updateStrategy,
embeddedFileFieldName);
}
}
diff --git
a/tika-pipes/tika-emitters/tika-emitter-opensearch/src/test/java/org/apache/tika/pipes/emitter/opensearch/OpenSearchClientTest.java
b/tika-pipes/tika-emitters/tika-emitter-opensearch/src/test/java/org/apache/tika/pipes/emitter/opensearch/OpenSearchClientTest.java
index 638533c33..994ff0044 100644
---
a/tika-pipes/tika-emitters/tika-emitter-opensearch/src/test/java/org/apache/tika/pipes/emitter/opensearch/OpenSearchClientTest.java
+++
b/tika-pipes/tika-emitters/tika-emitter-opensearch/src/test/java/org/apache/tika/pipes/emitter/opensearch/OpenSearchClientTest.java
@@ -31,7 +31,7 @@ public class OpenSearchClientTest extends TikaTest {
metadata.add("title", "title1");
for (OpenSearchEmitter.AttachmentStrategy strategy :
OpenSearchEmitter.AttachmentStrategy.values()) {
- String json = OpenSearchClient.metadataToJsonContainer(metadata,
+ String json =
OpenSearchClient.metadataToJsonContainerInsert(metadata,
strategy);
assertContains("author1", json);
assertContains("author2", json);
@@ -40,7 +40,7 @@ public class OpenSearchClientTest extends TikaTest {
}
for (OpenSearchEmitter.AttachmentStrategy strategy :
OpenSearchEmitter.AttachmentStrategy.values()) {
- String json = OpenSearchClient.metadataToJsonEmbedded(metadata,
strategy,
+ String json =
OpenSearchClient.metadataToJsonEmbeddedInsert(metadata, strategy,
"myEmitKey",
OpenSearchEmitter.DEFAULT_EMBEDDED_FILE_FIELD_NAME);
assertContains("author1", json);
assertContains("author2", json);
