You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by mj...@apache.org on 2018/10/23 00:21:30 UTC
[kafka] branch trunk updated: MINOR: Prohibit setting StreamsConfig
commit.interval.ms to a negative value (#5809)
This is an automated email from the ASF dual-hosted git repository.
mjsax 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 e9bbfde MINOR: Prohibit setting StreamsConfig commit.interval.ms to a negative value (#5809)
e9bbfde is described below
commit e9bbfde10f1d73e500063817ada7a9ad474bca3f
Author: occho <ao...@gmail.com>
AuthorDate: Tue Oct 23 09:21:19 2018 +0900
MINOR: Prohibit setting StreamsConfig commit.interval.ms to a negative value (#5809)
Reviewers: Guozhang Wang <gu...@confluent.io>, Matthias J. Sax <ma...@confluent.io>, Bill Bejeck <bi...@confluent.io>
---
.../main/java/org/apache/kafka/streams/StreamsConfig.java | 1 +
.../streams/processor/internals/GlobalStreamThread.java | 2 +-
.../kafka/streams/processor/internals/StreamThread.java | 2 +-
.../java/org/apache/kafka/streams/StreamsConfigTest.java | 12 ++++++++++++
.../apache/kafka/streams/integration/EosIntegrationTest.java | 2 +-
.../kafka/streams/processor/internals/StateConsumerTest.java | 9 ---------
6 files changed, 16 insertions(+), 12 deletions(-)
diff --git a/streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java b/streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java
index b112a54..b25894c 100644
--- a/streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java
+++ b/streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java
@@ -582,6 +582,7 @@ public class StreamsConfig extends AbstractConfig {
.define(COMMIT_INTERVAL_MS_CONFIG,
Type.LONG,
DEFAULT_COMMIT_INTERVAL_MS,
+ atLeast(0),
Importance.LOW,
COMMIT_INTERVAL_MS_DOC)
.define(CONNECTIONS_MAX_IDLE_MS_CONFIG,
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStreamThread.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStreamThread.java
index 9d529c5..d91aedf 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStreamThread.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStreamThread.java
@@ -241,7 +241,7 @@ public class GlobalStreamThread extends Thread {
stateMaintainer.update(record);
}
final long now = time.milliseconds();
- if (flushInterval >= 0 && now >= lastFlush + flushInterval) {
+ if (now >= lastFlush + flushInterval) {
stateMaintainer.flushState();
lastFlush = now;
}
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java
index 3a19aa7..6c3e7cd 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java
@@ -1020,7 +1020,7 @@ public class StreamThread extends Thread {
boolean maybeCommit() {
int committed = 0;
- if (commitTimeMs >= 0 && now - lastCommitMs > commitTimeMs) {
+ if (now - lastCommitMs > commitTimeMs) {
if (log.isTraceEnabled()) {
log.trace("Committing all active tasks {} and standby tasks {} since {}ms has elapsed (commit interval is {}ms)",
taskManager.activeTaskIds(), taskManager.standbyTaskIds(), now - lastCommitMs, commitTimeMs);
diff --git a/streams/src/test/java/org/apache/kafka/streams/StreamsConfigTest.java b/streams/src/test/java/org/apache/kafka/streams/StreamsConfigTest.java
index 83279cc..a86c389 100644
--- a/streams/src/test/java/org/apache/kafka/streams/StreamsConfigTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/StreamsConfigTest.java
@@ -522,6 +522,18 @@ public class StreamsConfigTest {
}
@Test
+ public void shouldThrowExceptionIfCommitIntervalMsIsNegative() {
+ final long commitIntervalMs = -1;
+ props.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, commitIntervalMs);
+ try {
+ new StreamsConfig(props);
+ fail("Should throw ConfigException when commitIntervalMs is set to a negative value");
+ } catch (final ConfigException e) {
+ assertEquals("Invalid value -1 for configuration commit.interval.ms: Value must be at least 0", e.getMessage());
+ }
+ }
+
+ @Test
public void shouldUseNewConfigsWhenPresent() {
final Properties props = getStreamsConfig();
props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.Long().getClass());
diff --git a/streams/src/test/java/org/apache/kafka/streams/integration/EosIntegrationTest.java b/streams/src/test/java/org/apache/kafka/streams/integration/EosIntegrationTest.java
index 9aa0772..b2bd0a8 100644
--- a/streams/src/test/java/org/apache/kafka/streams/integration/EosIntegrationTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/integration/EosIntegrationTest.java
@@ -677,7 +677,7 @@ public class EosIntegrationTest {
{
put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, StreamsConfig.EXACTLY_ONCE);
put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, numberOfStreamsThreads);
- put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, -1);
+ put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, Long.MAX_VALUE);
put(StreamsConfig.consumerPrefix(ConsumerConfig.METADATA_MAX_AGE_CONFIG), "1000");
put(StreamsConfig.consumerPrefix(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG), "earliest");
put(StreamsConfig.consumerPrefix(ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG), 5 * 1000);
diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StateConsumerTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StateConsumerTest.java
index 140f705..f44d2b4 100644
--- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StateConsumerTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StateConsumerTest.java
@@ -109,15 +109,6 @@ public class StateConsumerTest {
}
@Test
- public void shouldNotFlushWhenFlushIntervalIsZero() {
- stateConsumer = new GlobalStreamThread.StateConsumer(logContext, consumer, stateMaintainer, time, Duration.ofMillis(10L), -1);
- stateConsumer.initialize();
- time.sleep(100);
- stateConsumer.pollAndUpdate();
- assertFalse(stateMaintainer.flushed);
- }
-
- @Test
public void shouldCloseConsumer() throws IOException {
stateConsumer.close();
assertTrue(consumer.closed());