Skip to content

Commit 433f576

Browse files
tballisontballison
authored
TIKA-4815: trivial follow-ups from the PR #3022 review (#3026)
1 parent 8fced66 commit 433f576

4 files changed

Lines changed: 76 additions & 64 deletions

File tree

‎docs/modules/ROOT/pages/using-tika/grpc/index.adoc‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -29,6 +29,14 @@ register a fetcher (`SaveFetcher`) and then submit `FetchAndParseRequest`
2929
messages, each of which returns a `FetchAndParseReply` with extracted
3030
metadata and content.
3131

32+
== Concurrency
33+
34+
`FetchAndParse` and its streaming variants run on a pool of forked worker JVMs
35+
sized by `pipes.numClients` (default: derived from host cores, at most 4). A
36+
call that cannot get a worker within `pipes.maxWaitForClientMillis` (default
37+
60s) returns the in-band reply status `CLIENT_UNAVAILABLE_WITHIN_MS` -- at
38+
capacity, not failing. `pipes.useSharedServer: true` shares one JVM instead.
39+
3240
== Security
3341

3442
[WARNING]

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

Lines changed: 7 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -80,7 +80,7 @@ public class TikaGrpcConcurrencyTest {
8080
*/
8181
@Test
8282
public void concurrentCallsAllParseTheirOwnDocument(Resources resources) throws Exception {
83-
runConcurrentBurst(resources, writeConfig(null, null, null));
83+
runConcurrentBurst(resources, writeConfig(null, null, null), false);
8484
}
8585

8686
/**
@@ -91,12 +91,16 @@ public void concurrentCallsAllParseTheirOwnDocument(Resources resources) throws
9191
*/
9292
@Test
9393
public void sharedServerModeParsesConcurrently(Resources resources) throws Exception {
94-
runConcurrentBurst(resources, writeConfig(null, null, Boolean.TRUE));
94+
runConcurrentBurst(resources, writeConfig(null, null, Boolean.TRUE), true);
9595
}
9696

97-
private void runConcurrentBurst(Resources resources, Path config) throws Exception {
97+
private void runConcurrentBurst(Resources resources, Path config, boolean expectSharedMode)
98+
throws Exception {
9899
int concurrency = 4;
99100
TikaGrpcServerImpl service = new TikaGrpcServerImpl(config.toAbsolutePath().toString());
101+
// the burst alone can't tell the modes apart
102+
assertEquals(expectSharedMode, service.pipesParser.isSharedMode(),
103+
"pipes.useSharedServer did not take effect");
100104
List<File> testFiles = new ArrayList<>();
101105
ExecutorService pool = Executors.newFixedThreadPool(concurrency);
102106
try {

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

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -290,7 +290,7 @@ public void markServerForRestart() {
290290

291291
@Override
292292
public void connectionAbandoned() {
293-
LOG.info("clientId={}: connection abandoned mid-request, recycling the worker", clientId);
293+
LOG.info("clientId={}: connection abandoned, worker will be recycled on next use", clientId);
294294
pendingRestart = true;
295295
}
296296

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

Lines changed: 60 additions & 60 deletions
Original file line numberDiff line numberDiff line change
@@ -55,7 +55,7 @@ public class PipesClientInterruptTest {
5555
* open with an abandoned request on it.
5656
*/
5757
@Test
58-
@Timeout(30)
58+
@Timeout(45)
5959
public void interruptClosesTheConnection() throws Exception {
6060
try (ServerSocket serverSocket = new ServerSocket(0)) {
6161
CountDownLatch heartbeatStarted = new CountDownLatch(1);
@@ -67,37 +67,36 @@ public void interruptClosesTheConnection() throws Exception {
6767

6868
PipesConfig pipesConfig = new PipesConfig();
6969
SentinelServerManager manager = new SentinelServerManager(serverSocket.getLocalPort());
70-
PipesClient client = new PipesClient(pipesConfig, manager);
71-
72-
AtomicReference<Throwable> fromProcess = new AtomicReference<>();
73-
CountDownLatch processReturned = new CountDownLatch(1);
74-
Thread worker = new Thread(() -> {
75-
try {
76-
client.process(new FetchEmitTuple("interrupt-test",
77-
new FetchKey("fetcher", "key"), new EmitKey(), new Metadata(),
78-
new ParseContext(), FetchEmitTuple.ON_PARSE_EXCEPTION.SKIP));
79-
} catch (Throwable t) {
80-
fromProcess.set(t);
81-
} finally {
82-
processReturned.countDown();
83-
}
84-
});
85-
worker.start();
70+
try (PipesClient client = new PipesClient(pipesConfig, manager)) {
71+
AtomicReference<Throwable> fromProcess = new AtomicReference<>();
72+
CountDownLatch processReturned = new CountDownLatch(1);
73+
Thread worker = new Thread(() -> {
74+
try {
75+
client.process(new FetchEmitTuple("interrupt-test",
76+
new FetchKey("fetcher", "key"), new EmitKey(), new Metadata(),
77+
new ParseContext(), FetchEmitTuple.ON_PARSE_EXCEPTION.SKIP));
78+
} catch (Throwable t) {
79+
fromProcess.set(t);
80+
} finally {
81+
processReturned.countDown();
82+
}
83+
});
84+
worker.start();
8685

87-
assertTrue(heartbeatStarted.await(15, TimeUnit.SECONDS),
88-
"the scripted server never got the request; the test proves nothing");
89-
worker.interrupt();
86+
assertTrue(heartbeatStarted.await(15, TimeUnit.SECONDS),
87+
"the scripted server never got the request; the test proves nothing");
88+
worker.interrupt();
9089

91-
assertTrue(processReturned.await(15, TimeUnit.SECONDS),
92-
"process() must return after the interrupt");
93-
assertTrue(fromProcess.get() instanceof InterruptedException,
94-
"process() must rethrow the interrupt, got: " + fromProcess.get());
95-
assertTrue(connectionClosed.await(5, TimeUnit.SECONDS),
96-
"the interrupted client left its connection open with a request in flight");
97-
assertTrue(manager.abandoned,
98-
"the manager was not told; a per-client worker never dials back, so the "
99-
+ "next connect() would wait out the accept timeout for nothing");
100-
client.close();
90+
assertTrue(processReturned.await(15, TimeUnit.SECONDS),
91+
"process() must return after the interrupt");
92+
assertTrue(fromProcess.get() instanceof InterruptedException,
93+
"process() must rethrow the interrupt, got: " + fromProcess.get());
94+
assertTrue(connectionClosed.await(5, TimeUnit.SECONDS),
95+
"the interrupted client left its connection open with a request in flight");
96+
assertTrue(manager.abandoned,
97+
"the manager was not told; a per-client worker never dials back, so the "
98+
+ "next connect() would wait out the accept timeout for nothing");
99+
}
101100
}
102101
}
103102

@@ -109,7 +108,7 @@ public void interruptClosesTheConnection() throws Exception {
109108
* is delivered.
110109
*/
111110
@Test
112-
@Timeout(30)
111+
@Timeout(45)
113112
public void interruptDuringStartupBackoffAbandonsTheConnection() throws Exception {
114113
try (ServerSocket serverSocket = new ServerSocket(0)) {
115114
CountDownLatch badHandshakeSent = new CountDownLatch(1);
@@ -120,38 +119,39 @@ public void interruptDuringStartupBackoffAbandonsTheConnection() throws Exceptio
120119
sentinel.start();
121120

122121
PipesConfig pipesConfig = new PipesConfig();
122+
// handshake reads aren't interrupt-responsive; a late interrupt waits this out
123+
pipesConfig.setStartupTimeoutMillis(1000);
123124
SentinelServerManager manager = new SentinelServerManager(serverSocket.getLocalPort());
124-
PipesClient client = new PipesClient(pipesConfig, manager);
125-
126-
AtomicReference<Throwable> fromProcess = new AtomicReference<>();
127-
CountDownLatch processReturned = new CountDownLatch(1);
128-
Thread worker = new Thread(() -> {
129-
try {
130-
client.process(new FetchEmitTuple("interrupt-startup-test",
131-
new FetchKey("fetcher", "key"), new EmitKey(), new Metadata(),
132-
new ParseContext(), FetchEmitTuple.ON_PARSE_EXCEPTION.SKIP));
133-
} catch (Throwable t) {
134-
fromProcess.set(t);
135-
} finally {
136-
processReturned.countDown();
137-
}
138-
});
139-
worker.start();
125+
try (PipesClient client = new PipesClient(pipesConfig, manager)) {
126+
AtomicReference<Throwable> fromProcess = new AtomicReference<>();
127+
CountDownLatch processReturned = new CountDownLatch(1);
128+
Thread worker = new Thread(() -> {
129+
try {
130+
client.process(new FetchEmitTuple("interrupt-startup-test",
131+
new FetchKey("fetcher", "key"), new EmitKey(), new Metadata(),
132+
new ParseContext(), FetchEmitTuple.ON_PARSE_EXCEPTION.SKIP));
133+
} catch (Throwable t) {
134+
fromProcess.set(t);
135+
} finally {
136+
processReturned.countDown();
137+
}
138+
});
139+
worker.start();
140140

141-
assertTrue(badHandshakeSent.await(15, TimeUnit.SECONDS),
142-
"the scripted server never got a connection; the test proves nothing");
143-
worker.interrupt();
141+
assertTrue(badHandshakeSent.await(15, TimeUnit.SECONDS),
142+
"the scripted server never got a connection; the test proves nothing");
143+
worker.interrupt();
144144

145-
assertTrue(processReturned.await(15, TimeUnit.SECONDS),
146-
"process() must return after the interrupt");
147-
assertTrue(fromProcess.get() instanceof InterruptedException,
148-
"process() must rethrow the interrupt, got: " + fromProcess.get());
149-
assertTrue(connectionClosed.await(5, TimeUnit.SECONDS),
150-
"the interrupted client left its half-established connection open");
151-
assertTrue(manager.abandoned,
152-
"the manager was not told; an abandoned per-client worker never "
153-
+ "dials back, mid-handshake or not");
154-
client.close();
145+
assertTrue(processReturned.await(15, TimeUnit.SECONDS),
146+
"process() must return after the interrupt");
147+
assertTrue(fromProcess.get() instanceof InterruptedException,
148+
"process() must rethrow the interrupt, got: " + fromProcess.get());
149+
assertTrue(connectionClosed.await(5, TimeUnit.SECONDS),
150+
"the interrupted client left its half-established connection open");
151+
assertTrue(manager.abandoned,
152+
"the manager was not told; an abandoned per-client worker never "
153+
+ "dials back, mid-handshake or not");
154+
}
155155
}
156156
}
157157

0 commit comments

Comments
 (0)