kafka-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From jjko...@apache.org
Subject [kafka] branch trunk updated: MINOR: add docs for KIP-354 KAFKA-7321 (#6724)
Date Tue, 14 May 2019 05:54:23 GMT
This is an automated email from the ASF dual-hosted git repository.

jjkoshy pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git


The following commit(s) were added to refs/heads/trunk by this push:
     new b86e8c1  MINOR: add docs for KIP-354 KAFKA-7321 (#6724)
b86e8c1 is described below

commit b86e8c1ea908b8fa2576d058700b0ac14b1b4a3e
Author: Xiongqi Wu <xiongqi.wu@gmail.com>
AuthorDate: Mon May 13 22:54:07 2019 -0700

    MINOR: add docs for KIP-354 KAFKA-7321 (#6724)
    
    MINOR: update documentation for the log cleaner max compaction lag feature (KIP-354) implemented
in KAFKA-7321
    
    Author: Xiongqi Wu <xiowu@linkedin.com>
    Reviewer: Joel Koshy <jjkoshy@gmail.com>
---
 .../org/apache/kafka/common/config/TopicConfig.java  | 14 +++++++++-----
 core/src/main/scala/kafka/server/KafkaConfig.scala   |  9 +++++++--
 docs/configuration.html                              |  1 +
 docs/design.html                                     | 20 ++++++++++++++------
 4 files changed, 31 insertions(+), 13 deletions(-)

diff --git a/clients/src/main/java/org/apache/kafka/common/config/TopicConfig.java b/clients/src/main/java/org/apache/kafka/common/config/TopicConfig.java
index b1be6c8..f912cd2 100755
--- a/clients/src/main/java/org/apache/kafka/common/config/TopicConfig.java
+++ b/clients/src/main/java/org/apache/kafka/common/config/TopicConfig.java
@@ -104,6 +104,10 @@ public class TopicConfig {
     public static final String MIN_COMPACTION_LAG_MS_DOC = "The minimum time a message will
remain " +
         "uncompacted in the log. Only applicable for logs that are being compacted.";
 
+    public static final String MAX_COMPACTION_LAG_MS_CONFIG = "max.compaction.lag.ms";
+    public static final String MAX_COMPACTION_LAG_MS_DOC = "The maximum time a message will
remain " +
+        "ineligible for compaction in the log. Only applicable for logs that are being compacted.";
+
     public static final String MIN_CLEANABLE_DIRTY_RATIO_CONFIG = "min.cleanable.dirty.ratio";
     public static final String MIN_CLEANABLE_DIRTY_RATIO_DOC = "This configuration controls
how frequently " +
         "the log compactor will attempt to clean the log (assuming <a href=\"#compaction\">log
" +
@@ -111,11 +115,11 @@ public class TopicConfig {
         "50% of the log has been compacted. This ratio bounds the maximum space wasted in
" +
         "the log by duplicates (at 50% at most 50% of the log could be duplicates). A " +
         "higher ratio will mean fewer, more efficient cleanings but will mean more wasted
" +
-        "space in the log.";
-
-    public static final String MAX_COMPACTION_LAG_MS_CONFIG = "max.compaction.lag.ms";
-    public static final String MAX_COMPACTION_LAG_MS_DOC = "The maximum time a message will
remain " +
-        "ineligible for compaction in the log. Only applicable for logs that are being compacted.";
+        "space in the log. If the " + MAX_COMPACTION_LAG_MS_CONFIG + " or the " + MIN_COMPACTION_LAG_MS_CONFIG
+
+        " configurations are also specified, then the log compactor considers the log eligible
for compaction " +
+        "as soon as either: (i) the dirty ratio threshold has been met and the log has had
dirty (uncompacted) " +
+        "records for at least the " + MIN_COMPACTION_LAG_MS_CONFIG + " duration, or (ii)
if the log has had " +
+        "dirty (uncompacted) records for at most the " + MAX_COMPACTION_LAG_MS_CONFIG + "
period.";
 
     public static final String CLEANUP_POLICY_CONFIG = "cleanup.policy";
     public static final String CLEANUP_POLICY_COMPACT = "compact";
diff --git a/core/src/main/scala/kafka/server/KafkaConfig.scala b/core/src/main/scala/kafka/server/KafkaConfig.scala
index 526e174..2f98d3e 100755
--- a/core/src/main/scala/kafka/server/KafkaConfig.scala
+++ b/core/src/main/scala/kafka/server/KafkaConfig.scala
@@ -613,11 +613,16 @@ object KafkaConfig {
   val LogCleanerDedupeBufferLoadFactorDoc = "Log cleaner dedupe buffer load factor. The percentage
full the dedupe buffer can become. A higher value " +
   "will allow more log to be cleaned at once but will lead to more hash collisions"
   val LogCleanerBackoffMsDoc = "The amount of time to sleep when there are no logs to clean"
-  val LogCleanerMinCleanRatioDoc = "The minimum ratio of dirty log to total log for a log
to eligible for cleaning"
+  val LogCleanerMinCleanRatioDoc = "The minimum ratio of dirty log to total log for a log
to eligible for cleaning. " +
+    "If the " + LogCleanerMaxCompactionLagMsProp + " or the " + LogCleanerMinCompactionLagMsProp
+
+    " configurations are also specified, then the log compactor considers the log eligible
for compaction " +
+    "as soon as either: (i) the dirty ratio threshold has been met and the log has had dirty
(uncompacted) " +
+    "records for at least the " + LogCleanerMinCompactionLagMsProp + " duration, or (ii)
if the log has had " +
+    "dirty (uncompacted) records for at most the " + LogCleanerMaxCompactionLagMsProp + "
period."
   val LogCleanerEnableDoc = "Enable the log cleaner process to run on the server. Should
be enabled if using any topics with a cleanup.policy=compact including the internal offsets
topic. If disabled those topics will not be compacted and continually grow in size."
   val LogCleanerDeleteRetentionMsDoc = "How long are delete records retained?"
   val LogCleanerMinCompactionLagMsDoc = "The minimum time a message will remain uncompacted
in the log. Only applicable for logs that are being compacted."
-  val LogCleanerMaxCompactionLagMsDoc = "The maximum time a message will remain uncompacted
in the log. Only applicable for logs that are being compacted."
+  val LogCleanerMaxCompactionLagMsDoc = "The maximum time a message will remain ineligible
for compaction in the log. Only applicable for logs that are being compacted."
   val LogIndexSizeMaxBytesDoc = "The maximum size in bytes of the offset index"
   val LogIndexIntervalBytesDoc = "The interval with which we add an entry to the offset index"
   val LogFlushIntervalMessagesDoc = "The number of messages accumulated on a log partition
before messages are flushed to disk "
diff --git a/docs/configuration.html b/docs/configuration.html
index 06d7585..112c844 100644
--- a/docs/configuration.html
+++ b/docs/configuration.html
@@ -156,6 +156,7 @@
     <li><code>log.index.interval.bytes</code></li>
     <li><code>log.cleaner.delete.retention.ms</code></li>
     <li><code>log.cleaner.min.compaction.lag.ms</code></li>
+    <li><code>log.cleaner.max.compaction.lag.ms</code></li>
     <li><code>log.cleaner.min.cleanable.ratio</code></li>
     <li><code>log.cleanup.policy</code></li>
     <li><code>log.segment.delete.delay.ms</code></li>
diff --git a/docs/design.html b/docs/design.html
index 8c9b2eb..d389030 100644
--- a/docs/design.html
+++ b/docs/design.html
@@ -501,6 +501,7 @@
     <ol>
     <li>Any consumer that stays caught-up to within the head of the log will see every
message that is written; these messages will have sequential offsets. The topic's <code>min.compaction.lag.ms</code>
can be used to
     guarantee the minimum length of time must pass after a message is written before it could
be compacted. I.e. it provides a lower bound on how long each message will remain in the (uncompacted)
head.
+    The topic's <code>max.compaction.lag.ms</code> can be used to guarantee the
maximum delay between the time a message is written and the time the message becomes eligible
for compaction.
     <li>Ordering of messages is always maintained.  Compaction will never re-order
messages, just remove some.
     <li>The offset for a message never changes.  It is the permanent identifier for
a position in the log.
     <li>Any consumer progressing from the start of the log will see at least the final
state of all records in the order they were written.  Additionally, all delete markers for
deleted records will be seen, provided
@@ -525,18 +526,25 @@
     The log cleaner is enabled by default. This will start the pool of cleaner threads.
     To enable log cleaning on a particular topic, add the log-specific property
     <pre class="brush: text;"> log.cleanup.policy=compact</pre>
-    
-    The <code>log.cleanup.policy</code> property is a broker configuration setting
defined 
-    in the broker's <code>server.properties</code> file; it affects all of the
topics 
-    in the cluster that do not have a configuration override in place as documented 
+
+    The <code>log.cleanup.policy</code> property is a broker configuration setting
defined
+    in the broker's <code>server.properties</code> file; it affects all of the
topics
+    in the cluster that do not have a configuration override in place as documented
     <a href="/documentation.html#brokerconfigs">here</a>.
-    
+
     The log cleaner can be configured to retain a minimum amount of the uncompacted "head"
of the log. This is enabled by setting the compaction time lag.
     <pre class="brush: text;">  log.cleaner.min.compaction.lag.ms</pre>
 
     This can be used to prevent messages newer than a minimum message age from being subject
to compaction. If not set, all log segments are eligible for compaction except for the last
segment, i.e. the one currently
     being written to. The active segment will not be compacted even if all of its messages
are older than the minimum compaction time lag.
-    
+
+    The log cleaner can be configured to ensure a maximum delay after which the uncompacted
"head" of the log becomes eligible for log compaction.
+    <pre class="brush: text;">  log.cleaner.max.compaction.lag.ms</pre>
+
+    This can be used to prevent log with low produce rate from remaining ineligible for compaction
for an unbounded duration. If not set, logs that do not exceed min.cleanable.dirty.ratio are
not compacted.
+    Note that this compaction deadline is not a hard guarantee since it is still subjected
to the availability of log cleaner threads and the actual compaction time.
+    You will want to monitor the uncleanable-partitions-count, max-clean-time-secs and max-compaction-delay-secs
metrics.
+
     <p>
     Further cleaner configurations are described <a href="/documentation.html#brokerconfigs">here</a>.
 


Mime
View raw message