Skip to content

Commit 1ae7e46

Browse files
authored
TIKA-4560 -- clean up emit strategy (#2429)
1 parent 5aa5eb7 commit 1ae7e46

27 files changed

Lines changed: 371 additions & 82 deletions

File tree

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

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -43,7 +43,6 @@
4343
"queueSize": 10000,
4444
"numEmitters": 1,
4545
"emitIntermediateResults": false,
46-
"directEmitThresholdBytes": 100000,
4746
"timeoutMillis": 60000,
4847
"startupTimeoutMillis": 240000,
4948
"sleepOnStartupTimeoutMillis": 240000,
@@ -56,7 +55,11 @@
5655
"-Xmx1g",
5756
"-XX:+UseG1GC"
5857
],
59-
"javaPath": "java"
58+
"javaPath": "java",
59+
"emitStrategy": {
60+
"type": "DYNAMIC",
61+
"thresholdBytes": 100000
62+
}
6063
},
6164
"plugin-roots": "PLUGIN_ROOTS"
6265
}

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

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -94,7 +94,6 @@
9494
}
9595
},
9696
"pipes": {
97-
"directEmitThresholdBytes": 10000,
9897
"emitMaxEstimatedBytes": 100000,
9998
"emitWithinMillis": 10,
10099
"numEmitters": 1,
@@ -105,7 +104,11 @@
105104
"-XX:+ExitOnOutOfMemoryError",
106105
"-Dlog4j.configurationFile=LOG4J_PROPERTIES_FILE"
107106
],
108-
"timeoutMillis": 60000
107+
"timeoutMillis": 60000,
108+
"emitStrategy": {
109+
"type": "DYNAMIC",
110+
"thresholdBytes": 10000
111+
}
109112
},
110113
"plugin-roots": "target/plugins"
111114
}

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

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -60,7 +60,10 @@
6060
}
6161
},
6262
"pipes": {
63-
"directEmitThresholdBytes": 10000,
63+
"emitStrategy": {
64+
"type": "DYNAMIC",
65+
"thresholdBytes": 10000
66+
},
6467
"emitMaxEstimatedBytes": 100000,
6568
"emitWithinMillis": 60000,
6669
"numEmitters": 1,

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

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -104,7 +104,10 @@
104104
}
105105
},
106106
"pipes": {
107-
"directEmitThresholdBytes": 10000,
107+
"emitStrategy": {
108+
"type": "DYNAMIC",
109+
"thresholdBytes": 10000
110+
},
108111
"emitMaxEstimatedBytes": 100000,
109112
"emitWithinMillis": 60000,
110113
"numEmitters": 1,

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

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -61,7 +61,6 @@
6161
}
6262
},
6363
"pipes": {
64-
"directEmitThresholdBytes": 10000,
6564
"emitMaxEstimatedBytes": 100000,
6665
"emitWithinMillis": 10,
6766
"numEmitters": 1,
@@ -72,7 +71,11 @@
7271
"-XX:+ExitOnOutOfMemoryError",
7372
"-Dlog4j.configurationFile={LOG4J_PROPERTIES_FILE}"
7473
],
75-
"timeoutMillis": 60000
74+
"timeoutMillis": 60000,
75+
"emitStrategy": {
76+
"type": "DYNAMIC",
77+
"thresholdBytes": 10000
78+
}
7679
},
7780
"plugin-roots": "target/plugins"
7881
}

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

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -91,7 +91,10 @@
9191
}
9292
},
9393
"pipes": {
94-
"directEmitThresholdBytes": 10000,
94+
"emitStrategy": {
95+
"type": "DYNAMIC",
96+
"thresholdBytes": 10000
97+
},
9598
"emitMaxEstimatedBytes": 100000,
9699
"emitWithinMillis": 10,
97100
"numEmitters": 1,

tika-integration-tests/tika-pipes-solr-integration-tests/src/test/resources/tika-config-solr-urls.json

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -44,7 +44,6 @@
4444
}
4545
],
4646
"pipes": {
47-
"directEmitThresholdBytes": 10000,
4847
"emitMaxEstimatedBytes": 100000,
4948
"emitWithinMillis": 10,
5049
"numEmitters": 1,
@@ -55,7 +54,11 @@
5554
"-XX:+ExitOnOutOfMemoryError",
5655
"-Dlog4j.configurationFile={LOG4J_PROPERTIES_FILE}"
5756
],
58-
"timeoutMillis": 60000
57+
"timeoutMillis": 60000,
58+
"emitStrategy": {
59+
"type": "DYNAMIC",
60+
"thresholdBytes": 10000
61+
}
5962
},
6063
"fetchers": {
6164
"fsf": {

tika-pipes/tika-async-cli/src/test/resources/configs/config-template.json

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -29,7 +29,6 @@
2929
"queueSize": 10000,
3030
"numEmitters": 1,
3131
"emitIntermediateResults": false,
32-
"directEmitThresholdBytes": 100000,
3332
"timeoutMillis": 60000,
3433
"startupTimeoutMillis": 240000,
3534
"sleepOnStartupTimeoutMillis": 240000,
@@ -42,7 +41,11 @@
4241
"-Xmx1g",
4342
"-XX:+UseG1GC"
4443
],
45-
"javaPath": "java"
44+
"javaPath": "java",
45+
"emitStrategy": {
46+
"type": "DYNAMIC",
47+
"thresholdBytes": 100000
48+
}
4649
},
4750
"plugin-roots": "PLUGIN_ROOTS"
4851
}
Lines changed: 31 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,31 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright ownership.
5+
* The ASF licenses this file to You under the Apache License, Version 2.0
6+
* (the "License"); you may not use this file except in compliance with
7+
* the License. You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
package org.apache.tika.pipes.core;
18+
19+
/**
20+
* Strategy for how the forked PipesServer handles emitting data.
21+
* <ul>
22+
* <li>EMIT_ALL: Always emit directly from PipesServer (never pass back to client)</li>
23+
* <li>PASSBACK_ALL: Always pass back to client for batch emitting (never emit directly)</li>
24+
* <li>DYNAMIC: Emit directly if size >= threshold, otherwise pass back to client</li>
25+
* </ul>
26+
*/
27+
public enum EmitStrategy {
28+
EMIT_ALL,
29+
PASSBACK_ALL,
30+
DYNAMIC
31+
}
Lines changed: 130 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,130 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright ownership.
5+
* The ASF licenses this file to You under the Apache License, Version 2.0
6+
* (the "License"); you may not use this file except in compliance with
7+
* the License. You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
package org.apache.tika.pipes.core;
18+
19+
import org.apache.tika.exception.TikaConfigException;
20+
import org.apache.tika.exception.TikaException;
21+
22+
/**
23+
* Configuration for emit strategy in PipesConfig.
24+
* <p>
25+
* Example JSON configuration:
26+
* <pre>
27+
* {
28+
* "pipes": {
29+
* "emitStrategy": {
30+
* "type": "DYNAMIC",
31+
* "thresholdBytes": 100000
32+
* }
33+
* }
34+
* }
35+
* </pre>
36+
* Or for simpler strategies:
37+
* <pre>
38+
* {
39+
* "pipes": {
40+
* "emitStrategy": {
41+
* "type": "EMIT_ALL"
42+
* }
43+
* }
44+
* }
45+
* </pre>
46+
*/
47+
public class EmitStrategyConfig {
48+
49+
/**
50+
* Default emit strategy for PipesServer.
51+
* DYNAMIC means the strategy is determined by directEmitThresholdBytes.
52+
*/
53+
public static final EmitStrategy DEFAULT_EMIT_STRATEGY = EmitStrategy.DYNAMIC;
54+
55+
/**
56+
* Default threshold in bytes for direct emission from PipesServer.
57+
* If an extract is larger than this, it will be emitted
58+
* directly from the forked PipesServer rather than passed back to PipesClient.
59+
* Only used when emitStrategy is DYNAMIC.
60+
*/
61+
public static final long DEFAULT_DIRECT_EMIT_THRESHOLD_BYTES = 100000;
62+
63+
private EmitStrategy type = DEFAULT_EMIT_STRATEGY;
64+
private Long thresholdBytes = null;
65+
66+
public EmitStrategyConfig() {
67+
}
68+
69+
public EmitStrategyConfig(EmitStrategy type) {
70+
this.type = type;
71+
if (type == EmitStrategy.DYNAMIC) {
72+
thresholdBytes = DEFAULT_DIRECT_EMIT_THRESHOLD_BYTES;
73+
}
74+
}
75+
76+
public EmitStrategyConfig(EmitStrategy type, Long thresholdBytes) throws TikaException {
77+
this.type = type;
78+
this.thresholdBytes = thresholdBytes;
79+
validate();
80+
}
81+
82+
/**
83+
* Get the emit strategy type.
84+
*
85+
* @return the emit strategy
86+
*/
87+
public EmitStrategy getType() {
88+
return type;
89+
}
90+
91+
/**
92+
* Set the emit strategy type.
93+
*
94+
* @param type the emit strategy
95+
*/
96+
public void setType(EmitStrategy type) throws TikaConfigException {
97+
this.type = type;
98+
validate();
99+
}
100+
101+
/**
102+
* Get the threshold in bytes for DYNAMIC strategy.
103+
* Only applicable when type is DYNAMIC.
104+
*
105+
* @return the threshold in bytes, or null to use default
106+
*/
107+
public Long getThresholdBytes() {
108+
return thresholdBytes;
109+
}
110+
111+
/**
112+
* Set the threshold in bytes for DYNAMIC strategy.
113+
* Only applicable when type is DYNAMIC.
114+
*
115+
* @param thresholdBytes the threshold in bytes
116+
*/
117+
public void setThresholdBytes(Long thresholdBytes) throws TikaConfigException {
118+
this.thresholdBytes = thresholdBytes;
119+
validate();
120+
}
121+
122+
private void validate() throws TikaConfigException {
123+
if (thresholdBytes != null &&
124+
(type == EmitStrategy.EMIT_ALL || type == EmitStrategy.PASSBACK_ALL)) {
125+
throw new TikaConfigException(
126+
"thresholdBytes cannot be set for emit strategy type " + type +
127+
". thresholdBytes is only applicable for DYNAMIC strategy.");
128+
}
129+
}
130+
}

0 commit comments

Comments
 (0)