Skip to content

Commit 2df3069

Browse files
tballisontballison
authored
* TIKA-4556 -- refactor client/server protocol in tika-pipes
1 parent 200f224 commit 2df3069

60 files changed

Lines changed: 2881 additions & 1591 deletions

File tree

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

‎tika-app/src/test/resources/configs/config-template.json‎

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -43,7 +43,7 @@
4343
"queueSize": 10000,
4444
"numEmitters": 1,
4545
"emitIntermediateResults": false,
46-
"maxForEmitBatchBytes": 100000,
46+
"directEmitThresholdBytes": 100000,
4747
"timeoutMillis": 60000,
4848
"startupTimeoutMillis": 240000,
4949
"sleepOnStartupTimeoutMillis": 240000,
@@ -53,7 +53,6 @@
5353
"staleFetcherTimeoutSeconds": 600,
5454
"staleFetcherDelaySeconds": 60,
5555
"forkedJvmArgs": ["-Xmx1g", "-XX:+UseG1GC"],
56-
"tikaConfig": "TIKA_CONFIG",
5756
"javaPath": "java"
5857
},
5958
"plugin-roots": "PLUGIN_ROOTS"

‎tika-grpc/src/main/java/org/apache/tika/pipes/grpc/TikaGrpcServerImpl.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -244,7 +244,7 @@ private void fetchAndParseImpl(FetchAndParseRequest request,
244244
FetchAndParseReply.newBuilder()
245245
.setFetchKey(request.getFetchKey())
246246
.setStatus(pipesResult.status().name());
247-
if (pipesResult.status().equals(PipesResult.STATUS.FETCH_EXCEPTION)) {
247+
if (pipesResult.status().equals(PipesResult.RESULT_STATUS.FETCH_EXCEPTION)) {
248248
fetchReplyBuilder.setErrorMessage(pipesResult.message());
249249
}
250250
if (pipesResult.emitData() != null && pipesResult.emitData().getMetadataList() != null) {

‎tika-grpc/src/test/java/org/apache/tika/pipes/grpc/TikaGrpcServerTest.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -236,7 +236,7 @@ public void testBiStream(Resources resources) throws Exception {
236236
@Override
237237
public void onNext(FetchAndParseReply fetchAndParseReply) {
238238
LOG.debug("Fetched {} with metadata {}", fetchAndParseReply.getFetchKey(), fetchAndParseReply.getFieldsMap());
239-
if (PipesResult.STATUS.FETCH_EXCEPTION.name().equals(fetchAndParseReply.getStatus())) {
239+
if (PipesResult.RESULT_STATUS.FETCH_EXCEPTION.name().equals(fetchAndParseReply.getStatus())) {
240240
errors.add(fetchAndParseReply);
241241
} else {
242242
successes.add(fetchAndParseReply);

‎tika-grpc/src/test/resources/tika-pipes-test-config.xml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -27,7 +27,7 @@
2727
<arg>-XX:ParallelGCThreads=2</arg>
2828
</forkedJvmArgs>
2929
<timeoutMillis>60000</timeoutMillis>
30-
<maxForEmitBatchBytes>-1</maxForEmitBatchBytes> <!-- disable emit -->
30+
<directEmitThresholdBytes>-1</directEmitThresholdBytes> <!-- disable emit -->
3131
</params>
3232
</pipes>
3333
<fetchers>

‎tika-integration-tests/tika-pipes-kafka-integration-tests/src/test/resources/kafka/plugins-template.json‎

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -94,12 +94,11 @@
9494
}
9595
},
9696
"pipes": {
97-
"maxForEmitBatchBytes": 10000,
97+
"directEmitThresholdBytes": 10000,
9898
"emitMaxEstimatedBytes": 100000,
9999
"emitWithinMillis": 10,
100100
"numEmitters": 1,
101101
"numClients": 1,
102-
"tikaConfig": "TIKA_CONFIG",
103102
"forkedJvmArgs": [
104103
"-Xmx1g",
105104
"-XX:ParallelGCThreads=2",

‎tika-integration-tests/tika-pipes-kafka-integration-tests/src/test/resources/tika-config-kafka.xml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -66,7 +66,7 @@
6666
</metadataFilters>
6767
<async>
6868
<params>
69-
<maxForEmitBatchBytes>10000</maxForEmitBatchBytes>
69+
<directEmitThresholdBytes>10000</directEmitThresholdBytes>
7070
<emitMaxEstimatedBytes>100000</emitMaxEstimatedBytes>
7171
<emitWithinMillis>10</emitWithinMillis>
7272
<numEmitters>1</numEmitters>

‎tika-integration-tests/tika-pipes-opensearch-integration-tests/src/test/resources/opensearch/plugins-template.json‎

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -60,12 +60,11 @@
6060
}
6161
},
6262
"pipes": {
63-
"maxForEmitBatchBytes": 10000,
63+
"directEmitThresholdBytes": 10000,
6464
"emitMaxEstimatedBytes": 100000,
6565
"emitWithinMillis": 60000,
6666
"numEmitters": 1,
6767
"numClients": 3,
68-
"tikaConfig": "TIKA_CONFIG",
6968
"forkedJvmArgs": [
7069
"-Xmx512m",
7170
"-XX:ParallelGCThreads=2",

‎tika-integration-tests/tika-pipes-opensearch-integration-tests/src/test/resources/opensearch/tika-config-opensearch.json‎

Lines changed: 1 addition & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -101,13 +101,11 @@
101101
}
102102
},
103103
"pipes": {
104-
"maxForEmitBatchBytes": 10000,
104+
"directEmitThresholdBytes": 10000,
105105
"emitMaxEstimatedBytes": 100000,
106106
"emitWithinMillis": 60000,
107107
"numEmitters": 1,
108108
"numClients": 3,
109-
"tikaConfig": "TIKA_CONFIG",
110-
"pipesPluginsConfig": "PLUGINS_CONFIG",
111109
"forkedJvmArgs": [
112110
"-Xmx512m",
113111
"-XX:ParallelGCThreads=2",

‎tika-integration-tests/tika-pipes-s3-integration-tests/src/test/resources/s3/plugins-template.json‎

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -56,12 +56,11 @@
5656
}
5757
},
5858
"pipes": {
59-
"maxForEmitBatchBytes": 10000,
59+
"directEmitThresholdBytes": 10000,
6060
"emitMaxEstimatedBytes": 100000,
6161
"emitWithinMillis": 10,
6262
"numEmitters": 1,
6363
"numClients": 1,
64-
"tikaConfig": "{TIKA_CONFIG}",
6564
"forkedJvmArgs": [
6665
"-Xmx1g",
6766
"-XX:ParallelGCThreads=2",

‎tika-integration-tests/tika-pipes-s3-integration-tests/src/test/resources/tika-config-s3-integration-test.xml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -63,7 +63,7 @@
6363
</metadataFilter>
6464
</metadataFilters>
6565
<async>
66-
<maxForEmitBatchBytes>10000</maxForEmitBatchBytes>
66+
<directEmitThresholdBytes>10000</directEmitThresholdBytes>
6767
<emitMaxEstimatedBytes>100000</emitMaxEstimatedBytes>
6868
<emitWithinMillis>10</emitWithinMillis>
6969
<numEmitters>1</numEmitters>

0 commit comments

Comments
 (0)