luciano/amazon-kinesis-client
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

refactor(Worker): allow worker to fail after error in processing records

Browse source
This commit is contained in:
glarwood 2019-03-22 15:17:31 +00:00
parent 913f2e9377
commit f0230c9040
1 changed files with 5 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

5
src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/Worker.java
View file

@ -86,6 +86,9 @@ public class Worker implements Runnable {
private static final WorkerStateChangeListener DEFAULT_WORKER_STATE_CHANGE_LISTENER = new NoOpWorkerStateChangeListener(); private static final WorkerStateChangeListener DEFAULT_WORKER_STATE_CHANGE_LISTENER = new NoOpWorkerStateChangeListener();
private static final boolean DEFAULT_EXITS_ON_FAILURE = false; private static final boolean DEFAULT_EXITS_ON_FAILURE = false;
// WARNING: dynamo connector depends upon this exact string
private static final String PROCESS_RECORDS_ERROR_MESSAGE = "Failure to process records";
private WorkerLog wlog = new WorkerLog(); private WorkerLog wlog = new WorkerLog();
private final String applicationName; private final String applicationName;
@ -528,6 +531,8 @@ public class Worker implements Runnable {
wlog.info("Sleeping ..."); wlog.info("Sleeping ...");
Thread.sleep(idleTimeInMilliseconds); Thread.sleep(idleTimeInMilliseconds);
} catch (Exception e) { } catch (Exception e) {
if (e.getCause().getMessage().contains(PROCESS_RECORDS_ERROR_MESSAGE))
throw new RuntimeException("Failing worker after irrecoverable failure in processing records");
if (exitOnFailure && retries.getAndIncrement() > MAX_RETRIES) if (exitOnFailure && retries.getAndIncrement() > MAX_RETRIES)
throw new RuntimeException("Failing after " + MAX_RETRIES + " attempts", e); throw new RuntimeException("Failing after " + MAX_RETRIES + " attempts", e);