You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tika.apache.org by ta...@apache.org on 2021/03/24 18:27:00 UTC
[tika] 02/02: merge from main and required updates/conflict
resolution
This is an automated email from the ASF dual-hosted git repository.
tallison pushed a commit to branch TIKA-3304
in repository https://gitbox.apache.org/repos/asf/tika.git
commit 5dbffcd8eaae9135b509cf05a13d17c5542e2fc9
Author: tballison <ta...@apache.org>
AuthorDate: Wed Mar 24 14:26:41 2021 -0400
merge from main and required updates/conflict resolution
---
.gitignore | 1 +
.../tika/pipes/fetchiterator/FetchIterator.java | 46 ++++++++++++----------
.../org/apache/tika/server/client/TikaClient.java | 2 +-
.../apache/tika/server/client/TikaClientCLI.java | 35 ++++++++++++++--
.../tika/server/client/TikaEmitterResult.java | 7 ++--
.../apache/tika/server/client/TikaHttpClient.java | 6 +--
.../tika/server/core/resource/AsyncEmitter.java | 2 +-
.../tika/server/core/resource/EmitterResource.java | 23 ++---------
8 files changed, 72 insertions(+), 50 deletions(-)
diff --git a/.gitignore b/.gitignore
index dda6180..c608dc2 100644
--- a/.gitignore
+++ b/.gitignore
@@ -1,6 +1,7 @@
.svn
target
dependency-reduced-pom.xml
+.editorconfig
.idea
.classpath
.project
diff --git a/tika-core/src/main/java/org/apache/tika/pipes/fetchiterator/FetchIterator.java b/tika-core/src/main/java/org/apache/tika/pipes/fetchiterator/FetchIterator.java
index dde8222..2b9273d 100644
--- a/tika-core/src/main/java/org/apache/tika/pipes/fetchiterator/FetchIterator.java
+++ b/tika-core/src/main/java/org/apache/tika/pipes/fetchiterator/FetchIterator.java
@@ -17,18 +17,24 @@
package org.apache.tika.pipes.fetchiterator;
import java.io.IOException;
+import java.util.Iterator;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.FutureTask;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
import org.apache.tika.config.Field;
import org.apache.tika.config.Initializable;
import org.apache.tika.config.InitializableProblemHandler;
import org.apache.tika.config.Param;
import org.apache.tika.exception.TikaConfigException;
-
+import org.apache.tika.exception.TikaTimeoutException;
/**
* Abstract class that handles the testing for timeouts/thread safety
* issues. Concrete classes implement the blocking {@link #enqueue()}.
@@ -36,14 +42,13 @@ import org.apache.tika.exception.TikaConfigException;
* a RuntimeException. It will throw an IllegalStateException if
* next() is called after hasNext() has returned false.
*/
-public abstract class FetchIterator implements Callable<Integer>,
- Iterable<FetchEmitTuple>, Initializable {
+public abstract class FetchIterator
+ implements Callable<Integer>, Iterable<FetchEmitTuple>, Initializable {
public static final long DEFAULT_MAX_WAIT_MS = 300_000;
public static final int DEFAULT_QUEUE_SIZE = 1000;
- public static final FetchEmitTuple COMPLETED_SEMAPHORE =
- new FetchEmitTuple(null, null, null);
+ public static final FetchEmitTuple COMPLETED_SEMAPHORE = new FetchEmitTuple(null, null, null);
private static final Logger LOGGER = LoggerFactory.getLogger(FetchIterator.class);
@@ -53,16 +58,21 @@ public abstract class FetchIterator implements Callable<Integer>,
private String fetcherName;
private String emitterName;
private int added = 0;
- private FetchEmitTuple.ON_PARSE_EXCEPTION onParseException = FetchEmitTuple.ON_PARSE_EXCEPTION.EMIT;
+ private FetchEmitTuple.ON_PARSE_EXCEPTION onParseException =
+ FetchEmitTuple.ON_PARSE_EXCEPTION.EMIT;
private FutureTask<Integer> futureTask;
+ public String getFetcherName() {
+ return fetcherName;
+ }
+
@Field
public void setFetcherName(String fetcherName) {
this.fetcherName = fetcherName;
}
- public String getFetcherName() {
- return fetcherName;
+ public String getEmitterName() {
+ return emitterName;
}
@Field
@@ -70,10 +80,6 @@ public abstract class FetchIterator implements Callable<Integer>,
this.emitterName = emitterName;
}
- public String getEmitterName() {
- return emitterName;
- }
-
@Field
public void setMaxWaitMs(long maxWaitMs) {
this.maxWaitMs = maxWaitMs;
@@ -84,6 +90,10 @@ public abstract class FetchIterator implements Callable<Integer>,
this.queueSize = queueSize;
}
+ public FetchEmitTuple.ON_PARSE_EXCEPTION getOnParseException() {
+ return onParseException;
+ }
+
@Field
public void setOnParseException(String onParseException) throws TikaConfigException {
if ("skip".equalsIgnoreCase(onParseException)) {
@@ -99,10 +109,6 @@ public abstract class FetchIterator implements Callable<Integer>,
this.onParseException = onParseException;
}
- public FetchEmitTuple.ON_PARSE_EXCEPTION getOnParseException() {
- return onParseException;
- }
-
public Integer call() throws Exception {
enqueue();
tryToAdd(COMPLETED_SEMAPHORE);
@@ -168,19 +174,19 @@ public abstract class FetchIterator implements Callable<Integer>,
FetchEmitTuple t = null;
long start = System.currentTimeMillis();
try {
- long elapsed = System.currentTimeMillis()-start;
+ long elapsed = System.currentTimeMillis() - start;
while (t == null && elapsed < maxWaitMs) {
checkThreadOk();
t = queue.poll(100, TimeUnit.MILLISECONDS);
- elapsed = System.currentTimeMillis()-start;
+ elapsed = System.currentTimeMillis() - start;
}
} catch (InterruptedException e) {
LOGGER.warn("interrupted");
return COMPLETED_SEMAPHORE;
}
if (t == null) {
- throw new TikaTimeoutException("waited longer than "+
- maxWaitMs+"ms for the next tuple");
+ throw new TikaTimeoutException(
+ "waited longer than " + maxWaitMs + "ms for the next tuple");
}
return t;
}
diff --git a/tika-server/tika-server-client/src/main/java/org/apache/tika/server/client/TikaClient.java b/tika-server/tika-server-client/src/main/java/org/apache/tika/server/client/TikaClient.java
index 3b1fb25..a9f7319 100644
--- a/tika-server/tika-server-client/src/main/java/org/apache/tika/server/client/TikaClient.java
+++ b/tika-server/tika-server-client/src/main/java/org/apache/tika/server/client/TikaClient.java
@@ -31,7 +31,7 @@ import org.apache.tika.pipes.fetchiterator.FetchEmitTuple;
public class TikaClient {
private final Random random = new Random();
- private List<TikaHttpClient> clients;
+ private final List<TikaHttpClient> clients;
private TikaClient(List<TikaHttpClient> clients) {
diff --git a/tika-server/tika-server-client/src/main/java/org/apache/tika/server/client/TikaClientCLI.java b/tika-server/tika-server-client/src/main/java/org/apache/tika/server/client/TikaClientCLI.java
index b959839..5e583db 100644
--- a/tika-server/tika-server-client/src/main/java/org/apache/tika/server/client/TikaClientCLI.java
+++ b/tika-server/tika-server-client/src/main/java/org/apache/tika/server/client/TikaClientCLI.java
@@ -45,8 +45,9 @@ import org.apache.tika.pipes.fetchiterator.FetchIterator;
public class TikaClientCLI {
private static final Logger LOGGER = LoggerFactory.getLogger(TikaClientCLI.class);
+ private static final int QUEUE_SIZE = 10000;
- private long maxWaitMs = 300000;
+ private final long maxWaitMs = 300000;
public static void main(String[] args) throws Exception {
//TODO -- add an actual commandline,
@@ -65,9 +66,10 @@ public class TikaClientCLI {
ExecutorCompletionService<Integer> completionService =
new ExecutorCompletionService<>(executorService);
final FetchIterator fetchIterator = config.getFetchIterator();
- final ArrayBlockingQueue<FetchEmitTuple> queue = fetchIterator.init(numThreads);
+ final ArrayBlockingQueue<FetchEmitTuple> queue =
+ new ArrayBlockingQueue<>(QUEUE_SIZE);
- completionService.submit(fetchIterator);
+ completionService.submit(new FetchIteratorWrapper(fetchIterator, queue, numThreads));
if (tikaServerUrls.size() == numThreads) {
logDiffSizes(tikaServerUrls.size(), numThreads);
for (int i = 0; i < numThreads; i++) {
@@ -179,4 +181,31 @@ public class TikaClientCLI {
}
}
}
+
+ private class FetchIteratorWrapper implements Callable<Integer> {
+ private final FetchIterator fetchIterator;
+ private final ArrayBlockingQueue<FetchEmitTuple> queue;
+ private final int numThreads;
+
+ public FetchIteratorWrapper(FetchIterator fetchIterator,
+ ArrayBlockingQueue<FetchEmitTuple> queue,
+ int numThreads) {
+ this.fetchIterator = fetchIterator;
+ this.queue = queue;
+ this.numThreads = numThreads;
+
+ }
+
+ @Override
+ public Integer call() throws Exception {
+ for (FetchEmitTuple t : fetchIterator) {
+ //potentially blocks forever
+ queue.put(t);
+ }
+ for (int i = 0; i < numThreads; i ++) {
+ queue.put(FetchIterator.COMPLETED_SEMAPHORE);
+ }
+ return 1;
+ }
+ }
}
diff --git a/tika-server/tika-server-client/src/main/java/org/apache/tika/server/client/TikaEmitterResult.java b/tika-server/tika-server-client/src/main/java/org/apache/tika/server/client/TikaEmitterResult.java
index 1438885..4cb134c 100644
--- a/tika-server/tika-server-client/src/main/java/org/apache/tika/server/client/TikaEmitterResult.java
+++ b/tika-server/tika-server-client/src/main/java/org/apache/tika/server/client/TikaEmitterResult.java
@@ -19,9 +19,10 @@ package org.apache.tika.server.client;
public class TikaEmitterResult {
- private STATUS status;
- private String msg;//used for exceptions. will be null for status ok
- private long timeElapsed;
+ private final STATUS status;
+ private final String msg;//used for exceptions. will be null for status ok
+ private final long timeElapsed;
+
public TikaEmitterResult(STATUS status, long timeElapsed, String msg) {
this.status = status;
this.timeElapsed = timeElapsed;
diff --git a/tika-server/tika-server-client/src/main/java/org/apache/tika/server/client/TikaHttpClient.java b/tika-server/tika-server-client/src/main/java/org/apache/tika/server/client/TikaHttpClient.java
index 6d6517f..9a5ff7f 100644
--- a/tika-server/tika-server-client/src/main/java/org/apache/tika/server/client/TikaHttpClient.java
+++ b/tika-server/tika-server-client/src/main/java/org/apache/tika/server/client/TikaHttpClient.java
@@ -48,11 +48,11 @@ class TikaHttpClient {
private final String emitEndPointUrl;
private final String asyncEndPointUrl;
private final String tikaUrl;
- private int maxRetries = 3;
+ private final int maxRetries = 3;
//if can't make contact with Tika server, max wait time in ms
- private long maxWaitForTikaMs = 120000;
+ private final long maxWaitForTikaMs = 120000;
//how often to ping /tika (in ms) to see if the server is up and running
- private long pulseWaitForTikaMs = 1000;
+ private final long pulseWaitForTikaMs = 1000;
/**
* @param baseUrl url to base endpoint
diff --git a/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/resource/AsyncEmitter.java b/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/resource/AsyncEmitter.java
index 1ebe787..054a1d2 100644
--- a/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/resource/AsyncEmitter.java
+++ b/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/resource/AsyncEmitter.java
@@ -97,7 +97,7 @@ public class AsyncEmitter implements Callable<Integer> {
void add(EmitData data) {
size++;
long sz = AbstractEmitter
- .estimateSizeInBytes(data.getEmitKey().getKey(), data.getMetadataList());
+ .estimateSizeInBytes(data.getEmitKey().getEmitKey(), data.getMetadataList());
if (estimatedSize + sz > maxBytes) {
LOG.debug("estimated size ({}) > maxBytes({}), going to emitAll",
(estimatedSize + sz), maxBytes);
diff --git a/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/resource/EmitterResource.java b/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/resource/EmitterResource.java
index 56a71db..306732a 100644
--- a/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/resource/EmitterResource.java
+++ b/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/resource/EmitterResource.java
@@ -64,18 +64,13 @@ public class EmitterResource {
private static final String FETCH_KEY_ABBREV = "fk";
private static final String EMIT_KEY_ABBREV = "ek";
- /**
- * key that is safe to pass through http header.
- * The user _must_ specify this for the fsemitter if calling 'put'
- */
- public static final String EMIT_KEY_FOR_HTTP_HEADER = "emit-key";
private static final Logger LOG = LoggerFactory.getLogger(EmitterResource.class);
static EmitKey calcEmitKey(FetchEmitTuple t) {
//use fetch key if emitter key is not specified
//TODO: clean this up?
EmitKey emitKey = t.getEmitKey();
- if (StringUtils.isBlank(emitKey.getKey())) {
+ if (StringUtils.isBlank(emitKey.getEmitKey())) {
emitKey = new EmitKey(emitKey.getEmitterName(), t.getFetchKey().getKey());
}
return emitKey;
@@ -197,21 +192,11 @@ public class EmitterResource {
return emit(calcEmitKey(t), metadataList);
}
- static EmitKey calcEmitKey(FetchEmitTuple t) {
- //use fetch key if emitter key is not specified
- //TODO: clean this up?
- EmitKey emitKey = t.getEmitKey();
- if (StringUtils.isBlank(emitKey.getKey())) {
- emitKey = new EmitKey(emitKey.getEmitterName(), t.getFetchKey().getKey());
- }
- return emitKey;
- }
-
private Map<String, String> skip(FetchEmitTuple t, List<Metadata> metadataList) {
Map<String, String> statusMap = new HashMap<>();
statusMap.put("status", "ok");
statusMap.put("emitter", t.getEmitKey().getEmitterName());
- statusMap.put("emitKey", t.getEmitKey().getKey());
+ statusMap.put("emitKey", t.getEmitKey().getEmitKey());
String msg = metadataList.get(0).get(TikaCoreProperties.CONTAINER_EXCEPTION);
statusMap.put("parse_exception", msg);
return statusMap;
@@ -269,9 +254,9 @@ public class EmitterResource {
String status = "ok";
String exceptionMsg = "";
try {
- emitter.emit(emitKey.getKey(), metadataList);
+ emitter.emit(emitKey.getEmitKey(), metadataList);
} catch (IOException | TikaEmitterException e) {
- LOG.warn("problem emitting (" + emitKey.getKey() + ")", e);
+ LOG.warn("problem emitting (" + emitKey.getEmitterName() + ")", e);
status = "emitter_exception";
exceptionMsg = ExceptionUtils.getStackTrace(e);
}