From 5ac2fd41784c3734b893118e5c108723705937f8 Mon Sep 17 00:00:00 2001 From: David Cook Date: Fri, 14 Feb 2025 02:39:28 +0000 Subject: [PATCH] Bug 34070: Add error handling for ERROR frames This change causes es_indexer_daemon.pl to die on ERROR frames instead of spinning in an infinite loop. It also adds a content-type STOMP frame header to help consumers know what data type is in the MESSAGE frame body. To test ERROR handling for es_indexer_daemon.pl: 1. Switch to using Elasticsearch 2. Change your RabbitMQ connection details in koha-conf.xml or the hard-coded ones in Koha/BackgroundJob.pm to include an invalid password. For example, if you don't have a message_broker already defined, use the following: bad 3. sudo koha-es-indexer --restart kohadev 4. View the errors in the logs: tail -f /var/log/koha/kohadev/es-indexer-output.log tail -f /var/log/koha/kohadev/es-indexer-error.log To test MESSAGE handling for es_indexer_daemon.pl: 1. Switch to using Elasticsearch (and ensure RabbitMQ connection details are correct) 2. Change the title of a bib record to something new and unique 3. Try to search for the bib using the new title 4. Note you can find it 5. Look in the logs just to make sure there are no messages there --- Koha/BackgroundJob.pm | 8 ++++++-- misc/workers/es_indexer_daemon.pl | 23 +++++++++++++++++++++-- 2 files changed, 27 insertions(+), 4 deletions(-) diff --git a/Koha/BackgroundJob.pm b/Koha/BackgroundJob.pm index 211e2e32f3..0ab3b4c4ee 100644 --- a/Koha/BackgroundJob.pm +++ b/Koha/BackgroundJob.pm @@ -151,8 +151,12 @@ sub enqueue { my $namespace = C4::Context->config('memcached_namespace'); my $encoded_args = Encode::encode_utf8($json_args); # FIXME We should better leave this to Net::Stomp? my $destination = sprintf( "/queue/%s-%s", $namespace, $job_queue ); - $conn->send_with_receipt( { destination => $destination, body => $encoded_args, persistent => 'true' } ) - or Koha::Exceptions::BackgroundJob->throw('Job has not been enqueued'); + $conn->send_with_receipt( + { + destination => $destination, body => $encoded_args, persistent => 'true', + 'content-type' => 'application/json' + } + ) or Koha::Exceptions::BackgroundJob->throw('Job has not been enqueued'); } catch { $self->status('failed')->store; if ( ref($_) eq 'Koha::Exceptions::BackgroundJob' ) { diff --git a/misc/workers/es_indexer_daemon.pl b/misc/workers/es_indexer_daemon.pl index aa2f2a9db2..4f8a2f57a0 100755 --- a/misc/workers/es_indexer_daemon.pl +++ b/misc/workers/es_indexer_daemon.pl @@ -119,8 +119,27 @@ while (1) { } my $args = try { - my $body = $frame->body; - decode_json($body); # TODO Should this be from_json? Check utf8 flag. + my $command = $frame->command; + my $body = $frame->body; + my $content_type = $frame->content_type; + if ( $command && $command eq 'MESSAGE' ) { + if ( $content_type && $content_type eq 'application/json' ) { + decode_json($body); # TODO Should this be from_json? Check utf8 flag. + } else { + + #TODO: This is a fallback for older enqueued messages which are missing the content-type header + #TODO: Please replace this decode with a die in the future once content-type is well established + decode_json($body); # TODO Should this be from_json? Check utf8 flag. + } + } elsif ( $command && $command eq 'ERROR' ) { + + #Known errors: + #You must log in using CONNECT first + #"NACK" must include a valid "message-id" header + Koha::Logger->get( { interface => 'worker' } ) + ->warn( sprintf "Shutting down after receiving ERROR frame:\n%s\n", $frame->as_string ); + exit 1; + } } catch { Koha::Logger->get( { interface => 'worker' } )->warn( sprintf "Frame not processed - %s", $_ ); return; -- 2.39.5