Skip to content

Commit 3eba834

Browse files
tballisontballison
authored
TIKA-4556 - fix pipes client/server protocol bug
1 parent 612fb42 commit 3eba834

3 files changed

Lines changed: 104 additions & 23 deletions

File tree

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

Lines changed: 9 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -140,6 +140,7 @@ private void shutItAllDown() throws InterruptedException {
140140
if (serverTuple == null) {
141141
return;
142142
}
143+
LOG.debug("pipesClientId={}: shutting down server", pipesClientId);
143144
try {
144145
serverTuple.output.write(COMMANDS.SHUT_DOWN.getByte());
145146
serverTuple.output.flush();
@@ -233,8 +234,7 @@ private void maybeInit() throws InterruptedException, ServerInitializationExcept
233234
}
234235

235236
private void writeTask(FetchEmitTuple t) throws IOException {
236-
long start = System.currentTimeMillis();
237-
237+
LOG.debug("pipesClientId={}: sending NEW_REQUEST for id={}", pipesClientId, t.getId());
238238
UnsynchronizedByteArrayOutputStream bos = UnsynchronizedByteArrayOutputStream
239239
.builder()
240240
.get();
@@ -247,9 +247,6 @@ private void writeTask(FetchEmitTuple t) throws IOException {
247247
serverTuple.output.writeInt(bytes.length);
248248
serverTuple.output.write(bytes);
249249
serverTuple.output.flush();
250-
if (LOG.isTraceEnabled()) {
251-
LOG.trace("pipesClientId={}: timer -- write tuple: {} ms", pipesClientId, System.currentTimeMillis() - start);
252-
}
253250
}
254251

255252
private PipesResult waitForServer(FetchEmitTuple t, IntermediateResult intermediateResult) throws InterruptedException {
@@ -425,14 +422,15 @@ private void writeAck() throws IOException {
425422

426423

427424
private void restart() throws InterruptedException, IOException, TimeoutException {
425+
ServerSocket serverSocket = new ServerSocket(0, 50, InetAddress.getLoopbackAddress());
426+
int port = serverSocket.getLocalPort();
428427
if (serverTuple != null && serverTuple.process != null) {
428+
int oldPort = serverTuple.serverSocket.getLocalPort();
429429
shutItAllDown();
430-
LOG.info("pipesClientId={}: restarting process", pipesClientId);
430+
LOG.info("pipesClientId={}: restarting process on port={} (old port was {})", pipesClientId, port, oldPort);
431431
} else {
432-
LOG.info("pipesClientId={}: starting process", pipesClientId);
432+
LOG.info("pipesClientId={}: starting process on port={}", pipesClientId, port);
433433
}
434-
ServerSocket serverSocket = new ServerSocket(0, 50, InetAddress.getLoopbackAddress());
435-
int port = serverSocket.getLocalPort();
436434
Path tmpDir = Files.createTempDirectory("pipes-server-" + pipesClientId + "-");
437435
ProcessBuilder pb = new ProcessBuilder(getCommandline(port, tmpDir));
438436
pb.inheritIO();
@@ -485,7 +483,7 @@ private void waitForStartup() throws IOException {
485483
int b = serverTuple.input.read();
486484
writeAck();
487485
if (b == READY.getByte()) {
488-
LOG.debug("got ready byte");
486+
LOG.debug("pipesClientId={}: server ready", pipesClientId);
489487
} else if (b == FINISHED.getByte()) {
490488
int len = serverTuple.input.readInt();
491489
byte[] bytes = new byte[len];
@@ -537,7 +535,7 @@ private String[] getCommandline(int port, Path tmpDir) {
537535
if (arg.equals("-XX:+ExitOnOutOfMemoryError") || arg.equals("-XX:+CrashOnOutOfMemoryError")) {
538536
hasExitOnOOM = true;
539537
}
540-
if (arg.startsWith("-Dlog4j.configuration")) {
538+
if (arg.startsWith("-Dlog4j.configuration") || arg.startsWith("-Dlog4j2.configuration")) {
541539
hasLog4j = true;
542540
}
543541
if (arg.startsWith("-Xloggc:")) {

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

Lines changed: 45 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -86,6 +86,7 @@ public class PipesServer implements AutoCloseable {
8686
private static final Logger LOG = LoggerFactory.getLogger(PipesServer.class);
8787

8888
private final long heartbeatIntervalMs;
89+
private final String pipesClientId;
8990

9091
//this has to be some number not close to 0-3
9192
//it looks like the server crashes with exit value 3 on uncaught OOM, for example
@@ -151,6 +152,8 @@ public byte getByte() {
151152
private final PipesWorker.EMIT_STRATEGY emitStrategy;
152153

153154
public static PipesServer load(int port, Path tikaConfigPath) throws Exception {
155+
String pipesClientId = System.getProperty("pipesClientId", "unknown");
156+
LOG.debug("pipesClientId={}: connecting to client on port={}", pipesClientId, port);
154157
Socket socket = new Socket();
155158
socket.connect(new InetSocketAddress(InetAddress.getLoopbackAddress(), port), PipesClient.SOCKET_CONNECT_TIMEOUT_MS);
156159

@@ -165,8 +168,9 @@ public static PipesServer load(int port, Path tikaConfigPath) throws Exception {
165168
socket.setSoTimeout((int) pipesConfig.getSocketTimeoutMs());
166169

167170
MetadataFilter metadataFilter = tikaLoader.loadMetadataFilters();
168-
PipesServer pipesServer = new PipesServer(tikaLoader, pipesConfig, socket, dis, dos, metadataFilter);
171+
PipesServer pipesServer = new PipesServer(pipesClientId, tikaLoader, pipesConfig, socket, dis, dos, metadataFilter);
169172
pipesServer.initializeResources();
173+
LOG.debug("pipesClientId={}: PipesServer loaded and ready", pipesClientId);
170174
return pipesServer;
171175
} catch (Exception e) {
172176
LOG.error("Failed to start up", e);
@@ -196,10 +200,11 @@ public static PipesServer load(int port, Path tikaConfigPath) throws Exception {
196200
}
197201
}
198202

199-
public PipesServer(TikaLoader tikaLoader, PipesConfig pipesConfig, Socket socket, DataInputStream in,
203+
public PipesServer(String pipesClientId, TikaLoader tikaLoader, PipesConfig pipesConfig, Socket socket, DataInputStream in,
200204
DataOutputStream out, MetadataFilter metadataFilter) throws TikaConfigException,
201205
IOException {
202206

207+
this.pipesClientId = pipesClientId;
203208
this.tikaLoader = tikaLoader;
204209
this.pipesConfig = pipesConfig;
205210
this.socket = socket;
@@ -234,32 +239,51 @@ public PipesServer(TikaLoader tikaLoader, PipesConfig pipesConfig, Socket socket
234239
public static void main(String[] args) throws Exception {
235240
int port = Integer.parseInt(args[0]);
236241
Path tikaConfig = Paths.get(args[1]);
237-
LOG.debug("starting pipes server on port={} with tikaConfig={}", port, tikaConfig);
242+
String pipesClientId = System.getProperty("pipesClientId", "unknown");
243+
LOG.debug("pipesClientId={}: starting pipes server on port={}", pipesClientId, port);
238244
try (PipesServer server = PipesServer.load(port, tikaConfig)) {
239-
LOG.debug("successfully started pipes server");
240245
server.mainLoop();
241246
} catch (Throwable t) {
242-
LOG.error("crashed", t);
247+
LOG.error("pipesClientId={}: crashed", pipesClientId, t);
243248
throw t;
244249
} finally {
245-
LOG.info("server shutting down");
250+
LOG.info("pipesClientId={}: server shutting down", pipesClientId);
246251
}
247252
}
248253

249254
public void mainLoop() {
250255
write(PROCESSING_STATUS.READY.getByte());
256+
LOG.debug("pipesClientId={}: sent READY, entering main loop", pipesClientId);
251257
ArrayBlockingQueue<Metadata> intermediateResult = new ArrayBlockingQueue<>(1);
252258

253-
LOG.trace("processing requests");
254259
//main loop
255260
try {
256261
long start = System.currentTimeMillis();
257262
while (true) {
258263
int request = input.read();
264+
LOG.trace("pipesClientId={}: received command byte={}", pipesClientId, HexFormat.of().formatHex(new byte[]{(byte)request}));
259265
if (request == -1) {
260266
LOG.warn("received -1 from client; shutting down");
261267
exit(0);
262-
} else if (request == PipesClient.COMMANDS.PING.getByte()) {
268+
}
269+
270+
// Validate that we received a command byte, not a status/ACK byte
271+
if (request == PipesClient.COMMANDS.ACK.getByte()) {
272+
String msg = String.format(Locale.ROOT,
273+
"pipesClientId=%s: PROTOCOL ERROR - Received ACK (byte=0x%02x) when expecting a command. " +
274+
"This indicates a protocol synchronization issue where the server missed consuming an ACK. " +
275+
"Valid commands are: PING(0x%02x), NEW_REQUEST(0x%02x), SHUT_DOWN(0x%02x). " +
276+
"This is likely a bug in the server's message handling - check that all status messages " +
277+
"that trigger client ACKs are properly awaiting those ACKs.",
278+
pipesClientId, (byte)request,
279+
PipesClient.COMMANDS.PING.getByte(),
280+
PipesClient.COMMANDS.NEW_REQUEST.getByte(),
281+
PipesClient.COMMANDS.SHUT_DOWN.getByte());
282+
LOG.error(msg);
283+
throw new IllegalStateException(msg);
284+
}
285+
286+
if (request == PipesClient.COMMANDS.PING.getByte()) {
263287
writeNoAck(PipesClient.COMMANDS.PING.getByte());
264288
} else if (request == PipesClient.COMMANDS.NEW_REQUEST.getByte()) {
265289
intermediateResult.clear();
@@ -284,8 +308,16 @@ public void mainLoop() {
284308
}
285309
System.exit(0);
286310
} else {
287-
LOG.error("Unexpected request byte={}", HexFormat.of().formatHex(new byte[]{(byte)request}));
288-
throw new IllegalStateException("Unexpected request");
311+
String msg = String.format(Locale.ROOT,
312+
"pipesClientId=%s: Unexpected byte 0x%02x in command position. " +
313+
"Expected one of: PING(0x%02x), ACK(0x%02x), NEW_REQUEST(0x%02x), SHUT_DOWN(0x%02x)",
314+
pipesClientId, (byte)request,
315+
PipesClient.COMMANDS.PING.getByte(),
316+
PipesClient.COMMANDS.ACK.getByte(),
317+
PipesClient.COMMANDS.NEW_REQUEST.getByte(),
318+
PipesClient.COMMANDS.SHUT_DOWN.getByte());
319+
LOG.error(msg);
320+
throw new IllegalStateException(msg);
289321
}
290322
output.flush();
291323
}
@@ -346,7 +378,7 @@ private void loopUntilDone(FetchEmitTuple fetchEmitTuple, ExecutorCompletionServ
346378
long elapsed = System.currentTimeMillis() - start.toEpochMilli();
347379
if (elapsed > mockProgressCounter * heartbeatIntervalMs) {
348380
LOG.debug("still processing: {}", mockProgressCounter);
349-
output.write(PROCESSING_STATUS.WORKING.getByte());
381+
write(PROCESSING_STATUS.WORKING.getByte());
350382
output.writeLong(mockProgressCounter++);
351383
output.flush();
352384
}
@@ -484,6 +516,7 @@ private void awaitAck() throws IOException {
484516
if (b == ACK.getByte()) {
485517
return;
486518
}
519+
LOG.error("pipesClientId={}: expected ACK but got byte={}", pipesClientId, HexFormat.of().formatHex(new byte[]{ (byte) b}));
487520
throw new IOException("Wasn't expecting byte=" + HexFormat.of().formatHex(new byte[]{ (byte) b}));
488521
}
489522

@@ -503,7 +536,7 @@ private void write(byte b) {
503536
output.flush();
504537
awaitAck();
505538
} catch (IOException e) {
506-
LOG.error("problem writing data (forking process shutdown?)", e);
539+
LOG.error("pipesClientId={}: problem writing data (forking process shutdown?)", pipesClientId, e);
507540
exit(1);
508541
}
509542
}

‎tika-pipes/tika-pipes-integration-tests/src/test/java/org/apache/tika/pipes/core/PipesClientTest.java‎

Lines changed: 50 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -551,4 +551,54 @@ public void testEmitterNotFound(@TempDir Path tmp) throws Exception {
551551
"Error message should mention the missing emitter");
552552
}
553553
}
554+
555+
@Test
556+
public void testHeartbeatProtocol(@TempDir Path tmp) throws Exception {
557+
// Test that heartbeat protocol works correctly and doesn't cause protocol errors
558+
// This test exercises the WORKING status messages during long-running operations
559+
// to ensure the server properly awaits ACKs after sending heartbeats
560+
561+
Path inputDir = tmp.resolve("input");
562+
Files.createDirectories(inputDir);
563+
564+
// Create a mock file with 2 second delay to trigger multiple heartbeats
565+
String mockContent = "<?xml version=\"1.0\" encoding=\"UTF-8\" ?>" +
566+
"<mock>" +
567+
"<metadata action=\"add\" name=\"dc:creator\">Heartbeat Test</metadata>" +
568+
"<write element=\"p\">Testing heartbeat protocol synchronization</write>" +
569+
"<fakeload millis=\"2000\" cpu=\"1\" mb=\"10\"/>" +
570+
"</mock>";
571+
String testFile = "mock-heartbeat-test.xml";
572+
Files.write(inputDir.resolve(testFile), mockContent.getBytes(StandardCharsets.UTF_8));
573+
574+
// Create config with very short heartbeat interval (100ms) to ensure heartbeats are sent
575+
Path tikaConfigPath = PluginsTestHelper.getFileSystemFetcherConfig(tmp, inputDir, tmp.resolve("output"));
576+
String configContent = Files.readString(tikaConfigPath, StandardCharsets.UTF_8);
577+
578+
// Modify config to add very short heartbeat interval
579+
configContent = configContent.replace(
580+
"\"pipes\": {",
581+
"\"pipes\": {\n \"heartbeatIntervalMs\": 100,"
582+
);
583+
Files.writeString(tikaConfigPath, configContent, StandardCharsets.UTF_8);
584+
585+
TikaJsonConfig tikaJsonConfig = TikaJsonConfig.load(tikaConfigPath);
586+
PipesConfig pipesConfig = PipesConfig.load(tikaJsonConfig, tikaConfigPath);
587+
588+
try (PipesClient pipesClient = new PipesClient(pipesConfig)) {
589+
// Process file - should complete successfully despite multiple heartbeats
590+
PipesResult pipesResult = pipesClient.process(
591+
new FetchEmitTuple(testFile, new FetchKey(fetcherName, testFile),
592+
new EmitKey(), new Metadata(), new ParseContext(),
593+
FetchEmitTuple.ON_PARSE_EXCEPTION.SKIP));
594+
595+
// Verify successful completion
596+
assertTrue(pipesResult.isSuccess(),
597+
"Processing should succeed even with heartbeat messages. Got status: " + pipesResult.status());
598+
Assertions.assertNotNull(pipesResult.emitData().getMetadataList());
599+
assertEquals(1, pipesResult.emitData().getMetadataList().size());
600+
Metadata metadata = pipesResult.emitData().getMetadataList().get(0);
601+
assertEquals("Heartbeat Test", metadata.get("dc:creator"));
602+
}
603+
}
554604
}

0 commit comments

Comments
 (0)