diff --git a/core/src/main/scala/kafka/log/LogConfig.scala
b/core/src/main/scala/kafka/log/LogConfig.scala
index d872e09ed79..ab58949fd5d 100755
--- a/core/src/main/scala/kafka/log/LogConfig.scala
+++ b/core/src/main/scala/kafka/log/LogConfig.scala
@@ -232,7 +232,7 @@ object LogConfig {
.define(RetentionBytesProp, LONG, Defaults.RetentionSize, MEDIUM,
RetentionSizeDoc,
KafkaConfig.LogRetentionBytesProp)
// can be negative. See kafka.log.LogManager.cleanupExpiredSegments
- .define(RetentionMsProp, LONG, Defaults.RetentionMs, MEDIUM,
RetentionMsDoc,
+ .define(RetentionMsProp, LONG, Defaults.RetentionMs, atLeast(-1),
MEDIUM, RetentionMsDoc,
KafkaConfig.LogRetentionTimeMillisProp)
.define(MaxMessageBytesProp, INT, Defaults.MaxMessageSize, atLeast(0),
MEDIUM, MaxMessageSizeDoc,
KafkaConfig.MessageMaxBytesProp)
diff --git a/streams/src/main/java/org/apache/kafka/streams/StreamsBuilder.java
b/streams/src/main/java/org/apache/kafka/streams/StreamsBuilder.java
index 14b63e25a7b..c72ff9f8999 100644
--- a/streams/src/main/java/org/apache/kafka/streams/StreamsBuilder.java
+++ b/streams/src/main/java/org/apache/kafka/streams/StreamsBuilder.java
@@ -199,7 +199,7 @@
* You should only specify serdes in the {@link Consumed} instance as
these will also be used to overwrite the
* serdes in {@link Materialized}, i.e.,
* <pre> {@code
- * streamBuilder.table(topic, Consumed.with(Serde.String(),
Serde.String(), Materialized.<String, String, KeyValueStore<Bytes,
byte[]>as(storeName))
+ * streamBuilder.table(topic, Consumed.with(Serde.String(),
Serde.String()), Materialized.<String, String, KeyValueStore<Bytes,
byte[]>as(storeName))
* }
* </pre>
* To query the local {@link KeyValueStore} it must be obtained via
@@ -372,7 +372,7 @@
* You should only specify serdes in the {@link Consumed} instance as
these will also be used to overwrite the
* serdes in {@link Materialized}, i.e.,
* <pre> {@code
- * streamBuilder.globalTable(topic, Consumed.with(Serde.String(),
Serde.String(), Materialized.<String, String, KeyValueStore<Bytes,
byte[]>as(storeName))
+ * streamBuilder.globalTable(topic, Consumed.with(Serde.String(),
Serde.String()), Materialized.<String, String, KeyValueStore<Bytes,
byte[]>as(storeName))
* }
* </pre>
* To query the local {@link KeyValueStore} it must be obtained via
diff --git
a/streams/src/main/java/org/apache/kafka/streams/kstream/KStream.java
b/streams/src/main/java/org/apache/kafka/streams/kstream/KStream.java
index b961709b997..55bde9a867f 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/KStream.java
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/KStream.java
@@ -64,7 +64,7 @@
* @return a {@code KStream} that contains only those records that satisfy
the given predicate
* @see #filterNot(Predicate)
*/
- KStream<K, V> filter(Predicate<? super K, ? super V> predicate);
+ KStream<K, V> filter(final Predicate<? super K, ? super V> predicate);
/**
* Create a new {@code KStream} that consists all records of this stream
which do <em>not</em> satisfy the given
@@ -76,7 +76,7 @@
* @return a {@code KStream} that contains only those records that do
<em>not</em> satisfy the given predicate
* @see #filter(Predicate)
*/
- KStream<K, V> filterNot(Predicate<? super K, ? super V> predicate);
+ KStream<K, V> filterNot(final Predicate<? super K, ? super V> predicate);
/**
* Set a new key (with possibly new type) for each input record.
@@ -109,7 +109,7 @@
* @see #flatMapValues(ValueMapper)
* @see #flatMapValues(ValueMapperWithKey)
*/
- <KR> KStream<KR, V> selectKey(KeyValueMapper<? super K, ? super V, ?
extends KR> mapper);
+ <KR> KStream<KR, V> selectKey(final KeyValueMapper<? super K, ? super V, ?
extends KR> mapper);
/**
* Transform each record of the input stream into a new record in the
output stream (both key and value type can be
@@ -148,7 +148,7 @@
* @see #transformValues(ValueTransformerSupplier, String...)
* @see #transformValues(ValueTransformerWithKeySupplier, String...)
*/
- <KR, VR> KStream<KR, VR> map(KeyValueMapper<? super K, ? super V, ?
extends KeyValue<? extends KR, ? extends VR>> mapper);
+ <KR, VR> KStream<KR, VR> map(final KeyValueMapper<? super K, ? super V, ?
extends KeyValue<? extends KR, ? extends VR>> mapper);
/**
* Transform the value of each input record into a new value (with
possible new type) of the output record.
@@ -183,7 +183,7 @@
* @see #transformValues(ValueTransformerSupplier, String...)
* @see #transformValues(ValueTransformerWithKeySupplier, String...)
*/
- <VR> KStream<K, VR> mapValues(ValueMapper<? super V, ? extends VR> mapper);
+ <VR> KStream<K, VR> mapValues(final ValueMapper<? super V, ? extends VR>
mapper);
/**
* Transform the value of each input record into a new value (with
possible new type) of the output record.
diff --git
a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/suppress/EagerBufferConfigImpl.java
b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/suppress/EagerBufferConfigImpl.java
index 90456d9f008..e94abc1c844 100644
---
a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/suppress/EagerBufferConfigImpl.java
+++
b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/suppress/EagerBufferConfigImpl.java
@@ -75,6 +75,6 @@ public int hashCode() {
@Override
public String toString() {
- return "EagerBufferConfigImpl{maxKeys=" + maxRecords + ", maxBytes=" +
maxBytes + '}';
+ return "EagerBufferConfigImpl{maxRecords=" + maxRecords + ",
maxBytes=" + maxBytes + '}';
}
}
diff --git
a/streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopicConfig.java
b/streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopicConfig.java
index c9a334473ee..aa565e4aa5d 100644
---
a/streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopicConfig.java
+++
b/streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopicConfig.java
@@ -40,7 +40,7 @@
}
/**
- * Get the configured properties for this topic. If rententionMs is set
then
+ * Get the configured properties for this topic. If retentionMs is set then
* we add additionalRetentionMs to work out the desired retention when
cleanup.policy=compact,delete
*
* @param additionalRetentionMs - added to retention to allow for clock
drift etc
diff --git
a/streams/src/main/java/org/apache/kafka/streams/processor/internals/RepartitionTopicConfig.java
b/streams/src/main/java/org/apache/kafka/streams/processor/internals/RepartitionTopicConfig.java
index a7a8d88ca93..466520e3936 100644
---
a/streams/src/main/java/org/apache/kafka/streams/processor/internals/RepartitionTopicConfig.java
+++
b/streams/src/main/java/org/apache/kafka/streams/processor/internals/RepartitionTopicConfig.java
@@ -45,7 +45,7 @@
}
/**
- * Get the configured properties for this topic. If rententionMs is set
then
+ * Get the configured properties for this topic. If retentionMs is set then
* we add additionalRetentionMs to work out the desired retention when
cleanup.policy=compact,delete
*
* @param additionalRetentionMs - added to retention to allow for clock
drift etc
diff --git
a/streams/src/main/java/org/apache/kafka/streams/processor/internals/UnwindowedChangelogTopicConfig.java
b/streams/src/main/java/org/apache/kafka/streams/processor/internals/UnwindowedChangelogTopicConfig.java
index e55ce71ba84..acca8377843 100644
---
a/streams/src/main/java/org/apache/kafka/streams/processor/internals/UnwindowedChangelogTopicConfig.java
+++
b/streams/src/main/java/org/apache/kafka/streams/processor/internals/UnwindowedChangelogTopicConfig.java
@@ -40,7 +40,7 @@
}
/**
- * Get the configured properties for this topic. If rententionMs is set
then
+ * Get the configured properties for this topic. If retentionMs is set then
* we add additionalRetentionMs to work out the desired retention when
cleanup.policy=compact,delete
*
* @param additionalRetentionMs - added to retention to allow for clock
drift etc
diff --git
a/streams/src/main/java/org/apache/kafka/streams/processor/internals/WindowedChangelogTopicConfig.java
b/streams/src/main/java/org/apache/kafka/streams/processor/internals/WindowedChangelogTopicConfig.java
index e177bea3112..55d548c9795 100644
---
a/streams/src/main/java/org/apache/kafka/streams/processor/internals/WindowedChangelogTopicConfig.java
+++
b/streams/src/main/java/org/apache/kafka/streams/processor/internals/WindowedChangelogTopicConfig.java
@@ -42,7 +42,7 @@
}
/**
- * Get the configured properties for this topic. If rententionMs is set
then
+ * Get the configured properties for this topic. If retentionMs is set then
* we add additionalRetentionMs to work out the desired retention when
cleanup.policy=compact,delete
*
* @param additionalRetentionMs - added to retention to allow for clock
drift etc
@@ -57,7 +57,13 @@
topicConfig.putAll(topicConfigs);
if (retentionMs != null) {
- topicConfig.put(TopicConfig.RETENTION_MS_CONFIG,
String.valueOf(retentionMs + additionalRetentionMs));
+ long retentionValue;
+ try {
+ retentionValue = Math.addExact(retentionMs,
additionalRetentionMs);
+ } catch (final ArithmeticException swallow) {
+ retentionValue = Long.MAX_VALUE;
+ }
+ topicConfig.put(TopicConfig.RETENTION_MS_CONFIG,
String.valueOf(retentionValue));
}
return topicConfig;
With regards,
Apache Git Services
