-
Notifications
You must be signed in to change notification settings - Fork 441
GEOMESA-3539 - ability to truncate (delete messages) from a topic whe… #3450
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Merged
Merged
Changes from 2 commits
Commits
Show all changes
3 commits
Select commit
Hold shift + click to select a range
File filter
Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
175 changes: 175 additions & 0 deletions
175
...fka-datastore/src/main/scala/org/locationtech/geomesa/kafka/data/KafkaTruncateTopic.scala
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,175 @@ | ||
| /*********************************************************************** | ||
| * Copyright (c) 2013-2025 General Atomics Integrated Intelligence, Inc. | ||
| * All rights reserved. This program and the accompanying materials | ||
| * are made available under the terms of the Apache License, Version 2.0 | ||
| * which accompanies this distribution and is available at | ||
| * https://www.apache.org/licenses/LICENSE-2.0 | ||
| ***********************************************************************/ | ||
|
|
||
| package org.locationtech.geomesa.kafka.data | ||
|
|
||
| import com.typesafe.scalalogging.LazyLogging | ||
| import org.apache.kafka.clients.admin.ListOffsetsResult.ListOffsetsResultInfo | ||
| import org.apache.kafka.clients.admin._ | ||
| import org.apache.kafka.common.TopicPartition | ||
| import org.apache.kafka.common.config.{ConfigResource, TopicConfig} | ||
|
|
||
| import java.util | ||
| import java.util.Collections | ||
| import java.util.concurrent.ExecutionException | ||
| import scala.collection.JavaConverters._ | ||
|
|
||
| class KafkaTruncateTopic(private val admin: Admin) extends LazyLogging { | ||
|
|
||
| /** | ||
| * Truncation of a Kafka topic. | ||
| * | ||
| * @param topic the topic name to truncate | ||
| */ | ||
| def truncate(topic: String): Unit = { | ||
|
|
||
| val topicPartitions = getTopicPartitions(topic) | ||
| val latestOffsets = getLatestOffsets(topicPartitions) | ||
|
|
||
| val deleteCleanupPolicy = hasDeleteCleanupPolicy(topic) | ||
|
|
||
| try { | ||
| if (!deleteCleanupPolicy) { | ||
| logger.debug(s"adding 'delete' cleanup policy to topic=$topic so it can be truncated.") | ||
| addDeleteCleanupPolicy(topic) | ||
| } | ||
| logger.debug(s"truncate: topic: $topic starting") | ||
| deleteRecords(latestOffsets) | ||
| logger.debug(s"truncate: topic: $topic completed") | ||
| } finally { | ||
| if (!deleteCleanupPolicy) { | ||
| logger.debug(s"removing 'delete' cleanup policy to topic=$topic.") | ||
| removeDeleteCleanupPolicy(topic) | ||
| } | ||
| } | ||
|
|
||
| logger.info(s"$topic truncated.") | ||
|
|
||
| } | ||
|
|
||
| private def deleteRecords(latestOffsets: Map[TopicPartition, ListOffsetsResultInfo]): Unit = | ||
| try { | ||
| val recordsToDelete = generateRecordsToDelete(latestOffsets) | ||
| admin.deleteRecords(recordsToDelete.asJava).all().get() | ||
| } catch { | ||
| case _: InterruptedException => | ||
| Thread.currentThread().interrupt() | ||
| throw new RuntimeException("the deleteRecords operation was interrupted, aborting; it may have still completed.") | ||
| case e: ExecutionException => | ||
| throw convertExecutionException(e) | ||
| } | ||
|
|
||
| /** | ||
| * For the given offsets, generate the RecordToDelete objects; this is used with the latest offsets so the topic | ||
| * will be truncated. | ||
| */ | ||
| private def generateRecordsToDelete(latestOffsets: Map[TopicPartition, ListOffsetsResultInfo]): Map[TopicPartition, RecordsToDelete] = | ||
| latestOffsets.map { | ||
| case (tp, info) => tp -> RecordsToDelete.beforeOffset(info.offset()) | ||
| } | ||
|
|
||
| /** | ||
| * for the list of partitions, return the latest offsets. | ||
| */ | ||
| private def getLatestOffsets(partitions: List[TopicPartition]): Map[TopicPartition, ListOffsetsResultInfo] = { | ||
| val input = partitions.map(tp => tp -> OffsetSpec.latest()).toMap | ||
| getOffsets(input) | ||
| } | ||
|
|
||
| /** | ||
| * used by getLatestOffsets() and getEarliestOffsets() for obtaining the offsets | ||
| */ | ||
| private def getOffsets(offsetSpecs: Map[TopicPartition, OffsetSpec]): Map[TopicPartition, ListOffsetsResultInfo] = | ||
| try { | ||
| admin.listOffsets(offsetSpecs.asJava).all().get().asScala.toMap | ||
| } catch { | ||
| case _: InterruptedException => | ||
| Thread.currentThread().interrupt() | ||
| throw new RuntimeException("listOffsets operation interrupted, deleteRecords will not executed.") | ||
| case e: ExecutionException => | ||
| throw convertExecutionException(e) | ||
| } | ||
|
|
||
| /** | ||
| * Get all TopicPartitions for the given topic; truncation is performed is performed at the partition level. | ||
| */ | ||
| private def getTopicPartitions(topic: String): List[TopicPartition] = | ||
| try { | ||
| val topicInfo = admin.describeTopics(Collections.singleton(topic)).allTopicNames().get().get(topic) | ||
| topicInfo.partitions().asScala.map(info => new TopicPartition(topic, info.partition())).toList | ||
| } catch { | ||
| case _: InterruptedException => | ||
| Thread.currentThread().interrupt() | ||
| throw new RuntimeException("describeTopics operation interrupted, deleteRecords will not executed.") | ||
| case e: ExecutionException => | ||
| throw convertExecutionException(e) | ||
| } | ||
|
|
||
| /** | ||
| * Check if the topic has the 'delete' cleanup policy. | ||
| */ | ||
| private def hasDeleteCleanupPolicy(topicName: String): Boolean = | ||
| try { | ||
| val configResource = new ConfigResource(ConfigResource.Type.TOPIC, topicName) | ||
| val configsResult = admin.describeConfigs(Collections.singleton(configResource)) | ||
| val config = configsResult.all().get().get(configResource) | ||
|
|
||
| config.get(TopicConfig.CLEANUP_POLICY_CONFIG).value().contains(TopicConfig.CLEANUP_POLICY_DELETE) | ||
| } catch { | ||
elahrvivaz marked this conversation as resolved.
Outdated
Show resolved
Hide resolved
|
||
| case _: InterruptedException => | ||
| Thread.currentThread().interrupt() | ||
| throw new RuntimeException("describeTopics operation interrupted, deleteRecords will not executed.") | ||
| case e: ExecutionException => | ||
| throw convertExecutionException(e) | ||
| } | ||
|
|
||
| /** | ||
| * Add the 'delete' cleanup policy to the topic's 'cleanup.policy' config. | ||
| */ | ||
| private def addDeleteCleanupPolicy(topic: String): Unit = | ||
| try { | ||
| val configResource = new ConfigResource(ConfigResource.Type.TOPIC, topic) | ||
| val alterConfigOp = new AlterConfigOp(new ConfigEntry(TopicConfig.CLEANUP_POLICY_CONFIG, TopicConfig.CLEANUP_POLICY_DELETE), AlterConfigOp.OpType.APPEND) | ||
| val configs: util.Map[ConfigResource, util.Collection[AlterConfigOp]] = Map(configResource -> alterConfigOpColl(alterConfigOp)).asJava | ||
| admin.incrementalAlterConfigs(configs).all().get() | ||
| } catch { | ||
| case _: InterruptedException => | ||
| Thread.currentThread().interrupt() | ||
| throw new RuntimeException("incrementalAlterConfigs operation interrupted, ...") | ||
| case e: ExecutionException => | ||
| throw convertExecutionException(e) | ||
| } | ||
|
|
||
| /** | ||
| * Remove the 'delete' cleanup policy to the topic's 'cleanup.policy' config. | ||
| */ | ||
| private def removeDeleteCleanupPolicy(topic: String): Unit = { | ||
| val configResource = new ConfigResource(ConfigResource.Type.TOPIC, topic) | ||
| val alterConfigOp = new AlterConfigOp(new ConfigEntry(TopicConfig.CLEANUP_POLICY_CONFIG, TopicConfig.CLEANUP_POLICY_DELETE), AlterConfigOp.OpType.SUBTRACT) | ||
| val configs: util.Map[ConfigResource, util.Collection[AlterConfigOp]] = Map(configResource -> alterConfigOpColl(alterConfigOp)).asJava | ||
| admin.incrementalAlterConfigs(configs).all().get() | ||
| } | ||
|
|
||
| /** | ||
| * Singleton wrapper for AlertConfigOp. | ||
| */ | ||
| private def alterConfigOpColl(alterConfigOp: AlterConfigOp): util.Collection[AlterConfigOp] = | ||
| Collections.singleton(alterConfigOp) | ||
|
|
||
| /** | ||
| * Convert ExecutionException to RuntimeException, preserving the underlying cause. | ||
| */ | ||
| private def convertExecutionException(e: ExecutionException): RuntimeException = { | ||
| val cause = Option(e.getCause).getOrElse(e) | ||
| new RuntimeException(cause.getMessage, cause) | ||
| } | ||
| } | ||
|
|
||
| object KafkaTruncateTopic { | ||
| def apply(admin: Admin): KafkaTruncateTopic = new KafkaTruncateTopic(admin) | ||
| } | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Uh oh!
There was an error while loading. Please reload this page.