引言
kafka 是一个重度依赖系统文件系统的message queue，消息都是直接存于硬盘，与一些基于内存的 message queue 速度相比，kafka 并不慢甚至更快，原因有顺序读写、零拷贝、日志分段等。而这些保存在硬盘中的数据也不能无休止地存在， kafka 也提供了多种机制来删除它们。由于大数据运营管理平台中需要纳管 kafka，用户可以在该平台上创建 topic，但存储是有限的，为了能够限制用户的存储使用量，故需要一种基于文件大小的消息删除机制，而与此息息相关的一个参数是 retention.bytes ，下文会从该参数出发探究该删除机制。
小试牛刀
kafka中有两个地方可以设置参数 retention.bytes：server 启动时的配置文件和topic 级别的配置（后者的设置会覆盖前者），官网对该参数的解释分别如下：
为了验证该参数，将某一 topic 的 retention.bytes 设为4096B，server 和 topic 中相关配置如下：
往该topic发送一些消息观察log 文件的大小，下图显示出 log 的大小为 4206B，已经超出了参数 retention.bytes 的值，但此时 kafka 并没有执行任何的删除策略。
retention.bytes 的探究
为了了解 retention.bytes 是如何生效的，故找到kafka删除 log 的相关代码如下：
可以看出，kafka 会启动一个定时任务去删除日志，每隔 retentionCheckMs 毫秒（由参数 log.retention.check.interval.ms 指定）就执行一次 cleanupLogs ，该函数定义如下：
cleanupLogs 对所有 log 调用了 deleteOldSegments 函数，并记录了删除的总量（日志清理策略 log.cleanup.policy 需设置为 delete）。
而 deleteOldSegments 会调用 deleteRetentionSizeBreachedSegments 来执行基于 retention.bytes 的删除算法：首先，它计算出最多可以删除的文件大小（diff = size - config.config.retentionSize），接下来调用的 deleteOldSegments(shouldDelete) 会从最老的日志分段（segments）开始，遍历所有的分段，对于小于diff并且不是 active segment （用于写入消息的那个segment为active segment，一个 topic 至少要有一个分段用于读写消息）则可以被删除。下图是 deleteOldSegments(shouldDelete) 中的一段关键代码。
至此，筛选出了满足删除条件的 segments，接下来就是真正的删除它们，其主要逻辑是先对可删除segment 的文件添加后缀名（.delete）以标识，再启动一个定时任务去执行删除操作，定时任务的时延通过参数 file.delete.delay.ms 指定。
再试牛刀
此次，参数retention.bytes 和 log.segment.bytes都设置成了128B， log.segment.bytes 是日志文件分段阈值，当分段超过该值时就会创建新的分段，这样才会出现可能会被删除旧的日志分段（old segment），为了快速验证，故将其设置为一个较小的值。
根据前文的删除算法，上图所示的 topic 的diff=112+75-128（retention.bytes）=22，即可以删除的数据大小为22B，但是old segment 只有一个大小113B的分段，超过了22B，所以不会删除任何分段。
继续往该topic中发送一些消息，如下图 diff=113+77+75-128=137，该值大于最老的分段（34 ），所以它被标记为可删除。此时diff更新为137-113=24，而所剩下的旧分段（37）大小为75B大于diff，不符合要求，所以此次删除操作中只会将34分段删除。
总结
基于retention.bytes 的日志删除策略并不是一个严格限制topic大小的参数，要想该参数生效需要将log.cleanup.policy 需设置为 delete，并且需要合理的设置 log.segment.bytes 的大小，只有当有 old segment出现并且删除某old segment 后，数据仍然超过我们设定的 retention.bytes，才会真的触发删除操作。
引言
kafka 是一个重度依赖系统文件系统的message queue，消息都是直接存于硬盘，与一些基于内存的 message queue 速度相比，kafka 并不慢甚至更快，原因有顺序读写、零拷贝、日志分段等。而这些保存在硬盘中的数据也不能无休止地存在， kafka 也提供了多种机制来删除它们。由于大数据运营管理平台中需要纳管 kafka，用户可以在该平台上创建 topic，但存储是有限的，为了能够限制用户的存储使用量，故需要一种基于文件大小的消息删除机制，而与此息息相关的一个参数是 retention.bytes ，下文会从该参数出发探究该删除机制。
小试牛刀
kafka中有两个地方可以设置参数 retention.bytes：server 启动时的配置文件和topic 级别的配置（后者的设置会覆盖前者），官网对该参数的解释分别如下：
为了验证该参数，将某一 topic 的 retention.bytes 设为4096B，server 和 topic 中相关配置如下：
往该topic发送一些消息观察log 文件的大小，下图显示出 log 的大小为 4206B，已经超出了参数 retention.bytes 的值，但此时 kafka 并没有执行任何的删除策略。
retention.bytes 的探究
为了了解 retention.bytes 是如何生效的，故找到kafka删除 log 的相关代码如下：
可以看出，kafka 会启动一个定时任务去删除日志，每隔 retentionCheckMs 毫秒（由参数 log.retention.check.interval.ms 指定）就执行一次 cleanupLogs ，该函数定义如下：
cleanupLogs 对所有 log 调用了 deleteOldSegments 函数，并记录了删除的总量（日志清理策略 log.cleanup.policy 需设置为 delete）。
而 deleteOldSegments 会调用 deleteRetentionSizeBreachedSegments 来执行基于 retention.bytes 的删除算法：首先，它计算出最多可以删除的文件大小（diff = size - config.config.retentionSize），接下来调用的 deleteOldSegments(shouldDelete) 会从最老的日志分段（segments）开始，遍历所有的分段，对于小于diff并且不是 active segment （用于写入消息的那个segment为active segment，一个 topic 至少要有一个分段用于读写消息）则可以被删除。下图是 deleteOldSegments(shouldDelete) 中的一段关键代码。
至此，筛选出了满足删除条件的 segments，接下来就是真正的删除它们，其主要逻辑是先对可删除segment 的文件添加后缀名（.delete）以标识，再启动一个定时任务去执行删除操作，定时任务的时延通过参数 file.delete.delay.ms 指定。
再试牛刀
此次，参数retention.bytes 和 log.segment.bytes都设置成了128B， log.segment.bytes 是日志文件分段阈值，当分段超过该值时就会创建新的分段，这样才会出现可能会被删除旧的日志分段（old segment），为了快速验证，故将其设置为一个较小的值。
根据前文的删除算法，上图所示的 topic 的diff=112+75-128（retention.bytes）=22，即可以删除的数据大小为22B，但是old segment 只有一个大小113B的分段，超过了22B，所以不会删除任何分段。
继续往该topic中发送一些消息，如下图 diff=113+77+75-128=137，该值大于最老的分段（34 ），所以它被标记为可删除。此时diff更新为137-113=24，而所剩下的旧分段（37）大小为75B大于diff，不符合要求，所以此次删除操作中只会将34分段删除。
总结
基于retention.bytes 的日志删除策略并不是一个严格限制topic大小的参数，要想该参数生效需要将log.cleanup.policy 需设置为 delete，并且需要合理的设置 log.segment.bytes 的大小，只有当有 old segment出现并且删除某old segment 后，数据仍然超过我们设定的 retention.bytes，才会真的触发删除操作。