From 826070228861191c854ccd8e5d0fd2b007cc19af Mon Sep 17 00:00:00 2001 From: Danylo Bosenko Date: Tue, 18 Aug 2026 17:03:55 +0300 Subject: [PATCH 1/2] feat: add optional -ttl argument to set TTL for migrated Cassandra data Migrated data currently lives forever in Cassandra since no TTL is set on writes. Add an optional -ttl CLI argument (in days) that applies USING TTL to ts_kv_cf and ts_kv_partitions_cf inserts. ts_kv_latest_cf is intentionally left unaffected, matching ThingsBoard's own TTL semantics where only historical points expire, not the latest value. Co-Authored-By: Claude Sonnet 5 --- README.md | 18 ++++++++++++++++++ .../client/tools/migrator/MigratorTool.java | 16 +++++++++++++++- .../client/tools/migrator/PgCaMigrator.java | 5 +++-- .../client/tools/migrator/WriterBuilder.java | 14 ++++++++++---- .../migrator/writer/AbstractTbWriter.java | 6 +++++- .../tools/migrator/writer/TbLatestWriter.java | 3 ++- .../tools/migrator/writer/TbTsWriter.java | 10 +++++----- 7 files changed, 58 insertions(+), 14 deletions(-) diff --git a/README.md b/README.md index f75e7a7..4c147dd 100644 --- a/README.md +++ b/README.md @@ -336,6 +336,24 @@ sstableloader --verbose --nodes CASSANDRA_NODES --username cassandra --password Verify that historical data available in ThingsBoard. +# Setting TTL for migrated data + +By default migrated data never expires. To make migrated timeseries data expire automatically in Cassandra, +pass the optional `-ttl` argument (in days) when running the tool: + +``` +java -jar ./target/database-migrator-1.0-SNAPSHOT-jar-with-dependencies.jar \ + ... \ + -ttl 90 +``` + +Notes: +* TTL applies only to `ts_kv_cf` and `ts_kv_partitions_cf` (historical data points and their partition bookkeeping). +* `ts_kv_latest_cf` (last known value per key) is never affected by `-ttl` — this matches ThingsBoard's own TTL + behavior, where only historical points expire, not the latest value. +* TTL is counted from the moment the SSTables are generated by this tool, not from the original timestamp of + the data being migrated. + # Troubleshooting ## Continue migration in case of failure on particular migration line diff --git a/src/main/java/org/thingsboard/client/tools/migrator/MigratorTool.java b/src/main/java/org/thingsboard/client/tools/migrator/MigratorTool.java index 62db63a..720ed9e 100644 --- a/src/main/java/org/thingsboard/client/tools/migrator/MigratorTool.java +++ b/src/main/java/org/thingsboard/client/tools/migrator/MigratorTool.java @@ -25,6 +25,7 @@ import org.apache.commons.cli.ParseException; import java.io.File; +import java.util.concurrent.TimeUnit; @Slf4j public class MigratorTool { @@ -60,6 +61,12 @@ public static void main(String[] args) { linesToSkip = Integer.parseInt(cmd.getOptionValue("linesToSkip")); } + Long ttlSeconds = null; + if (cmd.getOptionValue("ttl") != null) { + long ttlDays = Long.parseLong(cmd.getOptionValue("ttl")); + ttlSeconds = TimeUnit.DAYS.toSeconds(ttlDays); + } + new PgCaMigrator( allTelemetrySource, tsSaveDir, @@ -68,7 +75,8 @@ public static void main(String[] args) { allEntityIdsAndTypes, dictionaryParser, castEnable, - partitioning).migrate(linesToSkip); + partitioning, + ttlSeconds).migrate(linesToSkip); } catch (Throwable th) { log.error("Failed to migrate", th); @@ -117,6 +125,12 @@ private static CommandLine parseArgs(String[] args) { linesToSkipOpt.setRequired(false); options.addOption(linesToSkipOpt); + Option ttlOpt = new Option("ttl", "ttl", true, + "TTL in days for migrated timeseries data (ts_kv_cf and ts_kv_partitions_cf). " + + "If not set, migrated data will never expire. Does not affect ts_kv_latest_cf."); + ttlOpt.setRequired(false); + options.addOption(ttlOpt); + HelpFormatter formatter = new HelpFormatter(); CommandLineParser parser = new BasicParser(); diff --git a/src/main/java/org/thingsboard/client/tools/migrator/PgCaMigrator.java b/src/main/java/org/thingsboard/client/tools/migrator/PgCaMigrator.java index c7409d2..9036d22 100644 --- a/src/main/java/org/thingsboard/client/tools/migrator/PgCaMigrator.java +++ b/src/main/java/org/thingsboard/client/tools/migrator/PgCaMigrator.java @@ -39,13 +39,14 @@ public PgCaMigrator(File sourceFile, RelatedEntitiesParser allEntityIdsAndTypes, DictionaryParser dictionaryParser, boolean castStringsIfPossible, - String partitioning) { + String partitioning, + Long ttlSeconds) { this.sourceFile = sourceFile; if (outTsLatestDir != null) { this.tbLatestWriter = new TbLatestWriter(dictionaryParser, allEntityIdsAndTypes, outTsLatestDir, castStringsIfPossible, partitioning); } if (ourTsDir != null) { - this.tbTsWriter = new TbTsWriter(dictionaryParser, allEntityIdsAndTypes, ourTsDir, outTsPartitionDir, castStringsIfPossible, partitioning); + this.tbTsWriter = new TbTsWriter(dictionaryParser, allEntityIdsAndTypes, ourTsDir, outTsPartitionDir, castStringsIfPossible, partitioning, ttlSeconds); } } diff --git a/src/main/java/org/thingsboard/client/tools/migrator/WriterBuilder.java b/src/main/java/org/thingsboard/client/tools/migrator/WriterBuilder.java index e2d1571..56a4526 100644 --- a/src/main/java/org/thingsboard/client/tools/migrator/WriterBuilder.java +++ b/src/main/java/org/thingsboard/client/tools/migrator/WriterBuilder.java @@ -57,15 +57,17 @@ public class WriterBuilder { ") WITH CLUSTERING ORDER BY ( partition ASC )\n" + " AND compaction = { 'class' : 'LeveledCompactionStrategy' };"; - public static CQLSSTableWriter getTsWriter(File dir) { + public static CQLSSTableWriter getTsWriter(File dir, Long ttlSeconds) { return CQLSSTableWriter.builder() .inDirectory(dir.getAbsolutePath()) .forTable(tsSchema) .using("INSERT INTO thingsboard.ts_kv_cf (entity_type, entity_id, key, partition, ts, bool_v, str_v, long_v, dbl_v, json_v) " + - "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)") + "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)" + ttlClause(ttlSeconds)) .build(); } + // Latest values intentionally never expire, regardless of ttlSeconds - they represent + // the current state of a key, not historical points, matching ThingsBoard's own TTL semantics. public static CQLSSTableWriter getLatestWriter(File dir) { return CQLSSTableWriter.builder() .inDirectory(dir.getAbsolutePath()) @@ -75,12 +77,16 @@ public static CQLSSTableWriter getLatestWriter(File dir) { .build(); } - public static CQLSSTableWriter getPartitionWriter(File dir) { + public static CQLSSTableWriter getPartitionWriter(File dir, Long ttlSeconds) { return CQLSSTableWriter.builder() .inDirectory(dir.getAbsolutePath()) .forTable(partitionSchema) .using("INSERT INTO thingsboard.ts_kv_partitions_cf (entity_type, entity_id, key, partition) " + - "VALUES (?, ?, ?, ?)") + "VALUES (?, ?, ?, ?)" + ttlClause(ttlSeconds)) .build(); } + + private static String ttlClause(Long ttlSeconds) { + return ttlSeconds != null ? " USING TTL " + ttlSeconds : ""; + } } diff --git a/src/main/java/org/thingsboard/client/tools/migrator/writer/AbstractTbWriter.java b/src/main/java/org/thingsboard/client/tools/migrator/writer/AbstractTbWriter.java index 2454601..271fa17 100644 --- a/src/main/java/org/thingsboard/client/tools/migrator/writer/AbstractTbWriter.java +++ b/src/main/java/org/thingsboard/client/tools/migrator/writer/AbstractTbWriter.java @@ -67,10 +67,14 @@ public abstract class AbstractTbWriter implements TbWriter { private final boolean castStringIfPossible; + // null means no TTL - migrated rows never expire, matching pre-existing behavior + protected final Long ttlSeconds; + public AbstractTbWriter(DictionaryParser keyParser, RelatedEntitiesParser entityIdsAndTypes, File outDir, - boolean castStringIfPossible, String partitioning) { + boolean castStringIfPossible, String partitioning, Long ttlSeconds) { this.keyParser = keyParser; this.entityIdsAndTypes = entityIdsAndTypes; + this.ttlSeconds = ttlSeconds; this.currentWriter = getWriter(outDir); this.outDir = outDir; this.castStringIfPossible = castStringIfPossible; diff --git a/src/main/java/org/thingsboard/client/tools/migrator/writer/TbLatestWriter.java b/src/main/java/org/thingsboard/client/tools/migrator/writer/TbLatestWriter.java index fcc4c3f..a726db9 100644 --- a/src/main/java/org/thingsboard/client/tools/migrator/writer/TbLatestWriter.java +++ b/src/main/java/org/thingsboard/client/tools/migrator/writer/TbLatestWriter.java @@ -29,7 +29,8 @@ public class TbLatestWriter extends AbstractTbWriter { public TbLatestWriter(DictionaryParser keyParser, RelatedEntitiesParser entityIdsAndTypes, File outDir, boolean castStringsIfPossible, String partitioning) { - super(keyParser, entityIdsAndTypes, outDir, castStringsIfPossible, partitioning); + // latest values never expire regardless of the migration's ttl setting + super(keyParser, entityIdsAndTypes, outDir, castStringsIfPossible, partitioning, null); } @Override diff --git a/src/main/java/org/thingsboard/client/tools/migrator/writer/TbTsWriter.java b/src/main/java/org/thingsboard/client/tools/migrator/writer/TbTsWriter.java index 39395a1..49c1486 100644 --- a/src/main/java/org/thingsboard/client/tools/migrator/writer/TbTsWriter.java +++ b/src/main/java/org/thingsboard/client/tools/migrator/writer/TbTsWriter.java @@ -34,8 +34,8 @@ public class TbTsWriter extends AbstractTbWriter { private final File outTsPartitionDir; public TbTsWriter(DictionaryParser keyParser, RelatedEntitiesParser entityIdsAndTypes, File outDir, - File outTsPartitionDir, boolean castStringsIfPossible, String partitioning) { - super(keyParser, entityIdsAndTypes, outDir, castStringsIfPossible, partitioning); + File outTsPartitionDir, boolean castStringsIfPossible, String partitioning, Long ttlSeconds) { + super(keyParser, entityIdsAndTypes, outDir, castStringsIfPossible, partitioning, ttlSeconds); this.outTsPartitionDir = outTsPartitionDir; } @@ -61,19 +61,19 @@ public List toValues(List raw) { @Override public void reOpenWriter() throws IOException { currentWriter.close(); - currentWriter = WriterBuilder.getTsWriter(outDir); + currentWriter = WriterBuilder.getTsWriter(outDir, ttlSeconds); } @Override public CQLSSTableWriter getWriter(File outDir) { - return WriterBuilder.getTsWriter(outDir); + return WriterBuilder.getTsWriter(outDir, ttlSeconds); } @Override public void writePartitions() throws IOException { CQLSSTableWriter currentPartitionsWriter = null; try { - currentPartitionsWriter = WriterBuilder.getPartitionWriter(outTsPartitionDir); + currentPartitionsWriter = WriterBuilder.getPartitionWriter(outTsPartitionDir, ttlSeconds); log.info("Partitions collected " + partitions.size()); long startTs = System.currentTimeMillis(); for (String partition : partitions) { From 8a1788acdf7dffb5dd2a7098a1f8a28c9dedd5ed Mon Sep 17 00:00:00 2001 From: Danylo Bosenko Date: Tue, 18 Aug 2026 17:08:29 +0300 Subject: [PATCH 2/2] fix: validate -ttl bounds against Cassandra's TTL limits Reject non-positive ttl values and ttl values exceeding Cassandra's hard-coded 20 year (7300 day) TTL maximum, instead of letting them fail later with an opaque error from CQLSSTableWriter or silently producing no TTL (ttl=0 means "no TTL" in Cassandra). Co-Authored-By: Claude Sonnet 5 --- README.md | 2 ++ .../client/tools/migrator/MigratorTool.java | 11 +++++++++++ 2 files changed, 13 insertions(+) diff --git a/README.md b/README.md index 4c147dd..15e3779 100644 --- a/README.md +++ b/README.md @@ -353,6 +353,8 @@ Notes: behavior, where only historical points expire, not the latest value. * TTL is counted from the moment the SSTables are generated by this tool, not from the original timestamp of the data being migrated. +* `-ttl` must be a positive number of days and cannot exceed 7300 days (20 years), which is Cassandra's + hard-coded maximum TTL. # Troubleshooting diff --git a/src/main/java/org/thingsboard/client/tools/migrator/MigratorTool.java b/src/main/java/org/thingsboard/client/tools/migrator/MigratorTool.java index 720ed9e..d60fb9b 100644 --- a/src/main/java/org/thingsboard/client/tools/migrator/MigratorTool.java +++ b/src/main/java/org/thingsboard/client/tools/migrator/MigratorTool.java @@ -30,6 +30,9 @@ @Slf4j public class MigratorTool { + private static final long MAX_TTL_DAYS = 7300; // Cassandra's hard-coded max TTL is 20 years + private static final long MAX_TTL_SECONDS = TimeUnit.DAYS.toSeconds(MAX_TTL_DAYS); + public static void main(String[] args) { CommandLine cmd = parseArgs(args); @@ -64,7 +67,14 @@ public static void main(String[] args) { Long ttlSeconds = null; if (cmd.getOptionValue("ttl") != null) { long ttlDays = Long.parseLong(cmd.getOptionValue("ttl")); + if (ttlDays <= 0) { + throw new RuntimeException("Failed to parse ttl property: ttl must be a positive number of days!"); + } ttlSeconds = TimeUnit.DAYS.toSeconds(ttlDays); + if (ttlSeconds > MAX_TTL_SECONDS) { + throw new RuntimeException("Failed to parse ttl property: ttl of " + ttlDays + + " days exceeds Cassandra's maximum allowed TTL of " + MAX_TTL_DAYS + " days (20 years)!"); + } } new PgCaMigrator( @@ -127,6 +137,7 @@ private static CommandLine parseArgs(String[] args) { Option ttlOpt = new Option("ttl", "ttl", true, "TTL in days for migrated timeseries data (ts_kv_cf and ts_kv_partitions_cf). " + + "Must be a positive number not exceeding 7300 days (Cassandra's 20 year TTL limit). " + "If not set, migrated data will never expire. Does not affect ts_kv_latest_cf."); ttlOpt.setRequired(false); options.addOption(ttlOpt);