Skip to content

Commit 05267dd

Browse files
tballisontballison
authored
TIKA-4848 pipes, step 3 (#3087)
1 parent b7e329a commit 05267dd

13 files changed

Lines changed: 248 additions & 24 deletions

File tree

‎CHANGES.txt‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,9 @@
11
Release 4.1.0 - unreleased
22

3+
* The exception-reporting policy now also governs the messages a pipes
4+
worker returns (fetch/emit/crash) and the container exception it
5+
records; part of TIKA-4848 step 3 (TIKA-4848).
6+
37
* Allow image compression settings in PDFBox-based renderer (TIKA-4862).
48

59
* The tika-server full and tika-grpc Docker images set OMP_THREAD_LIMIT=1:

‎tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/ConnectionHandler.java‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -105,7 +105,8 @@ public ConnectionHandler(Socket socket, SharedServerResources resources, PipesCo
105105
this.resources = resources;
106106
this.pipesConfig = pipesConfig;
107107
this.heartbeatIntervalMillis = pipesConfig.getHeartbeatIntervalMillis();
108-
this.protocolIO = new ServerProtocolIO(input, output, pipesConfig.getMaxIpcPayloadBytes());
108+
this.protocolIO = new ServerProtocolIO(input, output, pipesConfig.getMaxIpcPayloadBytes(),
109+
resources.getExceptionReporting());
109110
}
110111

111112
@Override

‎tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/EmitHandler.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -161,7 +161,7 @@ private PipesResult emit(String taskId, EmitKey emitKey,
161161
return new PipesResult(PipesResult.RESULT_STATUS.EMITTER_NOT_FOUND, noEmitterMsg);
162162
} catch (IOException | TikaException e) {
163163
LOG.warn("Couldn't initialize emitter for task id '" + taskId + "'", e);
164-
return new PipesResult(PipesResult.RESULT_STATUS.EMITTER_INITIALIZATION_EXCEPTION, ExceptionUtils.getStackTrace(e));
164+
return new PipesResult(PipesResult.RESULT_STATUS.EMITTER_INITIALIZATION_EXCEPTION, ExceptionUtils.format(e, parseContext));
165165
}
166166
try {
167167
ParseMode parseMode = parseContext.get(ParseMode.class);
@@ -175,7 +175,7 @@ private PipesResult emit(String taskId, EmitKey emitKey,
175175
}
176176
} catch (IOException e) {
177177
LOG.warn("emit exception", e);
178-
String msg = ExceptionUtils.getStackTrace(e);
178+
String msg = ExceptionUtils.format(e, parseContext);
179179
//for now, we're hiding the parse exception if there was also an emit exception
180180
return new PipesResult(PipesResult.RESULT_STATUS.EMIT_EXCEPTION, msg);
181181
}

‎tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/FetchHandler.java‎

Lines changed: 5 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -46,7 +46,7 @@ public FetchHandler(FetcherManager fetcherManager) {
4646
}
4747

4848
public TisOrResult fetch(FetchEmitTuple fetchEmitTuple, Metadata metadata, ParseContext parseContext) {
49-
FetcherOrResult fetcherResult = getFetcher(fetchEmitTuple);
49+
FetcherOrResult fetcherResult = getFetcher(fetchEmitTuple, parseContext);
5050
if (fetcherResult.pipesResult != null) {
5151
return new TisOrResult(null, fetcherResult.pipesResult);
5252
}
@@ -55,11 +55,12 @@ public TisOrResult fetch(FetchEmitTuple fetchEmitTuple, Metadata metadata, Parse
5555
fetchEmitTuple.getFetchKey().getFetchKey(), metadata, parseContext);
5656
return new TisOrResult(tis, null);
5757
} catch (IOException | TikaException e) {
58-
return new TisOrResult(null, new PipesResult(PipesResult.RESULT_STATUS.FETCH_EXCEPTION, ExceptionUtils.getStackTrace(e)));
58+
return new TisOrResult(null, new PipesResult(PipesResult.RESULT_STATUS.FETCH_EXCEPTION,
59+
ExceptionUtils.format(e, parseContext)));
5960
}
6061
}
6162

62-
private FetcherOrResult getFetcher(FetchEmitTuple t) {
63+
private FetcherOrResult getFetcher(FetchEmitTuple t, ParseContext parseContext) {
6364
String fetcherId = t.getFetchKey().getFetcherId();
6465
// Built in, not configured: the bytes come with the request, so there is nothing for an
6566
// operator to point at and no reason for it to occupy an id in the ConfigStore.
@@ -75,7 +76,7 @@ private FetcherOrResult getFetcher(FetchEmitTuple t) {
7576
} catch (IOException | TikaException e) {
7677
LOG.warn("Couldn't initialize fetcher for fetch id={}", t.getId(), e);
7778
return new FetcherOrResult(null, new PipesResult(PipesResult.RESULT_STATUS.FETCHER_INITIALIZATION_EXCEPTION,
78-
ExceptionUtils.getStackTrace(e)));
79+
ExceptionUtils.format(e, parseContext)));
7980
}
8081
}
8182

‎tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/ParseHandler.java‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -244,19 +244,19 @@ public List<Metadata> parseConcatenated(FetchEmitTuple fetchEmitTuple,
244244
try {
245245
autoDetectParser.parse(stream, handler, metadata, parseContext);
246246
} catch (SAXException e) {
247-
containerException = ExceptionUtils.getStackTrace(e);
247+
containerException = ExceptionUtils.format(e, parseContext);
248248
LOG.warn("sax problem:" + fetchEmitTuple.getId(), e);
249249
if (WriteLimitReachedException.isWriteLimitReached(e)) {
250250
writeLimitReached = true;
251251
}
252252
} catch (EncryptedDocumentException e) {
253-
containerException = ExceptionUtils.getStackTrace(e);
253+
containerException = ExceptionUtils.format(e, parseContext);
254254
LOG.warn("encrypted document:" + fetchEmitTuple.getId(), e);
255255
} catch (SecurityException e) {
256256
LOG.warn("security exception:" + fetchEmitTuple.getId(), e);
257257
throw e;
258258
} catch (Exception e) {
259-
containerException = ExceptionUtils.getStackTrace(e);
259+
containerException = ExceptionUtils.format(e, parseContext);
260260
LOG.warn("parse exception: " + fetchEmitTuple.getId(), e);
261261
} finally {
262262
// BasicContentHandlerFactory's "ignore" handler's toString() returns "" (not

‎tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/PipesServer.java‎

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -45,6 +45,7 @@
4545
import org.slf4j.LoggerFactory;
4646
import org.xml.sax.SAXException;
4747

48+
import org.apache.tika.config.ExceptionReporting;
4849
import org.apache.tika.config.ParseTimeout;
4950
import org.apache.tika.config.TimeoutLimits;
5051
import org.apache.tika.config.loader.TikaJsonConfig;
@@ -218,7 +219,8 @@ public static PipesServer load(int port, Path tikaConfigPath) throws Exception {
218219
} catch (Exception e) {
219220
LOG.error("Failed to start up", e);
220221
try {
221-
String msg = ExceptionUtils.getStackTrace(e);
222+
// Config may be what failed to load, so no ExceptionReporting policy is available.
223+
String msg = ExceptionUtils.format(e, ExceptionReporting.DEFAULT);
222224
byte[] bytes = msg.getBytes(StandardCharsets.UTF_8);
223225
PipesMessage.startupFailed(bytes).write(dos);
224226
// pipesConfig may not have loaded successfully (that may be why we're
@@ -252,7 +254,8 @@ public PipesServer(String pipesClientId, TikaLoader tikaLoader, PipesConfig pipe
252254
validateHeartbeatInterval(pipesConfig);
253255

254256
emitStrategy = pipesConfig.getEmitStrategy().getType();
255-
this.protocolIO = new ServerProtocolIO(input, output, pipesConfig.getMaxIpcPayloadBytes());
257+
this.protocolIO = new ServerProtocolIO(input, output, pipesConfig.getMaxIpcPayloadBytes(),
258+
ExceptionReporting.get(tikaLoader.loadParseContext()));
256259
}
257260

258261

‎tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/PipesWorker.java‎

Lines changed: 7 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -198,7 +198,7 @@ private PipesResult zipAndEmitEmbeddedFiles(TempFileUnpackHandler tempHandler) {
198198
} catch (IOException e) {
199199
LOG.warn("Failed to create zip file", e);
200200
return new PipesResult(PipesResult.RESULT_STATUS.EMIT_EXCEPTION,
201-
"Failed to create zip file: " + ExceptionUtils.getStackTrace(e));
201+
"Failed to create zip file: " + ExceptionUtils.format(e, parseContext));
202202
}
203203

204204
// Emit the zip file
@@ -208,7 +208,7 @@ private PipesResult zipAndEmitEmbeddedFiles(TempFileUnpackHandler tempHandler) {
208208
} catch (IOException e) {
209209
LOG.warn("Failed to emit zip file", e);
210210
return new PipesResult(PipesResult.RESULT_STATUS.EMIT_EXCEPTION,
211-
"Failed to emit zip file: " + ExceptionUtils.getStackTrace(e));
211+
"Failed to emit zip file: " + ExceptionUtils.format(e, parseContext));
212212
}
213213

214214
LOG.debug("Successfully zipped and emitted {} embedded files to {}",
@@ -289,7 +289,7 @@ private PipesResult emitFrictionlessZipped(FrictionlessUnpackHandler frictionles
289289
} catch (IOException e) {
290290
LOG.warn("Failed to create Frictionless zip file", e);
291291
return new PipesResult(PipesResult.RESULT_STATUS.EMIT_EXCEPTION,
292-
"Failed to create Frictionless zip file: " + ExceptionUtils.getStackTrace(e));
292+
"Failed to create Frictionless zip file: " + ExceptionUtils.format(e, parseContext));
293293
}
294294

295295
// Emit the zip file
@@ -299,7 +299,7 @@ private PipesResult emitFrictionlessZipped(FrictionlessUnpackHandler frictionles
299299
} catch (IOException e) {
300300
LOG.warn("Failed to emit Frictionless zip file", e);
301301
return new PipesResult(PipesResult.RESULT_STATUS.EMIT_EXCEPTION,
302-
"Failed to emit Frictionless zip file: " + ExceptionUtils.getStackTrace(e));
302+
"Failed to emit Frictionless zip file: " + ExceptionUtils.format(e, parseContext));
303303
}
304304

305305
LOG.debug("Successfully emitted Frictionless package with {} resources to {}",
@@ -351,7 +351,7 @@ private PipesResult emitFrictionlessDirectory(FrictionlessUnpackHandler friction
351351
} catch (IOException e) {
352352
LOG.warn("Failed to emit Frictionless directory output", e);
353353
return new PipesResult(PipesResult.RESULT_STATUS.EMIT_EXCEPTION,
354-
"Failed to emit Frictionless directory output: " + ExceptionUtils.getStackTrace(e));
354+
"Failed to emit Frictionless directory output: " + ExceptionUtils.format(e, parseContext));
355355
}
356356

357357
LOG.debug("Successfully emitted Frictionless package with {} resources (directory mode) to {}",
@@ -484,7 +484,7 @@ protected ParseDataOrPipesResult parseFromTuple() throws TikaException, Interrup
484484
} catch (IOException e) {
485485
LOG.warn("fetcher initialization exception id={}", fetchEmitTuple.getId(), e);
486486
return new ParseDataOrPipesResult(null,
487-
new PipesResult(PipesResult.RESULT_STATUS.FETCHER_INITIALIZATION_EXCEPTION, ExceptionUtils.getStackTrace(e)));
487+
new PipesResult(PipesResult.RESULT_STATUS.FETCHER_INITIALIZATION_EXCEPTION, ExceptionUtils.format(e, parseContext)));
488488
}
489489
// Use newMetadata() to apply any configured write limits
490490
Metadata metadata = localContext.newMetadata();
@@ -503,7 +503,7 @@ protected ParseDataOrPipesResult parseFromTuple() throws TikaException, Interrup
503503
} catch (TikaException | IOException e) {
504504
LOG.warn("fetch exception id={}", fetchEmitTuple.getId(), e);
505505
return new ParseDataOrPipesResult(null,
506-
new PipesResult(PipesResult.RESULT_STATUS.UNSPECIFIED_CRASH, ExceptionUtils.getStackTrace(e)));
506+
new PipesResult(PipesResult.RESULT_STATUS.UNSPECIFIED_CRASH, ExceptionUtils.format(e, parseContext)));
507507
}
508508
}
509509

‎tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/ServerProtocolIO.java‎

Lines changed: 9 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,7 @@
2626
import org.slf4j.Logger;
2727
import org.slf4j.LoggerFactory;
2828

29+
import org.apache.tika.config.ExceptionReporting;
2930
import org.apache.tika.config.TimeoutLimits;
3031
import org.apache.tika.exception.TikaConfigException;
3132
import org.apache.tika.metadata.Metadata;
@@ -82,8 +83,15 @@ public class ServerProtocolIO {
8283
private final DataInputStream input;
8384
private final DataOutputStream output;
8485
private final int maxIpcPayloadBytes;
86+
private final ExceptionReporting exceptionReporting;
8587

8688
public ServerProtocolIO(DataInputStream input, DataOutputStream output, int maxIpcPayloadBytes) {
89+
this(input, output, maxIpcPayloadBytes, ExceptionReporting.DEFAULT);
90+
}
91+
92+
public ServerProtocolIO(DataInputStream input, DataOutputStream output, int maxIpcPayloadBytes,
93+
ExceptionReporting exceptionReporting) {
94+
this.exceptionReporting = exceptionReporting;
8795
if (maxIpcPayloadBytes < MIN_FALLBACK_PAYLOAD_BYTES) {
8896
throw new IllegalArgumentException(String.format(Locale.ROOT,
8997
"maxIpcPayloadBytes %d is below the minimum %d required to carry a PAYLOAD_LIMIT_EXCEEDED response",
@@ -196,7 +204,7 @@ public void writeIntermediate(Metadata metadata) throws IOException {
196204
* @throws IOException on serialization, I/O, or unexpected ACK response
197205
*/
198206
public void writeCrash(PipesMessageType crashType, Throwable t) throws IOException {
199-
String msg = (t != null) ? ExceptionUtils.getStackTrace(t) : "";
207+
String msg = (t != null) ? ExceptionUtils.format(t, exceptionReporting) : "";
200208
BoundedOutputStream bos = new BoundedOutputStream(maxIpcPayloadBytes);
201209
try {
202210
JsonPipesIpc.toStream(msg, bos);

‎tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/server/SharedServerResources.java‎

Lines changed: 13 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,7 @@
2020

2121
import org.xml.sax.SAXException;
2222

23+
import org.apache.tika.config.ExceptionReporting;
2324
import org.apache.tika.config.loader.TikaJsonConfig;
2425
import org.apache.tika.config.loader.TikaLoader;
2526
import org.apache.tika.detect.Detector;
@@ -64,14 +65,17 @@ public class SharedServerResources {
6465
private final MetadataWriteLimiterFactory defaultMetadataWriteLimiterFactory;
6566
private final EmitStrategy emitStrategy;
6667
private final ConfigStore configStore;
68+
private final ExceptionReporting exceptionReporting;
6769

6870
private SharedServerResources(TikaLoader tikaLoader, PipesConfig pipesConfig,
6971
AutoDetectParser autoDetectParser, Detector detector,
7072
RecursiveParserWrapper rMetaParser, FetcherManager fetcherManager,
7173
EmitterManager emitterManager, MetadataFilter defaultMetadataFilter,
7274
ContentHandlerFactory defaultContentHandlerFactory,
7375
MetadataWriteLimiterFactory defaultMetadataWriteLimiterFactory,
74-
EmitStrategy emitStrategy, ConfigStore configStore) {
76+
EmitStrategy emitStrategy, ConfigStore configStore,
77+
ExceptionReporting exceptionReporting) {
78+
this.exceptionReporting = exceptionReporting;
7579
this.tikaLoader = tikaLoader;
7680
this.pipesConfig = pipesConfig;
7781
this.autoDetectParser = autoDetectParser;
@@ -117,14 +121,16 @@ public static SharedServerResources load(TikaLoader tikaLoader, PipesConfig pipe
117121
// Load filters and factories
118122
MetadataFilter metadataFilter = tikaLoader.loadMetadataFilters();
119123
ContentHandlerFactory contentHandlerFactory = tikaLoader.loadContentHandlerFactory();
124+
ParseContext configContext = tikaLoader.loadParseContext();
120125
MetadataWriteLimiterFactory metadataWriteLimiterFactory =
121-
tikaLoader.loadParseContext().get(MetadataWriteLimiterFactory.class);
126+
configContext.get(MetadataWriteLimiterFactory.class);
122127

123128
EmitStrategy emitStrategy = pipesConfig.getEmitStrategy().getType();
124129

125130
return new SharedServerResources(tikaLoader, pipesConfig, autoDetectParser, detector,
126131
rMetaParser, fetcherManager, emitterManager, metadataFilter, contentHandlerFactory,
127-
metadataWriteLimiterFactory, emitStrategy, configStore);
132+
metadataWriteLimiterFactory, emitStrategy, configStore,
133+
ExceptionReporting.get(configContext));
128134
}
129135

130136
private static ConfigStore createConfigStore(PipesConfig pipesConfig, TikaPluginManager tikaPluginManager)
@@ -206,6 +212,10 @@ public EmitStrategy getEmitStrategy() {
206212
return emitStrategy;
207213
}
208214

215+
public ExceptionReporting getExceptionReporting() {
216+
return exceptionReporting;
217+
}
218+
209219
public ConfigStore getConfigStore() {
210220
return configStore;
211221
}

‎tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/server/ServerProtocolIOTest.java‎

Lines changed: 17 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,7 @@
1717
package org.apache.tika.pipes.core.server;
1818

1919
import static org.junit.jupiter.api.Assertions.assertEquals;
20+
import static org.junit.jupiter.api.Assertions.assertFalse;
2021
import static org.junit.jupiter.api.Assertions.assertNotNull;
2122
import static org.junit.jupiter.api.Assertions.assertTrue;
2223

@@ -29,6 +30,7 @@
2930

3031
import org.junit.jupiter.api.Test;
3132

33+
import org.apache.tika.config.ExceptionReporting;
3234
import org.apache.tika.config.TimeoutLimits;
3335
import org.apache.tika.metadata.Metadata;
3436
import org.apache.tika.parser.ParseContext;
@@ -293,6 +295,11 @@ void testOversizedMessageOnAlreadyEmittedStatusPreservesStatus() throws Exceptio
293295
* sends ACK, and returns the deserialized stack trace string.
294296
*/
295297
private String exchangeCrash(Throwable t, int maxPayloadBytes) throws Exception {
298+
return exchangeCrash(t, maxPayloadBytes, ExceptionReporting.DEFAULT);
299+
}
300+
301+
private String exchangeCrash(Throwable t, int maxPayloadBytes, ExceptionReporting reporting)
302+
throws Exception {
296303
PipedOutputStream serverOutPipe = new PipedOutputStream();
297304
PipedInputStream clientInPipe = new PipedInputStream(serverOutPipe, 1024 * 1024);
298305
PipedOutputStream clientOutPipe = new PipedOutputStream();
@@ -320,7 +327,7 @@ private String exchangeCrash(Throwable t, int maxPayloadBytes) throws Exception
320327
ServerProtocolIO io = new ServerProtocolIO(
321328
new DataInputStream(serverInPipe),
322329
new DataOutputStream(serverOutPipe),
323-
maxPayloadBytes);
330+
maxPayloadBytes, reporting);
324331
io.writeCrash(PipesMessageType.UNSPECIFIED_CRASH, t);
325332

326333
clientThread.join(5000);
@@ -358,4 +365,13 @@ void testCrashOversizedPayloadFallsBackToEmptyString() throws Exception {
358365
// Empty string fallback: the trace was too large, we get an empty payload.
359366
assertEquals("", returned);
360367
}
368+
369+
@Test
370+
void testCrashHonorsExceptionReporting() throws Exception {
371+
RuntimeException ex = new RuntimeException("something failed");
372+
String returned = exchangeCrash(ex, PipesMessage.MAX_PAYLOAD_BYTES,
373+
new ExceptionReporting(ExceptionReporting.Level.MESSAGE_REDACTED, -1));
374+
assertTrue(returned.contains("java.lang.RuntimeException"));
375+
assertFalse(returned.contains("something failed"));
376+
}
361377
}

0 commit comments

Comments
 (0)