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

fix(worker): fail on TrimmedDataAccessException (101797)

Browse source
This commit is contained in:
bencvdb 2021-07-01 09:57:34 -07:00
parent e8dcb0a810
commit d4eb2b66e3
1 changed files with 15 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

15
src/main/java/com/fivetran/external/com/amazonaws/services/kinesis/clientlibrary/lib/worker/Worker.java vendored
View file

@ -31,6 +31,7 @@ import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Consumer;
import com.amazonaws.services.dynamodbv2.model.AmazonDynamoDBException;
import com.amazonaws.services.dynamodbv2.model.TrimmedDataAccessException;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
@ -526,6 +527,8 @@ public class Worker implements Runnable {
} catch (AmazonDynamoDBException e) {
throw e;
} catch (Exception e) {
handleTrimmedDataAccessException(e);
if (causedByStreamRecordProcessingError(e))
throw new RuntimeException("Failing worker after irrecoverable failure: " + e.getMessage());
if (exitOnFailure && retries.getAndIncrement() > MAX_RETRIES)
@ -552,6 +555,18 @@ public class Worker implements Runnable {
return causedByStreamRecordProcessingError(t.getCause());
}
private void handleTrimmedDataAccessException(Exception e) {
Optional<TrimmedDataAccessException> maybeTrimmedException = getTrimmedDataAccessException(e);
if (maybeTrimmedException.isPresent()) throw maybeTrimmedException.get();
}
private Optional<TrimmedDataAccessException> getTrimmedDataAccessException(Throwable t) {
if (t.getCause() == null) return Optional.empty();
if (t.getCause().getClass().equals(TrimmedDataAccessException.class)) return Optional.of( (TrimmedDataAccessException) t.getCause());
return getTrimmedDataAccessException(t.getCause());
}
private void initialize() {
workerStateChangeListener.onWorkerStateChange(WorkerStateChangeListener.WorkerState.INITIALIZING);
boolean isDone = false;