From b98b0369f423ec196105ee2420869d0a327ed26a Mon Sep 17 00:00:00 2001 From: Mark Gordon Date: Thu, 2 Jan 2025 19:01:37 -0800 Subject: [PATCH] Remove Kinesis producer's internal TTL by default --- kinesis-producer-library.properties.example | 2 +- .../zendesk/maxwell/producer/MaxwellKinesisProducer.java | 7 ++++++- 2 files changed, 7 insertions(+), 2 deletions(-) diff --git a/kinesis-producer-library.properties.example b/kinesis-producer-library.properties.example index 8bc3f36c2..a3c47968d 100644 --- a/kinesis-producer-library.properties.example +++ b/kinesis-producer-library.properties.example @@ -258,7 +258,7 @@ RecordMaxBufferedTime = 100 # Default: 30000 # Minimum: 100 # Maximum (inclusive): 9223372036854775807 -RecordTtl = 30000 +RecordTtl = 3600000 # Which region to send records to. # diff --git a/src/main/java/com/zendesk/maxwell/producer/MaxwellKinesisProducer.java b/src/main/java/com/zendesk/maxwell/producer/MaxwellKinesisProducer.java index 1ddb1409d..07489cf17 100644 --- a/src/main/java/com/zendesk/maxwell/producer/MaxwellKinesisProducer.java +++ b/src/main/java/com/zendesk/maxwell/producer/MaxwellKinesisProducer.java @@ -112,7 +112,12 @@ public MaxwellKinesisProducer(MaxwellContext context, String kinesisStream) { KinesisProducerConfiguration config = KinesisProducerConfiguration.fromPropertiesFile(path.toString()); this.kinesisProducer = new KinesisProducer(config); } else { - this.kinesisProducer = new KinesisProducer(); + // The default 30 second record Ttl is too aggressive and prevents our own back-pressure + // logic from backing as needed off before the producer fails. Setting it to 1 hour + // instead. + KinesisProducerConfiguration config = new KinesisProducerConfiguration(); + config.setRecordTtl(3600000); + this.kinesisProducer = new KinesisProducer(config); } }