View | Details | Raw Unified | Return to bug 35819
Collapse All | Expand All

(-)a/misc/workers/es_indexer_daemon.pl (-9 / +37 lines)
Lines 56-61 use Try::Tiny; Link Here
56
use Pod::Usage;
56
use Pod::Usage;
57
use Getopt::Long;
57
use Getopt::Long;
58
use List::MoreUtils qw( natatime );
58
use List::MoreUtils qw( natatime );
59
use Time::HiRes;
59
60
60
use C4::Context;
61
use C4::Context;
61
use Koha::Logger;
62
use Koha::Logger;
Lines 65-70 use Koha::SearchEngine::Indexer; Link Here
65
66
66
67
67
my ( $help, $batch_size );
68
my ( $help, $batch_size );
69
70
my $not_found_retries = {};
71
my $max_retries = $ENV{MAX_RETRIES} || 10;
72
68
GetOptions(
73
GetOptions(
69
    'h|help' => \$help,
74
    'h|help' => \$help,
70
    'b|batch_size=s' => \$batch_size
75
    'b|batch_size=s' => \$batch_size
Lines 116-137 while (1) { Link Here
116
            my $body = $frame->body;
121
            my $body = $frame->body;
117
            decode_json($body); # TODO Should this be from_json? Check utf8 flag.
122
            decode_json($body); # TODO Should this be from_json? Check utf8 flag.
118
        } catch {
123
        } catch {
119
            $logger->warn(sprintf "Frame not processed - %s", $_);
124
            Koha::Logger->get({ interface => 'worker' })->warn(sprintf "Frame not processed - %s", $_);
120
            return;
125
            return;
121
        } finally {
122
            $conn->ack( { frame => $frame } );
123
        };
126
        };
124
127
125
        next unless $args;
128
        unless ( $args ) {
129
            Koha::Logger->get({ interface => 'worker' })->warn(sprintf "Frame does not have correct args, ignoring it");
130
            $conn->nack( { frame => $frame, requeue => 'false' } );
131
            next;
132
        }
133
134
        my $job = Koha::BackgroundJobs->find( $args->{job_id} );
126
135
127
        # FIXME This means we need to have create the DB entry before
136
        if ( $job && $job->status ne 'new' ) {
128
        # It could work in a first step, but then we will want to handle job that will be created from the message received
137
            Koha::Logger->get( { interface => 'worker' } )
129
        my $job = Koha::BackgroundJobs->search( { id => $args->{job_id}, status => 'new' } )->next;
138
                ->warn( sprintf "Job %s has wrong status %s", $args->{job_id}, $job->status );
139
140
            # nack without requeue, we do not want to process this frame again
141
            $conn->nack( { frame => $frame, requeue => 'false' } );
142
            next;
143
        }
130
144
131
        unless ($job) {
145
        unless ($job) {
132
            $logger->warn( sprintf "Job %s not found, or has wrong status", $args->{job_id} );
146
            if ( ++$not_found_retries->{$args->{job_id}} >= $max_retries ) {
147
                Koha::Logger->get( { interface => 'worker' } )
148
                    ->warn( sprintf "Job %s not found, no more retry", $args->{job_id} );
149
150
                # nack without requeue, we do not want to process this frame again
151
                $conn->nack( { frame => $frame, requeue => 'false' } );
152
                next;
153
            }
154
155
            Koha::Logger->get( { interface => 'worker' } )
156
                ->debug( sprintf "Job %s not found, will retry later", $args->{job_id} );
157
158
            # nack to force requeue
159
            $conn->nack( { frame => $frame, requeue => 'true' } );
160
            Time::HiRes::sleep(0.5);
133
            next;
161
            next;
134
        }
162
        }
163
        $conn->ack( { frame => $frame } );
135
164
136
        push @jobs, $job;
165
        push @jobs, $job;
137
        if ( @jobs >= $batch_size || !$conn->can_read( { timeout => '0.1' } ) ) {
166
        if ( @jobs >= $batch_size || !$conn->can_read( { timeout => '0.1' } ) ) {
138
- 

Return to bug 35819