You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by da...@apache.org on 2017/08/11 11:14:07 UTC
[1/5] kafka git commit: KAFKA-5702; extract refactor StreamThread
Repository: kafka
Updated Branches:
refs/heads/trunk b2b529522 -> 3e69ce801
http://git-wip-us.apache.org/repos/asf/kafka/blob/3e69ce80/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java
index 465f8be..4752227 100644
--- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java
@@ -16,7 +16,6 @@
*/
package org.apache.kafka.streams.processor.internals;
-import org.apache.kafka.clients.consumer.CommitFailedException;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
import org.apache.kafka.clients.consumer.ConsumerRecord;
@@ -39,28 +38,23 @@ import org.apache.kafka.streams.kstream.internals.InternalStreamsBuilder;
import org.apache.kafka.streams.kstream.internals.InternalStreamsBuilderTest;
import org.apache.kafka.streams.processor.StateRestoreListener;
import org.apache.kafka.streams.processor.TaskId;
-import org.apache.kafka.streams.processor.TopologyBuilder;
import org.apache.kafka.streams.processor.internals.assignment.AssignmentInfo;
import org.apache.kafka.streams.state.HostInfo;
import org.apache.kafka.streams.state.Stores;
import org.apache.kafka.test.MockClientSupplier;
-import org.apache.kafka.test.MockInternalTopicManager;
import org.apache.kafka.test.MockProcessorSupplier;
import org.apache.kafka.test.MockStateRestoreListener;
-import org.apache.kafka.test.MockStateStoreSupplier;
import org.apache.kafka.test.MockTimestampExtractor;
import org.apache.kafka.test.TestCondition;
import org.apache.kafka.test.TestUtils;
import org.easymock.EasyMock;
+import org.easymock.IAnswer;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
-import java.io.File;
-import java.io.IOException;
import java.lang.reflect.Field;
import java.nio.ByteBuffer;
-import java.nio.file.Files;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
@@ -77,11 +71,9 @@ import java.util.regex.Pattern;
import static java.util.Collections.EMPTY_SET;
import static org.hamcrest.CoreMatchers.equalTo;
-import static org.hamcrest.CoreMatchers.is;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
@@ -93,13 +85,14 @@ public class StreamThreadTest {
private final String applicationId = "stream-thread-test";
private final MockTime mockTime = new MockTime();
private final Metrics metrics = new Metrics();
- private final MockClientSupplier clientSupplier = new MockClientSupplier();
+ private MockClientSupplier clientSupplier = new MockClientSupplier();
private UUID processId = UUID.randomUUID();
private final InternalStreamsBuilder internalStreamsBuilder = new InternalStreamsBuilder(new InternalTopologyBuilder());
private InternalTopologyBuilder internalTopologyBuilder;
private final StreamsConfig config = new StreamsConfig(configProps(false));
private final String stateDir = TestUtils.tempDirectory().getPath();
private final StateDirectory stateDirectory = new StateDirectory("applicationId", stateDir, mockTime);
+ private StreamsMetadataState streamsMetadataState;
@Before
public void setUp() throws Exception {
@@ -107,6 +100,7 @@ public class StreamThreadTest {
internalTopologyBuilder = InternalStreamsBuilderTest.internalTopologyBuilder(internalStreamsBuilder);
internalTopologyBuilder.setApplicationId(applicationId);
+ streamsMetadataState = new StreamsMetadataState(internalTopologyBuilder, StreamsMetadataState.UNKNOWN_HOST);
}
private final TopicPartition t1p1 = new TopicPartition("topic1", 1);
@@ -267,9 +261,9 @@ public class StreamThreadTest {
final StreamThread thread = getStreamThread();
final Map<TaskId, Set<TopicPartition>> activeTasks = new HashMap<>();
- thread.setPartitionAssignor(new StreamPartitionAssignor() {
+ thread.setThreadMetadataProvider(new StreamPartitionAssignor() {
@Override
- Map<TaskId, Set<TopicPartition>> activeTasks() {
+ public Map<TaskId, Set<TopicPartition>> activeTasks() {
return activeTasks;
}
});
@@ -364,9 +358,9 @@ public class StreamThreadTest {
final StreamThread thread = getStreamThread();
final Map<TaskId, Set<TopicPartition>> activeTasks = new HashMap<>();
- thread.setPartitionAssignor(new StreamPartitionAssignor() {
+ thread.setThreadMetadataProvider(new StreamPartitionAssignor() {
@Override
- Map<TaskId, Set<TopicPartition>> activeTasks() {
+ public Map<TaskId, Set<TopicPartition>> activeTasks() {
return activeTasks;
}
});
@@ -451,19 +445,7 @@ public class StreamThreadTest {
@Test
public void testStateChangeStartClose() throws InterruptedException {
- final StreamThread thread = new StreamThread(
- internalTopologyBuilder,
- config,
- clientSupplier,
- applicationId,
- clientId,
- processId,
- metrics,
- Time.SYSTEM,
-
- new StreamsMetadataState(internalTopologyBuilder, StreamsMetadataState.UNKNOWN_HOST),
- 0,
- stateDirectory);
+ final StreamThread thread = createStreamThread(clientId, config, false);
final StateListenerStub stateListener = new StateListenerStub();
thread.setStateListener(stateListener);
@@ -486,6 +468,23 @@ public class StreamThreadTest {
assertEquals(thread.state(), StreamThread.State.DEAD);
}
+ private StreamThread createStreamThread(final String clientId, final StreamsConfig config, final boolean eosEnabled) {
+ if (eosEnabled) {
+ clientSupplier = new MockClientSupplier(applicationId);
+ }
+ return StreamThread.create(internalTopologyBuilder,
+ config,
+ clientSupplier,
+ processId,
+ clientId,
+ metrics,
+ mockTime,
+ streamsMetadataState,
+ 0,
+ stateDirectory,
+ new MockStateRestoreListener());
+ }
+
private final static String TOPIC = "topic";
private final Set<TopicPartition> task0Assignment = Collections.singleton(new TopicPartition(TOPIC, 0));
private final Set<TopicPartition> task1Assignment = Collections.singleton(new TopicPartition(TOPIC, 1));
@@ -505,30 +504,9 @@ public class StreamThreadTest {
//clientSupplier.consumer.assign(Arrays.asList(new TopicPartition(TOPIC, 0), new TopicPartition(TOPIC, 1)));
- final StreamThread thread1 = new StreamThread(
- internalTopologyBuilder,
- config,
- clientSupplier,
- applicationId,
- clientId + 1,
- processId,
- metrics,
- Time.SYSTEM,
- new StreamsMetadataState(internalTopologyBuilder, StreamsMetadataState.UNKNOWN_HOST),
- 0,
- stateDirectory);
- final StreamThread thread2 = new StreamThread(
- internalTopologyBuilder,
- config,
- clientSupplier,
- applicationId,
- clientId + 2,
- processId,
- metrics,
- Time.SYSTEM,
- new StreamsMetadataState(internalTopologyBuilder, StreamsMetadataState.UNKNOWN_HOST),
- 0,
- stateDirectory);
+ final StreamThread thread1 = createStreamThread(clientId + 1, config, false);
+ final StreamThread thread2 = createStreamThread(clientId + 2, config, false);
+
final Map<TaskId, Set<TopicPartition>> task0 = Collections.singletonMap(new TaskId(0, 0), task0Assignment);
final Map<TaskId, Set<TopicPartition>> task1 = Collections.singletonMap(new TaskId(0, 1), task1Assignment);
@@ -536,8 +514,8 @@ public class StreamThreadTest {
final Map<TaskId, Set<TopicPartition>> thread1Assignment = new HashMap<>(task0);
final Map<TaskId, Set<TopicPartition>> thread2Assignment = new HashMap<>(task1);
- thread1.setPartitionAssignor(new MockStreamsPartitionAssignor(thread1Assignment));
- thread2.setPartitionAssignor(new MockStreamsPartitionAssignor(thread2Assignment));
+ thread1.setThreadMetadataProvider(new MockStreamsPartitionAssignor(thread1Assignment));
+ thread2.setThreadMetadataProvider(new MockStreamsPartitionAssignor(thread2Assignment));
thread1.start();
@@ -615,12 +593,12 @@ public class StreamThreadTest {
}
@Override
- Map<TaskId, Set<TopicPartition>> activeTasks() {
+ public Map<TaskId, Set<TopicPartition>> activeTasks() {
return activeTaskAssignment;
}
@Override
- Map<TaskId, Set<TopicPartition>> standbyTasks() {
+ public Map<TaskId, Set<TopicPartition>> standbyTasks() {
return standbyTaskAssignment;
}
@@ -630,17 +608,7 @@ public class StreamThreadTest {
@Test
public void testMetrics() {
- final StreamThread thread = new StreamThread(
- internalTopologyBuilder,
- config,
- clientSupplier,
- applicationId,
- clientId,
- processId,
- metrics,
- mockTime,
- new StreamsMetadataState(internalTopologyBuilder, StreamsMetadataState.UNKNOWN_HOST),
- 0, stateDirectory);
+ final StreamThread thread = createStreamThread(clientId, config, false);
final String defaultGroupName = "stream-metrics";
final String defaultPrefix = "thread." + thread.threadClientId();
final Map<String, String> defaultTags = Collections.singletonMap("client-id", thread.threadClientId());
@@ -671,128 +639,105 @@ public class StreamThreadTest {
}
+ @SuppressWarnings({"unchecked", "ThrowableNotThrown"})
@Test
- public void testMaybeCommit() throws IOException, InterruptedException {
- final File baseDir = Files.createTempDirectory("test").toFile();
- try {
- final long commitInterval = 1000L;
- final Properties props = configProps(false);
- props.setProperty(StreamsConfig.STATE_DIR_CONFIG, baseDir.getCanonicalPath());
- props.setProperty(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, Long.toString(commitInterval));
-
- final StreamsConfig config = new StreamsConfig(props);
-
- internalTopologyBuilder.addSource(null, "source1", null, null, null, "topic1");
-
- final StreamThread thread = new StreamThread(
- internalTopologyBuilder,
- config,
- clientSupplier,
- applicationId,
- clientId,
- processId,
- metrics,
- mockTime,
- new StreamsMetadataState(internalTopologyBuilder, StreamsMetadataState.UNKNOWN_HOST),
- 0, stateDirectory) {
-
- @Override
- public void maybeCommit(final long now) {
- super.maybeCommit(now);
- }
-
- @Override
- protected StreamTask createStreamTask(final TaskId id, final Collection<TopicPartition> partitionsForTask) {
- final ProcessorTopology topology = builder.build(id.topicGroupId);
- return new TestStreamTask(
- id,
- applicationId,
- partitionsForTask,
- topology,
- consumer,
- clientSupplier.getProducer(new HashMap<String, Object>()),
- restoreConsumer,
- config,
- new MockStreamsMetrics(new Metrics()),
- stateDirectory);
- }
- };
-
- initPartitionGrouper(config, thread, clientSupplier);
-
- final ConsumerRebalanceListener rebalanceListener = thread.rebalanceListener;
+ public void shouldNotCommitBeforeTheCommitInterval() {
+ final long commitInterval = 1000L;
+ final Properties props = configProps(false);
+ props.setProperty(StreamsConfig.STATE_DIR_CONFIG, stateDir);
+ props.setProperty(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, Long.toString(commitInterval));
- final List<TopicPartition> revokedPartitions;
- final List<TopicPartition> assignedPartitions;
+ final StreamsConfig config = new StreamsConfig(props);
+ final Consumer<byte[], byte[]> consumer = EasyMock.createNiceMock(Consumer.class);
+ final TaskManager taskManager = mockTaskMangerCommit(consumer, 1);
- //
- // Assign t1p1 and t1p2. This should create Task 1 & 2
- //
- revokedPartitions = Collections.emptyList();
- assignedPartitions = Arrays.asList(t1p1, t1p2);
-
- thread.setState(StreamThread.State.RUNNING);
- thread.rebalanceListener.onPartitionsRevoked(Collections.<TopicPartition>emptyList());
- rebalanceListener.onPartitionsRevoked(revokedPartitions);
- rebalanceListener.onPartitionsAssigned(assignedPartitions);
+ StreamThread.StreamsMetricsThreadImpl streamsMetrics = new StreamThread.StreamsMetricsThreadImpl(metrics, "", "", Collections.<String, String>emptyMap());
+ final StreamThread thread = new StreamThread(internalTopologyBuilder,
+ clientId,
+ "",
+ config,
+ processId,
+ mockTime,
+ streamsMetadataState,
+ taskManager,
+ streamsMetrics,
+ clientSupplier,
+ consumer,
+ stateDirectory);
+ thread.maybeCommit(mockTime.milliseconds());
+ mockTime.sleep(commitInterval - 10L);
+ thread.maybeCommit(mockTime.milliseconds());
+
+ EasyMock.verify(taskManager);
+ }
- assertEquals(2, thread.tasks().size());
- // no task is committed before the commit interval
- mockTime.sleep(commitInterval - 10L);
- thread.maybeCommit(mockTime.milliseconds());
- for (final StreamTask task : thread.tasks().values()) {
- assertFalse(((TestStreamTask) task).committed);
- }
+ @SuppressWarnings("unchecked")
+ @Test
+ public void shouldCommitAfterTheCommitInterval() {
+ final long commitInterval = 1000L;
+ final Properties props = configProps(false);
+ props.setProperty(StreamsConfig.STATE_DIR_CONFIG, stateDir);
+ props.setProperty(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, Long.toString(commitInterval));
- // all tasks are committed after the commit interval
- mockTime.sleep(11L);
- thread.maybeCommit(mockTime.milliseconds());
- for (final StreamTask task : thread.tasks().values()) {
- assertTrue(((TestStreamTask) task).committed);
- ((TestStreamTask) task).committed = false;
- }
+ final StreamsConfig config = new StreamsConfig(props);
+ final Consumer<byte[], byte[]> consumer = EasyMock.createNiceMock(Consumer.class);
+ final TaskManager taskManager = mockTaskMangerCommit(consumer, 2);
- // no task is committed before the commit interval, again
- mockTime.sleep(commitInterval - 10L);
- thread.maybeCommit(mockTime.milliseconds());
- for (final StreamTask task : thread.tasks().values()) {
- assertFalse(((TestStreamTask) task).committed);
- }
+ StreamThread.StreamsMetricsThreadImpl streamsMetrics = new StreamThread.StreamsMetricsThreadImpl(metrics, "", "", Collections.<String, String>emptyMap());
+ final StreamThread thread = new StreamThread(internalTopologyBuilder,
+ clientId,
+ "",
+ config,
+ processId,
+ mockTime,
+ streamsMetadataState,
+ taskManager,
+ streamsMetrics,
+ clientSupplier,
+ consumer,
+ stateDirectory);
+ thread.maybeCommit(mockTime.milliseconds());
+ mockTime.sleep(commitInterval + 1);
+ thread.maybeCommit(mockTime.milliseconds());
+
+ EasyMock.verify(taskManager);
+ }
- // all tasks are committed after the commit interval, again
- mockTime.sleep(11L);
- thread.maybeCommit(mockTime.milliseconds());
- for (final StreamTask task : thread.tasks().values()) {
- assertTrue(((TestStreamTask) task).committed);
- ((TestStreamTask) task).committed = false;
+ @SuppressWarnings({"ThrowableNotThrown", "unchecked"})
+ private TaskManager mockTaskMangerCommit(final Consumer<byte[], byte[]> consumer, final int numberOfCommits) {
+ final TaskManager taskManager = EasyMock.createMock(TaskManager.class);
+ taskManager.setConsumer(EasyMock.anyObject(Consumer.class));
+ EasyMock.expectLastCall();
+ IAnswer<Object> checkCommitAction = new IAnswer<Object>() {
+ @Override
+ public Object answer() throws Throwable {
+ final Object[] currentArguments = EasyMock.getCurrentArguments();
+ TaskManager.TaskAction action = (TaskManager.TaskAction) currentArguments[0];
+ if (!action.name().equals("commit")) {
+ throw new IllegalArgumentException("expected to get commit action but was:" + action.name());
+ }
+ return null;
}
- } finally {
- Utils.delete(baseDir);
- }
+ };
+ taskManager.performOnActiveTasks(EasyMock.anyObject(TaskManager.TaskAction.class));
+ EasyMock.expectLastCall().andAnswer(checkCommitAction).times(numberOfCommits);
+ taskManager.performOnStandbyTasks(EasyMock.anyObject(TaskManager.TaskAction.class));
+ EasyMock.expectLastCall().andAnswer(checkCommitAction).times(numberOfCommits);
+ EasyMock.replay(taskManager, consumer);
+ return taskManager;
}
@Test
public void shouldInjectSharedProducerForAllTasksUsingClientSupplierOnCreateIfEosDisabled() throws InterruptedException {
internalTopologyBuilder.addSource(null, "source1", null, null, null, "someTopic");
- final StreamThread thread = new StreamThread(
- internalTopologyBuilder,
- config,
- clientSupplier,
- applicationId,
- clientId,
- processId,
- metrics,
- mockTime,
- new StreamsMetadataState(internalTopologyBuilder, StreamsMetadataState.UNKNOWN_HOST),
- 0,
- stateDirectory);
+ final StreamThread thread = createStreamThread(clientId, config, false);
final Map<TaskId, Set<TopicPartition>> assignment = new HashMap<>();
assignment.put(new TaskId(0, 0), Collections.singleton(new TopicPartition("someTopic", 0)));
assignment.put(new TaskId(0, 1), Collections.singleton(new TopicPartition("someTopic", 1)));
- thread.setPartitionAssignor(new MockStreamsPartitionAssignor(assignment));
+ thread.setThreadMetadataProvider(new MockStreamsPartitionAssignor(assignment));
thread.setState(StreamThread.State.RUNNING);
thread.rebalanceListener.onPartitionsRevoked(Collections.<TopicPartition>emptyList());
@@ -800,9 +745,8 @@ public class StreamThreadTest {
assertEquals(1, clientSupplier.producers.size());
final Producer globalProducer = clientSupplier.producers.get(0);
- assertSame(globalProducer, thread.threadProducer);
- for (final StreamTask task : thread.tasks().values()) {
- assertSame(globalProducer, ((RecordCollectorImpl) task.recordCollector()).producer());
+ for (final Task task : thread.tasks().values()) {
+ assertSame(globalProducer, ((RecordCollectorImpl) ((StreamTask) task).recordCollector()).producer());
}
assertSame(clientSupplier.consumer, thread.consumer);
assertSame(clientSupplier.restoreConsumer, thread.restoreConsumer);
@@ -812,25 +756,13 @@ public class StreamThreadTest {
public void shouldInjectProducerPerTaskUsingClientSupplierOnCreateIfEosEnable() throws InterruptedException {
internalTopologyBuilder.addSource(null, "source1", null, null, null, "someTopic");
- final MockClientSupplier clientSupplier = new MockClientSupplier(applicationId);
- final StreamThread thread = new StreamThread(
- internalTopologyBuilder,
- new StreamsConfig(configProps(true)),
- clientSupplier,
- applicationId,
- clientId,
- processId,
- metrics,
- mockTime,
- new StreamsMetadataState(internalTopologyBuilder, StreamsMetadataState.UNKNOWN_HOST),
- 0,
- stateDirectory);
+ final StreamThread thread = createStreamThread(clientId, new StreamsConfig(configProps(true)), true);
final Map<TaskId, Set<TopicPartition>> assignment = new HashMap<>();
assignment.put(new TaskId(0, 0), Collections.singleton(new TopicPartition("someTopic", 0)));
assignment.put(new TaskId(0, 1), Collections.singleton(new TopicPartition("someTopic", 1)));
assignment.put(new TaskId(0, 2), Collections.singleton(new TopicPartition("someTopic", 2)));
- thread.setPartitionAssignor(new MockStreamsPartitionAssignor(assignment));
+ thread.setThreadMetadataProvider(new MockStreamsPartitionAssignor(assignment));
final Set<TopicPartition> assignedPartitions = new HashSet<>();
Collections.addAll(assignedPartitions, new TopicPartition("someTopic", 0), new TopicPartition("someTopic", 2));
@@ -838,11 +770,10 @@ public class StreamThreadTest {
thread.rebalanceListener.onPartitionsRevoked(Collections.<TopicPartition>emptyList());
thread.rebalanceListener.onPartitionsAssigned(assignedPartitions);
- assertNull(thread.threadProducer);
assertEquals(thread.tasks().size(), clientSupplier.producers.size());
final Iterator it = clientSupplier.producers.iterator();
- for (final StreamTask task : thread.tasks().values()) {
- assertSame(it.next(), ((RecordCollectorImpl) task.recordCollector()).producer());
+ for (final Task task : thread.tasks().values()) {
+ assertSame(it.next(), ((RecordCollectorImpl) ((StreamTask) task).recordCollector()).producer());
}
assertSame(clientSupplier.consumer, thread.consumer);
assertSame(clientSupplier.restoreConsumer, thread.restoreConsumer);
@@ -852,24 +783,12 @@ public class StreamThreadTest {
public void shouldCloseAllTaskProducersOnCloseIfEosEnabled() throws InterruptedException {
internalTopologyBuilder.addSource(null, "source1", null, null, null, "someTopic");
- final MockClientSupplier clientSupplier = new MockClientSupplier(applicationId);
- final StreamThread thread = new StreamThread(
- internalTopologyBuilder,
- new StreamsConfig(configProps(true)),
- clientSupplier,
- applicationId,
- clientId,
- processId,
- metrics,
- mockTime,
- new StreamsMetadataState(internalTopologyBuilder, StreamsMetadataState.UNKNOWN_HOST),
- 0,
- stateDirectory);
+ final StreamThread thread = createStreamThread(clientId, new StreamsConfig(configProps(true)), true);
final Map<TaskId, Set<TopicPartition>> assignment = new HashMap<>();
assignment.put(new TaskId(0, 0), Collections.singleton(new TopicPartition("someTopic", 0)));
assignment.put(new TaskId(0, 1), Collections.singleton(new TopicPartition("someTopic", 1)));
- thread.setPartitionAssignor(new MockStreamsPartitionAssignor(assignment));
+ thread.setThreadMetadataProvider(new MockStreamsPartitionAssignor(assignment));
thread.setState(StreamThread.State.RUNNING);
thread.rebalanceListener.onPartitionsRevoked(Collections.<TopicPartition>emptyList());
@@ -878,41 +797,39 @@ public class StreamThreadTest {
thread.close();
thread.run();
- for (final StreamTask task : thread.tasks().values()) {
- assertTrue(((MockProducer) ((RecordCollectorImpl) task.recordCollector()).producer()).closed());
+ for (final Task task : thread.tasks().values()) {
+ assertTrue(((MockProducer) ((RecordCollectorImpl) ((StreamTask) task).recordCollector()).producer()).closed());
}
}
@Test
public void shouldCloseThreadProducerOnCloseIfEosDisabled() throws InterruptedException {
- internalTopologyBuilder.addSource(null, "source1", null, null, null, "someTopic");
+ final Consumer<byte[], byte[]> consumer = EasyMock.createNiceMock(Consumer.class);
+ final TaskManager taskManager = EasyMock.createNiceMock(TaskManager.class);
- final StreamThread thread = new StreamThread(
- internalTopologyBuilder,
- config,
- clientSupplier,
- applicationId,
- clientId,
- processId,
- metrics,
- mockTime,
- new StreamsMetadataState(internalTopologyBuilder, StreamsMetadataState.UNKNOWN_HOST),
- 0,
- stateDirectory);
-
- final Map<TaskId, Set<TopicPartition>> assignment = new HashMap<>();
- assignment.put(new TaskId(0, 0), Collections.singleton(new TopicPartition("someTopic", 0)));
- assignment.put(new TaskId(0, 1), Collections.singleton(new TopicPartition("someTopic", 1)));
- thread.setPartitionAssignor(new MockStreamsPartitionAssignor(assignment));
+ taskManager.setConsumer(EasyMock.anyObject(Consumer.class));
+ EasyMock.expectLastCall();
+ taskManager.closeProducer();
+ EasyMock.expectLastCall();
+ EasyMock.replay(taskManager, consumer);
+ StreamThread.StreamsMetricsThreadImpl streamsMetrics = new StreamThread.StreamsMetricsThreadImpl(metrics, "", "", Collections.<String, String>emptyMap());
+ final StreamThread thread = new StreamThread(internalTopologyBuilder,
+ clientId,
+ "",
+ config,
+ processId,
+ mockTime,
+ streamsMetadataState,
+ taskManager,
+ streamsMetrics,
+ clientSupplier,
+ consumer,
+ stateDirectory);
thread.setState(StreamThread.State.RUNNING);
- thread.rebalanceListener.onPartitionsRevoked(Collections.<TopicPartition>emptyList());
- thread.rebalanceListener.onPartitionsAssigned(Collections.singleton(new TopicPartition("someTopic", 0)));
-
thread.close();
thread.run();
-
- assertTrue(((MockProducer) thread.threadProducer).closed());
+ EasyMock.verify(taskManager);
}
@Test
@@ -920,150 +837,18 @@ public class StreamThreadTest {
internalTopologyBuilder.addSource(null, "name", null, null, null, "topic");
internalTopologyBuilder.addSink("out", "output", null, null, null);
- final StreamThread thread = new StreamThread(internalTopologyBuilder,
- config,
- clientSupplier,
- applicationId,
- clientId,
- processId,
- metrics,
- mockTime,
- new StreamsMetadataState(internalTopologyBuilder, StreamsMetadataState.UNKNOWN_HOST),
- 0,
- stateDirectory);
-
- thread.setPartitionAssignor(new StreamPartitionAssignor() {
- @Override
- Map<TaskId, Set<TopicPartition>> standbyTasks() {
- return Collections.singletonMap(new TaskId(0, 0), Utils.mkSet(new TopicPartition("topic", 0)));
- }
- });
-
- thread.setState(StreamThread.State.RUNNING);
- thread.rebalanceListener.onPartitionsRevoked(Collections.<TopicPartition>emptyList());
- thread.rebalanceListener.onPartitionsAssigned(Collections.<TopicPartition>emptyList());
- }
-
- @Test
- public void shouldNotCloseSuspendedTaskswice() throws Exception {
- internalTopologyBuilder.addSource(null, "name", null, null, null, "topic");
- internalTopologyBuilder.addSink("out", "output", null, null, null);
-
- final TestStreamTask testStreamTask = new TestStreamTask(
- new TaskId(0, 0),
- applicationId,
- Utils.mkSet(new TopicPartition("topic", 0)),
- internalTopologyBuilder.build(0),
- clientSupplier.consumer,
- clientSupplier.getProducer(new HashMap<String, Object>()),
- clientSupplier.restoreConsumer,
- config,
- new MockStreamsMetrics(new Metrics()),
- new StateDirectory(applicationId, config.getString(StreamsConfig.STATE_DIR_CONFIG), mockTime));
-
- final StreamThread thread = new StreamThread(
- internalTopologyBuilder,
- config,
- clientSupplier,
- applicationId,
- clientId,
- processId,
- metrics,
- mockTime,
- new StreamsMetadataState(internalTopologyBuilder, StreamsMetadataState.UNKNOWN_HOST),
- 0,
- stateDirectory) {
-
- @Override
- protected StreamTask createStreamTask(final TaskId id, final Collection<TopicPartition> partitionsForTask) {
- return testStreamTask;
- }
- };
-
- final Set<TopicPartition> activeTasks = new HashSet<>();
- activeTasks.add(new TopicPartition("topic", 0));
-
- thread.setPartitionAssignor(new StreamPartitionAssignor() {
- @Override
- Map<TaskId, Set<TopicPartition>> activeTasks() {
- return new HashMap<TaskId, Set<TopicPartition>>() {
- {
- put(new TaskId(0, 0), activeTasks);
- }
- };
- }
- });
- thread.setState(StreamThread.State.RUNNING);
- thread.setState(StreamThread.State.PARTITIONS_REVOKED);
- thread.rebalanceListener.onPartitionsAssigned(activeTasks);
- thread.rebalanceListener.onPartitionsRevoked(activeTasks);
-
- assertTrue(testStreamTask.suspended);
- assertFalse(testStreamTask.closed);
-
- activeTasks.clear();
- // this should succeed without exception
- thread.rebalanceListener.onPartitionsAssigned(Collections.<TopicPartition>emptyList());
-
- assertTrue(testStreamTask.closed);
- }
-
- @Test
- public void shouldInitializeRestoreConsumerWithOffsetsFromStandbyTasks() {
- internalStreamsBuilder.stream(null, null, null, null, "t1").groupByKey().count("count-one");
- internalStreamsBuilder.stream(null, null, null, null, "t2").groupByKey().count("count-two");
-
- final StreamThread thread = new StreamThread(
- internalTopologyBuilder,
- config,
- clientSupplier,
- applicationId,
- clientId,
- processId,
- metrics,
- mockTime,
- new StreamsMetadataState(internalTopologyBuilder, StreamsMetadataState.UNKNOWN_HOST),
- 0,
- stateDirectory);
-
- final MockConsumer<byte[], byte[]> restoreConsumer = clientSupplier.restoreConsumer;
- restoreConsumer.updatePartitions("stream-thread-test-count-one-changelog",
- Collections.singletonList(new PartitionInfo("stream-thread-test-count-one-changelog",
- 0,
- null,
- new Node[0],
- new Node[0])));
- restoreConsumer.updatePartitions("stream-thread-test-count-two-changelog",
- Collections.singletonList(new PartitionInfo("stream-thread-test-count-two-changelog",
- 0,
- null,
- new Node[0],
- new Node[0])));
+ final StreamThread thread = createStreamThread(clientId, config, false);
- final Map<TaskId, Set<TopicPartition>> standbyTasks = new HashMap<>();
- final TopicPartition t1 = new TopicPartition("t1", 0);
- standbyTasks.put(new TaskId(0, 0), Utils.mkSet(t1));
-
- thread.setPartitionAssignor(new StreamPartitionAssignor() {
+ thread.setThreadMetadataProvider(new StreamPartitionAssignor() {
@Override
- Map<TaskId, Set<TopicPartition>> standbyTasks() {
- return standbyTasks;
+ public Map<TaskId, Set<TopicPartition>> standbyTasks() {
+ return Collections.singletonMap(new TaskId(0, 0), Utils.mkSet(new TopicPartition("topic", 0)));
}
});
thread.setState(StreamThread.State.RUNNING);
thread.rebalanceListener.onPartitionsRevoked(Collections.<TopicPartition>emptyList());
thread.rebalanceListener.onPartitionsAssigned(Collections.<TopicPartition>emptyList());
-
- assertThat(restoreConsumer.assignment(), equalTo(Utils.mkSet(new TopicPartition("stream-thread-test-count-one-changelog", 0))));
-
- // assign an existing standby plus a new one
- standbyTasks.put(new TaskId(1, 0), Utils.mkSet(new TopicPartition("t2", 0)));
- thread.rebalanceListener.onPartitionsRevoked(Collections.<TopicPartition>emptyList());
- thread.rebalanceListener.onPartitionsAssigned(Collections.<TopicPartition>emptyList());
-
- assertThat(restoreConsumer.assignment(), equalTo(Utils.mkSet(new TopicPartition("stream-thread-test-count-one-changelog", 0),
- new TopicPartition("stream-thread-test-count-two-changelog", 0))));
}
@Test
@@ -1071,18 +856,7 @@ public class StreamThreadTest {
internalStreamsBuilder.stream(null, null, null, null, "t1").groupByKey().count("count-one");
internalStreamsBuilder.stream(null, null, null, null, "t2").groupByKey().count("count-two");
- final StreamThread thread = new StreamThread(
- internalTopologyBuilder,
- config,
- clientSupplier,
- applicationId,
- clientId,
- processId,
- metrics,
- mockTime,
- new StreamsMetadataState(internalTopologyBuilder, StreamsMetadataState.UNKNOWN_HOST),
- 0,
- stateDirectory);
+ final StreamThread thread = createStreamThread(clientId, config, false);
final MockConsumer<byte[], byte[]> restoreConsumer = clientSupplier.restoreConsumer;
restoreConsumer.updatePartitions("stream-thread-test-count-one-changelog",
Collections.singletonList(new PartitionInfo("stream-thread-test-count-one-changelog",
@@ -1112,14 +886,14 @@ public class StreamThreadTest {
final TopicPartition t2 = new TopicPartition("t2", 0);
activeTasks.put(new TaskId(1, 0), Utils.mkSet(t2));
- thread.setPartitionAssignor(new StreamPartitionAssignor() {
+ thread.setThreadMetadataProvider(new StreamPartitionAssignor() {
@Override
- Map<TaskId, Set<TopicPartition>> standbyTasks() {
+ public Map<TaskId, Set<TopicPartition>> standbyTasks() {
return standbyTasks;
}
@Override
- Map<TaskId, Set<TopicPartition>> activeTasks() {
+ public Map<TaskId, Set<TopicPartition>> activeTasks() {
return activeTasks;
}
});
@@ -1139,105 +913,11 @@ public class StreamThreadTest {
}
@Test
- public void shouldCloseActiveTasksThatAreAssignedToThisStreamThreadButAssignmentHasChangedBeforeCreatingNewTasks() throws Exception {
- internalStreamsBuilder.stream(null, null, null, null, Pattern.compile("t.*")).to("out");
-
- final Map<Collection<TopicPartition>, TestStreamTask> createdTasks = new HashMap<>();
-
- final StreamThread thread = new StreamThread(
- internalTopologyBuilder,
- config,
- clientSupplier,
- applicationId,
- clientId,
- processId,
- metrics,
- mockTime,
- new StreamsMetadataState(internalTopologyBuilder, StreamsMetadataState.UNKNOWN_HOST),
- 0,
- stateDirectory) {
-
- @Override
- protected StreamTask createStreamTask(final TaskId id, final Collection<TopicPartition> partitions) {
- final ProcessorTopology topology = builder.build(id.topicGroupId);
- final TestStreamTask task = new TestStreamTask(
- id,
- applicationId,
- partitions,
- topology,
- consumer,
- clientSupplier.getProducer(new HashMap<String, Object>()),
- restoreConsumer,
- config,
- new MockStreamsMetrics(new Metrics()),
- stateDirectory);
- createdTasks.put(partitions, task);
- return task;
- }
- };
-
- final Map<TaskId, Set<TopicPartition>> activeTasks = new HashMap<>();
- final TopicPartition t1 = new TopicPartition("t1", 0);
- final Set<TopicPartition> task00Partitions = new HashSet<>();
- task00Partitions.add(t1);
- final TaskId taskId = new TaskId(0, 0);
- activeTasks.put(taskId, task00Partitions);
-
- thread.setPartitionAssignor(new StreamPartitionAssignor() {
- @Override
- Map<TaskId, Set<TopicPartition>> activeTasks() {
- return activeTasks;
- }
- });
-
- final StreamPartitionAssignor.SubscriptionUpdates subscriptionUpdates = new StreamPartitionAssignor.SubscriptionUpdates();
- final Field updatedTopicsField = subscriptionUpdates.getClass().getDeclaredField("updatedTopicSubscriptions");
- updatedTopicsField.setAccessible(true);
- final Set<String> updatedTopics = (Set<String>) updatedTopicsField.get(subscriptionUpdates);
- updatedTopics.add(t1.topic());
- internalTopologyBuilder.updateSubscriptions(subscriptionUpdates, null);
-
- // should create task for id 0_0 with a single partition
- thread.setState(StreamThread.State.RUNNING);
-
- thread.rebalanceListener.onPartitionsRevoked(Collections.<TopicPartition>emptyList());
- thread.rebalanceListener.onPartitionsAssigned(task00Partitions);
-
- final TestStreamTask firstTask = createdTasks.get(task00Partitions);
- assertThat(firstTask.id(), is(taskId));
-
- // update assignment for the task 0_0 so it now has 2 partitions
- task00Partitions.add(new TopicPartition("t2", 0));
- updatedTopics.add("t2");
-
- thread.rebalanceListener.onPartitionsRevoked(Collections.<TopicPartition>emptyList());
- thread.rebalanceListener.onPartitionsAssigned(task00Partitions);
-
- // should close the first task as the assignment has changed
- assertTrue("task should have been closed as assignment has changed", firstTask.closed);
- assertTrue("tasks state manager should have been closed as assignment has changed", firstTask.closedStateManager);
- // should have created a new task for 00
- assertThat(createdTasks.get(task00Partitions).id(), is(taskId));
- }
-
- @Test
public void shouldCloseTaskAsZombieAndRemoveFromActiveTasksIfProducerWasFencedWhileProcessing() throws Exception {
internalTopologyBuilder.addSource(null, "source", null, null, null, TOPIC);
internalTopologyBuilder.addSink("sink", "dummyTopic", null, null, null, "source");
- final MockClientSupplier clientSupplier = new MockClientSupplier(applicationId);
- final StreamThread thread = new StreamThread(
- internalTopologyBuilder,
- new StreamsConfig(configProps(true)),
- clientSupplier,
- applicationId,
- clientId,
- processId,
- metrics,
- mockTime,
- new StreamsMetadataState(internalTopologyBuilder, StreamsMetadataState.UNKNOWN_HOST),
- 0,
- stateDirectory);
+ final StreamThread thread = createStreamThread(clientId, new StreamsConfig(configProps(true)), true);
final MockConsumer consumer = clientSupplier.consumer;
consumer.updatePartitions(TOPIC, Collections.singletonList(new PartitionInfo(TOPIC, 0, null, null, null)));
@@ -1245,7 +925,7 @@ public class StreamThreadTest {
final Map<TaskId, Set<TopicPartition>> activeTasks = new HashMap<>();
activeTasks.put(task1, task0Assignment);
- thread.setPartitionAssignor(new MockStreamsPartitionAssignor(activeTasks));
+ thread.setThreadMetadataProvider(new MockStreamsPartitionAssignor(activeTasks));
thread.start();
TestUtils.waitForCondition(new TestCondition() {
@@ -1323,24 +1003,12 @@ public class StreamThreadTest {
internalTopologyBuilder.addSource(null, "name", null, null, null, "topic");
internalTopologyBuilder.addSink("out", "output", null, null, null);
- final MockClientSupplier clientSupplier = new MockClientSupplier(applicationId);
- final StreamThread thread = new StreamThread(
- internalTopologyBuilder,
- new StreamsConfig(configProps(true)),
- clientSupplier,
- applicationId,
- clientId,
- processId,
- new Metrics(),
- new MockTime(),
- new StreamsMetadataState(internalTopologyBuilder, StreamsMetadataState.UNKNOWN_HOST),
- 0,
- stateDirectory);
+ final StreamThread thread = createStreamThread(clientId, new StreamsConfig(configProps(true)), true);
final Map<TaskId, Set<TopicPartition>> activeTasks = new HashMap<>();
activeTasks.put(task1, task0Assignment);
- thread.setPartitionAssignor(new MockStreamsPartitionAssignor(activeTasks));
+ thread.setThreadMetadataProvider(new MockStreamsPartitionAssignor(activeTasks));
thread.setState(StreamThread.State.RUNNING);
thread.rebalanceListener.onPartitionsRevoked(null);
@@ -1358,320 +1026,12 @@ public class StreamThreadTest {
}
@Test
- public void shouldNotViolateAtLeastOnceWhenAnExceptionOccursOnTaskCloseDuringShutdown() throws Exception {
- internalStreamsBuilder.stream(null, null, null, null, "t1").groupByKey();
-
- final TestStreamTask testStreamTask = new TestStreamTask(
- new TaskId(0, 0),
- applicationId,
- Utils.mkSet(new TopicPartition("t1", 0)),
- internalTopologyBuilder.build(0),
- clientSupplier.consumer,
- clientSupplier.getProducer(new HashMap<String, Object>()),
- clientSupplier.restoreConsumer,
- config,
- new MockStreamsMetrics(new Metrics()),
- new StateDirectory(applicationId, config.getString(StreamsConfig.STATE_DIR_CONFIG), mockTime)) {
-
- @Override
- public void close(final boolean clean) {
- throw new RuntimeException("KABOOM!");
- }
- };
-
- final StreamThread thread = new StreamThread(
- internalTopologyBuilder,
- config,
- clientSupplier,
- applicationId,
- clientId,
- processId,
- metrics,
- mockTime,
- new StreamsMetadataState(internalTopologyBuilder, StreamsMetadataState.UNKNOWN_HOST),
- 0,
- stateDirectory) {
-
- @Override
- protected StreamTask createStreamTask(final TaskId id, final Collection<TopicPartition> partitions) {
- return testStreamTask;
- }
- };
-
- final Map<TaskId, Set<TopicPartition>> activeTasks = new HashMap<>();
- activeTasks.put(testStreamTask.id(), testStreamTask.partitions);
-
- thread.setPartitionAssignor(new MockStreamsPartitionAssignor(activeTasks));
-
- thread.setState(StreamThread.State.RUNNING);
- thread.rebalanceListener.onPartitionsRevoked(Collections.<TopicPartition>emptyList());
- thread.rebalanceListener.onPartitionsAssigned(testStreamTask.partitions);
-
- thread.close();
- thread.join();
- assertFalse("task shouldn't have been committed as there was an exception during shutdown", testStreamTask.committed);
- }
-
- @Test
- public void shouldNotViolateAtLeastOnceWhenAnExceptionOccursOnTaskFlushDuringShutdown() throws Exception {
- final MockStateStoreSupplier.MockStateStore stateStore = new MockStateStoreSupplier.MockStateStore("foo", false);
- internalStreamsBuilder.stream(null, null, null, null, "t1").groupByKey().count(new MockStateStoreSupplier(stateStore));
- final TestStreamTask testStreamTask = new TestStreamTask(
- new TaskId(0, 0),
- applicationId,
- Utils.mkSet(new TopicPartition("t1", 0)),
- internalTopologyBuilder.build(0),
- clientSupplier.consumer,
- clientSupplier.getProducer(new HashMap<String, Object>()),
- clientSupplier.restoreConsumer,
- config,
- new MockStreamsMetrics(metrics),
- new StateDirectory(applicationId, config.getString(StreamsConfig.STATE_DIR_CONFIG), mockTime)) {
-
- @Override
- public void flushState() {
- throw new RuntimeException("KABOOM!");
- }
- };
-
- final StreamThread thread = new StreamThread(
- internalTopologyBuilder,
- config,
- clientSupplier,
- applicationId,
- clientId,
- processId,
- metrics,
- mockTime,
- new StreamsMetadataState(internalTopologyBuilder, StreamsMetadataState.UNKNOWN_HOST),
- 0,
- stateDirectory) {
-
- @Override
- protected StreamTask createStreamTask(final TaskId id, final Collection<TopicPartition> partitions) {
- return testStreamTask;
- }
- };
-
-
- final Map<TaskId, Set<TopicPartition>> activeTasks = new HashMap<>();
- activeTasks.put(testStreamTask.id(), testStreamTask.partitions);
-
-
- thread.setPartitionAssignor(new MockStreamsPartitionAssignor(activeTasks));
-
- thread.start();
- TestUtils.waitForCondition(new TestCondition() {
- @Override
- public boolean conditionMet() {
- return thread.state() == StreamThread.State.RUNNING;
- }
- }, 10 * 1000, "Thread never started.");
- thread.rebalanceListener.onPartitionsRevoked(Collections.<TopicPartition>emptyList());
- thread.rebalanceListener.onPartitionsAssigned(testStreamTask.partitions);
- // store should have been opened
- assertTrue(stateStore.isOpen());
-
- thread.close();
- thread.join();
- assertFalse("task shouldn't have been committed as there was an exception during shutdown", testStreamTask.committed);
- // store should be closed even if we had an exception
- assertFalse(stateStore.isOpen());
- }
-
- @Test
- public void shouldCaptureCommitFailedExceptionOnTaskSuspension() throws Exception {
- internalStreamsBuilder.stream(null, null, null, null, "t1");
-
- final TestStreamTask testStreamTask = new TestStreamTask(
- new TaskId(0, 0),
- applicationId,
- Utils.mkSet(new TopicPartition("t1", 0)),
- internalTopologyBuilder.build(0),
- clientSupplier.consumer,
- clientSupplier.getProducer(new HashMap<String, Object>()),
- clientSupplier.restoreConsumer,
- config,
- new MockStreamsMetrics(new Metrics()),
- new StateDirectory(applicationId, config.getString(StreamsConfig.STATE_DIR_CONFIG), mockTime)) {
-
- @Override
- public void suspend() {
- throw new CommitFailedException();
- }
- };
-
- final StreamThread thread = new StreamThread(
- internalTopologyBuilder,
- config,
- clientSupplier,
- applicationId,
- clientId,
- processId,
- metrics,
- mockTime,
- new StreamsMetadataState(internalTopologyBuilder, StreamsMetadataState.UNKNOWN_HOST),
- 0,
- stateDirectory) {
-
- @Override
- protected StreamTask createStreamTask(final TaskId id, final Collection<TopicPartition> partitions) {
- return testStreamTask;
- }
- };
-
- final Map<TaskId, Set<TopicPartition>> activeTasks = new HashMap<>();
- activeTasks.put(testStreamTask.id(), testStreamTask.partitions);
-
- thread.setPartitionAssignor(new MockStreamsPartitionAssignor(activeTasks));
- thread.setState(StreamThread.State.RUNNING);
-
- thread.rebalanceListener.onPartitionsRevoked(Collections.<TopicPartition>emptyList());
- thread.rebalanceListener.onPartitionsAssigned(testStreamTask.partitions);
-
- thread.rebalanceListener.onPartitionsRevoked(Collections.<TopicPartition>emptyList());
-
- assertFalse(testStreamTask.committed);
- }
-
-
- @Test
- public void shouldNotViolateAtLeastOnceWhenExceptionOccursDuringTaskSuspension() throws Exception {
- internalStreamsBuilder.stream(null, null, null, null, "t1").groupByKey();
-
- final TestStreamTask testStreamTask = new TestStreamTask(
- new TaskId(0, 0),
- applicationId,
- Utils.mkSet(new TopicPartition("t1", 0)),
- internalTopologyBuilder.build(0),
- clientSupplier.consumer,
- clientSupplier.getProducer(new HashMap<String, Object>()),
- clientSupplier.restoreConsumer,
- config,
- new MockStreamsMetrics(new Metrics()),
- new StateDirectory(applicationId, config.getString(StreamsConfig.STATE_DIR_CONFIG), mockTime)) {
-
- @Override
- public void suspend() {
- throw new RuntimeException("KABOOM!");
- }
- };
-
- final StreamThread thread = new StreamThread(
- internalTopologyBuilder,
- config,
- clientSupplier,
- applicationId,
- clientId,
- processId,
- metrics,
- mockTime,
- new StreamsMetadataState(internalTopologyBuilder, StreamsMetadataState.UNKNOWN_HOST),
- 0,
- stateDirectory) {
-
- @Override
- protected StreamTask createStreamTask(final TaskId id, final Collection<TopicPartition> partitions) {
- return testStreamTask;
- }
- };
-
-
- final Map<TaskId, Set<TopicPartition>> activeTasks = new HashMap<>();
- activeTasks.put(testStreamTask.id(), testStreamTask.partitions);
-
-
- thread.setPartitionAssignor(new MockStreamsPartitionAssignor(activeTasks));
-
- thread.setState(StreamThread.State.RUNNING);
- thread.rebalanceListener.onPartitionsRevoked(Collections.<TopicPartition>emptyList());
- thread.rebalanceListener.onPartitionsAssigned(testStreamTask.partitions);
- try {
- thread.rebalanceListener.onPartitionsRevoked(Collections.<TopicPartition>emptyList());
- fail("should have thrown exception");
- } catch (final Exception e) {
- // expected
- }
- assertFalse(testStreamTask.committed);
- }
-
- @Test
- public void shouldNotViolateAtLeastOnceWhenExceptionOccursDuringFlushStateWhileSuspendingState() throws Exception {
- internalStreamsBuilder.stream(null, null, null, null, "t1").groupByKey();
-
- final TestStreamTask testStreamTask = new TestStreamTask(
- new TaskId(0, 0),
- applicationId,
- Utils.mkSet(new TopicPartition("t1", 0)),
- internalTopologyBuilder.build(0),
- clientSupplier.consumer,
- clientSupplier.getProducer(new HashMap<String, Object>()),
- clientSupplier.restoreConsumer,
- config,
- new MockStreamsMetrics(new Metrics()),
- new StateDirectory(applicationId, config.getString(StreamsConfig.STATE_DIR_CONFIG), mockTime)) {
-
- @Override
- protected void flushState() {
- throw new RuntimeException("KABOOM!");
- }
- };
-
- final StreamThread thread = new StreamThread(
- internalTopologyBuilder,
- config,
- clientSupplier,
- applicationId,
- clientId,
- processId,
- metrics,
- mockTime,
- new StreamsMetadataState(internalTopologyBuilder, StreamsMetadataState.UNKNOWN_HOST), 0, stateDirectory) {
-
- @Override
- protected StreamTask createStreamTask(final TaskId id, final Collection<TopicPartition> partitions) {
- return testStreamTask;
- }
- };
-
-
- final Map<TaskId, Set<TopicPartition>> activeTasks = new HashMap<>();
- activeTasks.put(testStreamTask.id(), testStreamTask.partitions);
-
-
- thread.setPartitionAssignor(new MockStreamsPartitionAssignor(activeTasks));
-
- thread.setState(StreamThread.State.RUNNING);
- thread.rebalanceListener.onPartitionsRevoked(Collections.<TopicPartition>emptyList());
- thread.rebalanceListener.onPartitionsAssigned(testStreamTask.partitions);
- try {
- thread.rebalanceListener.onPartitionsRevoked(Collections.<TopicPartition>emptyList());
- fail("should have thrown exception");
- } catch (final Exception e) {
- // expected
- }
- assertFalse(testStreamTask.committed);
- }
-
-
- @Test
@SuppressWarnings("unchecked")
public void shouldAlwaysUpdateWithLatestTopicsFromStreamPartitionAssignor() throws Exception {
internalTopologyBuilder.addSource(null, "source", null, null, null, Pattern.compile("t.*"));
internalTopologyBuilder.addProcessor("processor", new MockProcessorSupplier(), "source");
- final StreamThread thread = new StreamThread(
- internalTopologyBuilder,
- config,
- clientSupplier,
- applicationId,
- clientId,
- processId,
- metrics,
- mockTime,
- new StreamsMetadataState(internalTopologyBuilder, StreamsMetadataState.UNKNOWN_HOST),
- 0,
- stateDirectory);
+ final StreamThread thread = createStreamThread(clientId, config, false);
final StreamPartitionAssignor partitionAssignor = new StreamPartitionAssignor();
final Map<String, Object> configurationMap = new HashMap<>();
@@ -1680,7 +1040,7 @@ public class StreamThreadTest {
configurationMap.put(StreamsConfig.NUM_STANDBY_REPLICAS_CONFIG, 0);
partitionAssignor.configure(configurationMap);
- thread.setPartitionAssignor(partitionAssignor);
+ thread.setThreadMetadataProvider(partitionAssignor);
final Field nodeToSourceTopicsField =
internalTopologyBuilder.getClass().getDeclaredField("nodeToSourceTopics");
@@ -1739,240 +1099,6 @@ public class StreamThreadTest {
}
- @Test
- public void shouldReleaseStateDirLockIfFailureOnTaskSuspend() throws Exception {
- final TaskId taskId = new TaskId(0, 0);
-
- final StateDirectory stateDirMock = mockStateDirInteractions(taskId);
- final StreamThread thread = setupTest(taskId, stateDirMock);
-
- try {
- thread.rebalanceListener.onPartitionsRevoked(Collections.<TopicPartition>emptyList());
- fail("Should have thrown exception");
- } catch (final Exception e) {
- //
- } finally {
- thread.close();
- }
-
- EasyMock.verify(stateDirMock);
- }
-
- @Test
- public void shouldReleaseStateDirLockIfFailureOnTaskCloseForSuspendedTask() throws Exception {
- final TaskId taskId = new TaskId(0, 0);
-
- final StateDirectory stateDirMock = mockStateDirInteractions(taskId);
-
- final StreamThread thread = setupTest(taskId, stateDirMock);
- thread.close();
- thread.join();
- EasyMock.verify(stateDirMock);
- }
-
-
- private StreamThread setupTest(final TaskId taskId, final StateDirectory stateDirectory) throws InterruptedException {
- final TopologyBuilder builder = new TopologyBuilder();
- builder.setApplicationId(applicationId);
- builder.addSource("source", "topic");
-
- final MockClientSupplier clientSupplier = new MockClientSupplier();
-
- final TestStreamTask testStreamTask = new TestStreamTask(taskId,
- applicationId,
- Utils.mkSet(new TopicPartition("topic", 0)),
- builder.build(0),
- clientSupplier.consumer,
- clientSupplier.getProducer(new HashMap<String, Object>()),
- clientSupplier.restoreConsumer,
- config,
- new MockStreamsMetrics(new Metrics()),
- stateDirectory) {
-
- @Override
- public void suspend() {
- throw new RuntimeException("KABOOM!!!");
- }
- };
-
- final StreamThread thread = new StreamThread(
- builder.internalTopologyBuilder,
- config,
- clientSupplier,
- applicationId,
- clientId,
- processId,
- new Metrics(),
- new MockTime(),
- new StreamsMetadataState(builder.internalTopologyBuilder, StreamsMetadataState.UNKNOWN_HOST),
- 0, stateDirectory) {
-
- @Override
- protected StreamTask createStreamTask(final TaskId id, final Collection<TopicPartition> partitions) {
- return testStreamTask;
- }
- };
-
- final Map<TaskId, Set<TopicPartition>> activeTasks = new HashMap<>();
- activeTasks.put(testStreamTask.id, testStreamTask.partitions);
- thread.setPartitionAssignor(new MockStreamsPartitionAssignor(activeTasks));
- thread.start();
-
- TestUtils.waitForCondition(new TestCondition() {
- @Override
- public boolean conditionMet() {
- return thread.state() == StreamThread.State.RUNNING;
- }
- }, "thread didn't transition to running");
- thread.rebalanceListener.onPartitionsRevoked(Collections.<TopicPartition>emptySet());
- thread.rebalanceListener.onPartitionsAssigned(testStreamTask.partitions);
-
- return thread;
- }
-
- @Test
- public void shouldReleaseStateDirLockIfFailureOnStandbyTaskSuspend() throws Exception {
- final TaskId taskId = new TaskId(0, 0);
-
- final StateDirectory stateDirMock = mockStateDirInteractions(taskId);
- final StreamThread thread = setupStandbyTest(taskId, stateDirMock);
-
- startThreadAndRebalance(thread);
-
- try {
- thread.rebalanceListener.onPartitionsRevoked(Collections.<TopicPartition>emptyList());
- fail("Should have thrown exception");
- } catch (final Exception e) {
- // ok
- } finally {
- thread.close();
- }
- EasyMock.verify(stateDirMock);
- }
-
- private void startThreadAndRebalance(final StreamThread thread) throws InterruptedException {
- thread.start();
- TestUtils.waitForCondition(new TestCondition() {
- @Override
- public boolean conditionMet() {
- return thread.state() == StreamThread.State.RUNNING;
- }
- }, "thread didn't transition to running");
- thread.rebalanceListener.onPartitionsRevoked(Collections.<TopicPartition>emptySet());
- thread.rebalanceListener.onPartitionsAssigned(Collections.<TopicPartition>emptySet());
- }
-
- @Test
- public void shouldReleaseStateDirLockIfFailureOnStandbyTaskCloseForUnassignedSuspendedStandbyTask() throws Exception {
- final TaskId taskId = new TaskId(0, 0);
-
- final StateDirectory stateDirMock = mockStateDirInteractions(taskId);
- final StreamThread thread = setupStandbyTest(taskId, stateDirMock);
- startThreadAndRebalance(thread);
-
-
- try {
- thread.close();
- thread.join();
- } finally {
- thread.close();
- }
- EasyMock.verify(stateDirMock);
- }
-
- private StateDirectory mockStateDirInteractions(final TaskId taskId) throws IOException {
- final StateDirectory stateDirMock = EasyMock.createNiceMock(StateDirectory.class);
- EasyMock.expect(stateDirMock.lock(EasyMock.eq(taskId), EasyMock.anyInt())).andReturn(true);
- EasyMock.expect(stateDirMock.directoryForTask(taskId)).andReturn(new File(stateDir));
- stateDirMock.unlock(taskId);
- EasyMock.expectLastCall();
- EasyMock.replay(stateDirMock);
- return stateDirMock;
- }
-
- private StreamThread setupStandbyTest(final TaskId taskId, final StateDirectory stateDirectory) {
- final String storeName = "store";
- final String changelogTopic = applicationId + "-" + storeName + "-changelog";
-
- internalStreamsBuilder.stream(null, null, null, null, "topic1").groupByKey().count(storeName);
-
- final MockClientSupplier clientSupplier = new MockClientSupplier();
- clientSupplier.restoreConsumer.updatePartitions(changelogTopic,
- Collections.singletonList(new PartitionInfo(changelogTopic, 0, null, null, null)));
- clientSupplier.restoreConsumer.updateBeginningOffsets(new HashMap<TopicPartition, Long>() {
- {
- put(new TopicPartition(changelogTopic, 0), 0L);
- }
- });
- clientSupplier.restoreConsumer.updateEndOffsets(new HashMap<TopicPartition, Long>() {
- {
- put(new TopicPartition(changelogTopic, 0), 0L);
- }
- });
-
- final StreamThread thread = new StreamThread(
- internalTopologyBuilder,
- config,
- clientSupplier,
- applicationId,
- clientId,
- processId,
- new Metrics(),
- new MockTime(),
- new StreamsMetadataState(internalTopologyBuilder, StreamsMetadataState.UNKNOWN_HOST),
- 0,
- stateDirectory) {
-
- @Override
- protected StandbyTask createStandbyTask(final TaskId id, final Collection<TopicPartition> partitions) {
- return new StandbyTask(
- taskId,
- applicationId,
- partitions,
- builder.build(0),
- clientSupplier.consumer,
- new StoreChangelogReader(getName(), clientSupplier.restoreConsumer, mockTime, 1000,
- new MockStateRestoreListener()),
- StreamThreadTest.this.config,
- new StreamsMetricsImpl(new Metrics(), "groupName", Collections.<String, String>emptyMap()),
- stateDirectory) {
-
- @Override
- public void suspend() {
- throw new RuntimeException("KABOOM!!!");
- }
-
- @Override
- public void commit() {
- throw new RuntimeException("KABOOM!!!");
- }
- };
- }
- };
-
- final Map<TaskId, Set<TopicPartition>> standbyTasks = new HashMap<>();
- standbyTasks.put(taskId, Collections.singleton(new TopicPartition("topic", 0)));
- thread.setPartitionAssignor(new MockStreamsPartitionAssignor(Collections.<TaskId, Set<TopicPartition>>emptyMap(), standbyTasks));
-
- return thread;
- }
-
- private void initPartitionGrouper(final StreamsConfig config,
- final StreamThread thread,
- final MockClientSupplier clientSupplier) {
-
- final StreamPartitionAssignor partitionAssignor = new StreamPartitionAssignor();
-
- partitionAssignor.configure(config.getConsumerConfigs(thread, thread.applicationId, thread.clientId));
- final MockInternalTopicManager internalTopicManager = new MockInternalTopicManager(thread.config, clientSupplier.restoreConsumer);
- partitionAssignor.setInternalTopicManager(internalTopicManager);
-
- final Map<String, PartitionAssignor.Assignment> assignments =
- partitionAssignor.assign(metadata, Collections.singletonMap("client", subscription));
-
- partitionAssignor.onAssignment(assignments.get("client"));
- }
-
private static class StateListenerStub implements StreamThread.StateListener {
int numChanges = 0;
ThreadStateTransitionValidator oldState = null;
@@ -1993,34 +1119,6 @@ public class StreamThreadTest {
}
private StreamThread getStreamThread() {
- return new StreamThread(
- internalTopologyBuilder,
- config,
- clientSupplier,
- applicationId,
- clientId,
- processId,
- metrics,
- Time.SYSTEM,
- new StreamsMetadataState(internalTopologyBuilder, StreamsMetadataState.UNKNOWN_HOST),
- 0,
- stateDirectory) {
-
- @Override
- protected StreamTask createStreamTask(final TaskId id, final Collection<TopicPartition> partitionsForTask) {
- final ProcessorTopology topology = builder.build(id.topicGroupId);
- return new TestStreamTask(
- id,
- applicationId,
- partitionsForTask,
- topology,
- consumer,
- clientSupplier.getProducer(new HashMap()),
- restoreConsumer,
- config,
- new MockStreamsMetrics(new Metrics()),
- stateDirectory);
- }
- };
+ return createStreamThread(clientId, config, false);
}
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/3e69ce80/streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java
new file mode 100644
index 0000000..0c31a81
--- /dev/null
+++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java
@@ -0,0 +1,240 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kafka.streams.processor.internals;
+
+import org.apache.kafka.clients.consumer.CommitFailedException;
+import org.apache.kafka.clients.consumer.Consumer;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.utils.MockTime;
+import org.apache.kafka.common.utils.Time;
+import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.streams.errors.StreamsException;
+import org.apache.kafka.streams.processor.TaskId;
+import org.easymock.EasyMock;
+import org.easymock.EasyMockRunner;
+import org.easymock.Mock;
+import org.easymock.MockType;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+import java.util.Collections;
+import java.util.Map;
+import java.util.Set;
+
+import static org.easymock.EasyMock.checkOrder;
+import static org.easymock.EasyMock.expect;
+import static org.easymock.EasyMock.expectLastCall;
+import static org.easymock.EasyMock.replay;
+import static org.easymock.EasyMock.verify;
+import static org.junit.Assert.fail;
+
+@RunWith(EasyMockRunner.class)
+public class TaskManagerTest {
+
+ private final Time time = new MockTime();
+ private final TaskId taskId0 = new TaskId(0, 0);
+ private final TopicPartition t1p0 = new TopicPartition("t1", 0);
+ private final TopicPartition t1p1 = new TopicPartition("t1", 1);
+ private final Set<TopicPartition> taskId0Partitions = Utils.mkSet(t1p0);
+ private final Map<TaskId, Set<TopicPartition>> taskId0Assignment = Collections.singletonMap(taskId0, taskId0Partitions);
+
+ @Mock(type = MockType.NICE)
+ private ChangelogReader changeLogReader;
+ @Mock(type = MockType.NICE)
+ private Consumer<byte[], byte[]> restoreConsumer;
+ @Mock(type = MockType.NICE)
+ private Consumer<byte[], byte[]> consumer;
+ @Mock(type = MockType.NICE)
+ private StreamThread.AbstractTaskCreator activeTaskCreator;
+ @Mock(type = MockType.NICE)
+ private StreamThread.AbstractTaskCreator standbyTaskCreator;
+ @Mock(type = MockType.NICE)
+ private ThreadMetadataProvider threadMetadataProvider;
+ @Mock(type = MockType.NICE)
+ private Task firstTask;
+
+ private TaskManager taskManager;
+
+
+ @Before
+ public void setUp() throws Exception {
+ taskManager = new TaskManager(changeLogReader, time, "", restoreConsumer, activeTaskCreator, standbyTaskCreator);
+ taskManager.setThreadMetadataProvider(threadMetadataProvider);
+ taskManager.setConsumer(consumer);
+ }
+
+ @SuppressWarnings("unchecked")
+ @Test
+ public void shouldCloseActiveUnAssignedSuspendedTasksBeforeCreatingNewTasksWhenTaskPartitionAssignmentHasChanged() {
+ final Set<TopicPartition> secondPartitionAssignment = Utils.mkSet(t1p0, t1p1);
+ final Map<TaskId, Set<TopicPartition>> secondAssignment = Collections.singletonMap(taskId0, secondPartitionAssignment);
+
+ mockSingleActiveTask();
+ expect(activeTaskCreator.retryWithBackoff(EasyMock.anyObject(Consumer.class),
+ EasyMock.eq(secondAssignment),
+ EasyMock.eq(time.milliseconds())))
+ .andReturn(Collections.singletonMap(firstTask, secondPartitionAssignment));
+
+ expect(threadMetadataProvider.activeTasks())
+ .andReturn(secondAssignment);
+
+ firstTask.closeSuspended(true, null);
+ expectLastCall();
+
+ replay(threadMetadataProvider, activeTaskCreator, standbyTaskCreator, firstTask);
+
+ taskManager.createTasks(taskId0Partitions);
+ taskManager.suspendTasksAndState();
+
+ taskManager.createTasks(secondPartitionAssignment);
+
+ verify(activeTaskCreator, firstTask);
+ }
+
+
+ @SuppressWarnings("unchecked")
+ @Test
+ public void shouldCloseStandbyTaskIfFailureOnSuspend() {
+ checkOrder(firstTask, true);
+ mockStandbyTaskExpectations(Collections.<TopicPartition, Long>emptyMap());
+ verifyTaskIsClosedOnSuspendFailure(Collections.<TopicPartition>emptySet());
+ }
+
+ @Test
+ public void shouldCloseActiveTaskIfFailureOnSuspend() {
+ checkOrder(firstTask, true);
+ mockSingleActiveTask();
+ verifyTaskIsClosedOnSuspendFailure(taskId0Partitions);
+ }
+
+ @Test
+ public void shouldInitializeRestoreConsumerWithOffsetsFromStandbyTasks() {
+ mockStandbyTaskExpectations(Collections.singletonMap(t1p0, 0L));
+ restoreConsumer.assign(EasyMock.eq(taskId0Partitions));
+ expectLastCall();
+ replay(threadMetadataProvider, firstTask, activeTaskCreator, standbyTaskCreator, restoreConsumer);
+
+ taskManager.createTasks(Collections.<TopicPartition>emptySet());
+
+ EasyMock.verify(restoreConsumer);
+ }
+
+ @Test
+ public void shouldNotCloseSuspendedTasksTwice() {
+ mockSingleActiveTask();
+ expect(threadMetadataProvider.activeTasks())
+ .andReturn(Collections.<TaskId, Set<TopicPartition>>emptyMap());
+ firstTask.suspend();
+ expectLastCall();
+ firstTask.closeSuspended(true, null);
+ expectLastCall();
+
+ replay(threadMetadataProvider, activeTaskCreator, standbyTaskCreator, firstTask);
+
+ taskManager.createTasks(taskId0Partitions);
+ taskManager.suspendTasksAndState();
+
+ taskManager.createTasks(Collections.<TopicPartition>emptySet());
+
+ verify(firstTask);
+ }
+
+ @Test
+ public void shouldNotCloseActiveTaskOnCommitFailedExceptionDuringTaskSuspend() {
+ checkOrder(firstTask, true);
+ mockSingleActiveTask();
+ firstTask.suspend();
+ expectLastCall().andThrow(new CommitFailedException());
+
+ replay(threadMetadataProvider, firstTask, activeTaskCreator, standbyTaskCreator);
+
+ taskManager.createTasks(taskId0Partitions);
+
+ taskManager.suspendTasksAndState();
+ verify(firstTask);
+ }
+
+
+ @SuppressWarnings("unchecked")
+ private void mockStandbyTaskExpectations(final Map<TopicPartition, Long> checkpoint) {
+ expect(threadMetadataProvider.standbyTasks())
+ .andReturn(taskId0Assignment)
+ .anyTimes();
+ expect(threadMetadataProvider.activeTasks())
+ .andStubReturn(Collections.<TaskId, Set<TopicPartition>>emptyMap());
+
+ expect(standbyTaskCreator.retryWithBackoff(EasyMock.anyObject(Consumer.class),
+ EasyMock.eq(taskId0Assignment),
+ EasyMock.eq(time.milliseconds())))
+ .andReturn(Collections.singletonMap(firstTask, taskId0Partitions));
+
+ stubTaskCreator(activeTaskCreator);
+
+ expect(firstTask.checkpointedOffsets())
+ .andReturn(checkpoint)
+ .anyTimes();
+ }
+
+ @SuppressWarnings("unchecked")
+ private void mockSingleActiveTask() {
+ expect(threadMetadataProvider.standbyTasks())
+ .andReturn(Collections.<TaskId, Set<TopicPartition>>emptyMap())
+ .anyTimes();
+ expect(threadMetadataProvider.activeTasks())
+ .andReturn(taskId0Assignment);
+
+ expect(activeTaskCreator.retryWithBackoff(EasyMock.anyObject(Consumer.class),
+ EasyMock.eq(taskId0Assignment),
+ EasyMock.eq(time.milliseconds())))
+ .andReturn(Collections.singletonMap(firstTask, taskId0Partitions));
+
+ stubTaskCreator(standbyTaskCreator);
+
+ expect(firstTask.id()).andStubReturn(taskId0);
+ expect(firstTask.partitions()).andStubReturn(taskId0Partitions);
+ }
+
+ private void verifyTaskIsClosedOnSuspendFailure(final Set<TopicPartition> assignment) {
+ firstTask.suspend();
+ expectLastCall().andThrow(new RuntimeException("KABOOM!"));
+ firstTask.close(false);
+ expectLastCall();
+ replay(threadMetadataProvider, firstTask, activeTaskCreator, standbyTaskCreator);
+
+ taskManager.createTasks(assignment);
+
+ try {
+ taskManager.suspendTasksAndState();
+ fail("should have thrown StreamsException");
+ } catch (final StreamsException e) {
+ // pass
+ }
+ verify(firstTask);
+ }
+
+ @SuppressWarnings("unchecked")
+ private void stubTaskCreator(final StreamThread.AbstractTaskCreator taskCreator) {
+ expect(taskCreator.retryWithBackoff(EasyMock.anyObject(Consumer.class),
+ EasyMock.anyObject(Map.class),
+ EasyMock.anyLong()))
+ .andReturn(Collections.<Task, Set<TopicPartition>>emptyMap())
+ .anyTimes();
+ }
+
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/kafka/blob/3e69ce80/streams/src/test/java/org/apache/kafka/streams/state/internals/StreamThreadStateStoreProviderTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/StreamThreadStateStoreProviderTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/StreamThreadStateStoreProviderTest.java
index 62cb09c..ef24cf4 100644
--- a/streams/src/test/java/org/apache/kafka/streams/state/internals/StreamThreadStateStoreProviderTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/StreamThreadStateStoreProviderTest.java
@@ -33,7 +33,7 @@ import org.apache.kafka.streams.processor.internals.StateDirectory;
import org.apache.kafka.streams.processor.internals.StoreChangelogReader;
import org.apache.kafka.streams.processor.internals.StreamTask;
import org.apache.kafka.streams.processor.internals.StreamThread;
-import org.apache.kafka.streams.processor.internals.StreamsMetadataState;
+import org.apache.kafka.streams.processor.internals.Task;
import org.apache.kafka.streams.state.QueryableStoreTypes;
import org.apache.kafka.streams.state.ReadOnlyKeyValueStore;
import org.apache.kafka.streams.state.ReadOnlyWindowStore;
@@ -42,6 +42,7 @@ import org.apache.kafka.test.MockClientSupplier;
import org.apache.kafka.test.MockProcessorSupplier;
import org.apache.kafka.test.MockStateRestoreListener;
import org.apache.kafka.test.TestUtils;
+import org.easymock.EasyMock;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
@@ -54,7 +55,6 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
-import java.util.UUID;
import static org.apache.kafka.streams.state.QueryableStoreTypes.windowStore;
import static org.junit.Assert.assertEquals;
@@ -66,9 +66,11 @@ public class StreamThreadStateStoreProviderTest {
private StreamThreadStateStoreProvider provider;
private StateDirectory stateDirectory;
private File stateDir;
- private boolean storesAvailable;
private final String topicName = "topic";
+ private StreamThread threadMock;
+ private Map<TaskId, Task> tasks;
+ @SuppressWarnings("deprecation")
@Before
public void before() throws IOException {
final TopologyBuilder builder = new TopologyBuilder();
@@ -100,7 +102,7 @@ public class StreamThreadStateStoreProviderTest {
builder.setApplicationId(applicationId);
final ProcessorTopology topology = builder.build(null);
- final Map<TaskId, StreamTask> tasks = new HashMap<>();
+ tasks = new HashMap<>();
stateDirectory = new StateDirectory(applicationId, stateConfigDir, new MockTime());
taskOne = createStreamsTask(applicationId, streamsConfig, clientSupplier, topology,
new TaskId(0, 0));
@@ -111,30 +113,9 @@ public class StreamThreadStateStoreProviderTest {
tasks.put(new TaskId(0, 1),
taskTwo);
- storesAvailable = true;
- provider = new StreamThreadStateStoreProvider(
- new StreamThread(
- builder.internalTopologyBuilder,
- streamsConfig,
- clientSupplier,
- applicationId,
- "clientId",
- UUID.randomUUID(),
- new Metrics(),
- Time.SYSTEM,
- new StreamsMetadataState(builder.internalTopologyBuilder, StreamsMetadataState.UNKNOWN_HOST),
- 0, stateDirectory) {
-
- @Override
- public Map<TaskId, StreamTask> tasks() {
- return tasks;
- }
-
- @Override
- public boolean isInitialized() {
- return storesAvailable;
- }
- });
+ threadMock = EasyMock.createNiceMock(StreamThread.class);
+ provider = new StreamThreadStateStoreProvider(threadMock);
+
}
@After
@@ -144,6 +125,7 @@ public class StreamThreadStateStoreProviderTest {
@Test
public void shouldFindKeyValueStores() throws Exception {
+ mockThread(true);
final List<ReadOnlyKeyValueStore<String, String>> kvStores =
provider.stores("kv-store", QueryableStoreTypes.<String, String>keyValueStore());
assertEquals(2, kvStores.size());
@@ -151,6 +133,7 @@ public class StreamThreadStateStoreProviderTest {
@Test
public void shouldFindWindowStores() throws Exception {
+ mockThread(true);
final List<ReadOnlyWindowStore<Object, Object>>
windowStores =
provider.stores("window-store", windowStore());
@@ -159,18 +142,21 @@ public class StreamThreadStateStoreProviderTest {
@Test(expected = InvalidStateStoreException.class)
public void shouldThrowInvalidStoreExceptionIfWindowStoreClosed() throws Exception {
+ mockThread(true);
taskOne.getStore("window-store").close();
provider.stores("window-store", QueryableStoreTypes.windowStore());
}
@Test(expected = InvalidStateStoreException.class)
public void shouldThrowInvalidStoreExceptionIfKVStoreClosed() throws Exception {
+ mockThread(true);
taskOne.getStore("kv-store").close();
provider.stores("kv-store", QueryableStoreTypes.keyValueStore());
}
@Test
public void shouldReturnEmptyListIfNoStoresFoundWithName() throws Exception {
+ mockThread(true);
assertEquals(Collections.emptyList(), provider.stores("not-a-store", QueryableStoreTypes
.keyValueStore()));
}
@@ -178,13 +164,14 @@ public class StreamThreadStateStoreProviderTest {
@Test
public void shouldReturnEmptyListIfStoreExistsButIsNotOfTypeValueStore() throws Exception {
+ mockThread(true);
assertEquals(Collections.emptyList(), provider.stores("window-store",
QueryableStoreTypes.keyValueStore()));
}
@Test(expected = InvalidStateStoreException.class)
public void shouldThrowInvalidStoreExceptionIfNotAllStoresAvailable() throws Exception {
- storesAvailable = false;
+ mockThread(false);
provider.stores("kv-store", QueryableStoreTypes.keyValueStore());
}
@@ -212,6 +199,12 @@ public class StreamThreadStateStoreProviderTest {
};
}
+ private void mockThread(final boolean initialized) {
+ EasyMock.expect(threadMock.isInitialized()).andReturn(initialized);
+ EasyMock.expect(threadMock.tasks()).andStubReturn(tasks);
+ EasyMock.replay(threadMock);
+ }
+
private void configureRestoreConsumer(final MockClientSupplier clientSupplier,
final String topic) {
clientSupplier.restoreConsumer
http://git-wip-us.apache.org/repos/asf/kafka/blob/3e69ce80/streams/src/test/java/org/apache/kafka/test/MockChangelogReader.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/test/MockChangelogReader.java b/streams/src/test/java/org/apache/kafka/test/MockChangelogReader.java
index 4db10e7..1389bbe 100644
--- a/streams/src/test/java/org/apache/kafka/test/MockChangelogReader.java
+++ b/streams/src/test/java/org/apache/kafka/test/MockChangelogReader.java
@@ -48,6 +48,11 @@ public class MockChangelogReader implements ChangelogReader {
return Collections.emptyMap();
}
+ @Override
+ public void clear() {
+ registered.clear();
+ }
+
public boolean wasRegistered(final TopicPartition partition) {
return registered.contains(partition);
}
[5/5] kafka git commit: KAFKA-5702; extract refactor StreamThread
Posted by da...@apache.org.
KAFKA-5702; extract refactor StreamThread
Extracted `TaskManager` to handle all task related activities.
Make `StandbyTaskCreator`, `TaskCreator`, and `RebalanceListener` static classes so they must define their dependencies and can be testing independently of `StreamThread`
Added interfaces between `StreamPartitionAssignor` & `StreamThread` to reduce coupling.
Author: Damian Guy <da...@gmail.com>
Reviewers: Bill Bejeck <bi...@confluent.io>, Guozhang Wang <wa...@gmail.com>, Eno Thereska <en...@gmail.com>
Closes #3624 from dguy/stream-thread-refactor
Project: http://git-wip-us.apache.org/repos/asf/kafka/repo
Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/3e69ce80
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/3e69ce80
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/3e69ce80
Branch: refs/heads/trunk
Commit: 3e69ce80157eb6a5e6dd05e3be522b5208a41bc5
Parents: b2b5295
Author: Damian Guy <da...@gmail.com>
Authored: Fri Aug 11 12:14:01 2017 +0100
Committer: Damian Guy <da...@gmail.com>
Committed: Fri Aug 11 12:14:01 2017 +0100
----------------------------------------------------------------------
.../org/apache/kafka/streams/KafkaStreams.java | 51 +-
.../processor/internals/AbstractTask.java | 16 +-
.../processor/internals/ChangelogReader.java | 5 +
.../processor/internals/StandbyTask.java | 32 +-
.../internals/StoreChangelogReader.java | 6 +
.../internals/StreamPartitionAssignor.java | 50 +-
.../streams/processor/internals/StreamTask.java | 20 +-
.../processor/internals/StreamThread.java | 1038 ++++++---------
.../internals/StreamsMetadataState.java | 2 +-
.../kafka/streams/processor/internals/Task.java | 67 +
.../processor/internals/TaskManager.java | 524 ++++++++
.../processor/internals/ThreadDataProvider.java | 36 +
.../internals/ThreadMetadataProvider.java | 36 +
.../apache/kafka/streams/state/HostInfo.java | 3 +-
.../StreamThreadStateStoreProvider.java | 4 +-
.../integration/RegexSourceIntegrationTest.java | 280 ++--
.../processor/internals/AbstractTaskTest.java | 43 +
.../processor/internals/StandbyTaskTest.java | 46 +
.../internals/StreamPartitionAssignorTest.java | 397 ++----
.../processor/internals/StreamTaskTest.java | 92 +-
.../processor/internals/StreamThreadTest.java | 1222 +++---------------
.../processor/internals/TaskManagerTest.java | 240 ++++
.../StreamThreadStateStoreProviderTest.java | 51 +-
.../apache/kafka/test/MockChangelogReader.java | 5 +
24 files changed, 2038 insertions(+), 2228 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kafka/blob/3e69ce80/streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java b/streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
index ec09730..46f6cd8 100644
--- a/streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
+++ b/streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
@@ -21,6 +21,7 @@ import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.Metric;
import org.apache.kafka.common.MetricName;
+import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.annotation.InterfaceStability;
import org.apache.kafka.common.config.ConfigException;
import org.apache.kafka.common.metrics.JmxReporter;
@@ -145,6 +146,29 @@ public class KafkaStreams {
private final StreamsMetadataState streamsMetadataState;
private final StreamsConfig config;
private final StateDirectory stateDirectory;
+ private StateRestoreListener globalStateRestoreListener;
+ private final StateRestoreListener delegatingStateRestoreListener = new StateRestoreListener() {
+ @Override
+ public void onRestoreStart(final TopicPartition topicPartition, final String storeName, final long startingOffset, final long endingOffset) {
+ if (globalStateRestoreListener != null) {
+ globalStateRestoreListener.onRestoreStart(topicPartition, storeName, startingOffset, endingOffset);
+ }
+ }
+
+ @Override
+ public void onBatchRestored(final TopicPartition topicPartition, final String storeName, final long batchEndOffset, final long numRestored) {
+ if (globalStateRestoreListener != null) {
+ globalStateRestoreListener.onBatchRestored(topicPartition, storeName, batchEndOffset, numRestored);
+ }
+ }
+
+ @Override
+ public void onRestoreEnd(final TopicPartition topicPartition, final String storeName, final long totalRestored) {
+ if (globalStateRestoreListener != null) {
+ globalStateRestoreListener.onRestoreEnd(topicPartition, storeName, totalRestored);
+ }
+ }
+ };
// container states
/**
@@ -525,18 +549,19 @@ public class KafkaStreams {
globalThreadState = globalStreamThread.state();
}
+
for (int i = 0; i < threads.length; i++) {
- threads[i] = new StreamThread(internalTopologyBuilder,
- config,
- clientSupplier,
- applicationId,
- clientId,
- processId,
- metrics,
- time,
- streamsMetadataState,
- cacheSizeBytes,
- stateDirectory);
+ threads[i] = StreamThread.create(internalTopologyBuilder,
+ config,
+ clientSupplier,
+ processId,
+ clientId,
+ metrics,
+ time,
+ streamsMetadataState,
+ cacheSizeBytes,
+ stateDirectory,
+ delegatingStateRestoreListener);
threadState.put(threads[i].getId(), threads[i].state());
storeProviders.add(new StreamThreadStateStoreProvider(threads[i]));
}
@@ -823,9 +848,7 @@ public class KafkaStreams {
public void setGlobalStateRestoreListener(final StateRestoreListener globalStateRestoreListener) {
synchronized (stateLock) {
if (state == State.CREATED) {
- for (StreamThread thread : threads) {
- thread.setGlobalStateRestoreListener(globalStateRestoreListener);
- }
+ this.globalStateRestoreListener = globalStateRestoreListener;
} else {
throw new IllegalStateException("Can only set the GlobalRestoreListener in the CREATED state");
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/3e69ce80/streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractTask.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractTask.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractTask.java
index 8427e11..2688a8f 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractTask.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractTask.java
@@ -37,7 +37,7 @@ import java.util.HashSet;
import java.util.Map;
import java.util.Set;
-public abstract class AbstractTask {
+public abstract class AbstractTask implements Task {
private static final Logger log = LoggerFactory.getLogger(AbstractTask.class);
final TaskId id;
@@ -87,32 +87,32 @@ public abstract class AbstractTask {
}
}
- public abstract void resume();
-
- public abstract void commit();
- public abstract void suspend();
- public abstract void close(final boolean clean);
-
- public final TaskId id() {
+ @Override
+ public TaskId id() {
return id;
}
+ @Override
public final String applicationId() {
return applicationId;
}
+ @Override
public final Set<TopicPartition> partitions() {
return partitions;
}
+ @Override
public final ProcessorTopology topology() {
return topology;
}
+ @Override
public final ProcessorContext context() {
return processorContext;
}
+ @Override
public StateStore getStore(final String name) {
return stateMgr.getStore(name);
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/3e69ce80/streams/src/main/java/org/apache/kafka/streams/processor/internals/ChangelogReader.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/ChangelogReader.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/ChangelogReader.java
index 2e006a0..f06f760 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/ChangelogReader.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/ChangelogReader.java
@@ -49,4 +49,9 @@ public interface ChangelogReader {
* @return the restored offsets for all persistent stores.
*/
Map<TopicPartition, Long> restoredOffsets();
+
+ /**
+ * Clear out any internal state so this can be re-used
+ */
+ void clear();
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/3e69ce80/streams/src/main/java/org/apache/kafka/streams/processor/internals/StandbyTask.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StandbyTask.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StandbyTask.java
index 754700f..98a907b 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StandbyTask.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StandbyTask.java
@@ -133,6 +133,26 @@ public class StandbyTask extends AbstractTask {
}
}
+ @Override
+ public void closeSuspended(final boolean clean, final RuntimeException e) {
+ throw new UnsupportedOperationException("closeSuspended not supported by StandbyTask");
+ }
+
+ @Override
+ public boolean maybePunctuateStreamTime() {
+ throw new UnsupportedOperationException("maybePunctuateStreamTime not supported by StandbyTask");
+ }
+
+ @Override
+ public boolean maybePunctuateSystemTime() {
+ throw new UnsupportedOperationException("maybePunctuateSystemTime not supported by StandbyTask");
+ }
+
+ @Override
+ public boolean commitNeeded() {
+ return false;
+ }
+
/**
* Updates a state store using records from one change log partition
*
@@ -144,8 +164,18 @@ public class StandbyTask extends AbstractTask {
return stateMgr.updateStandbyStates(partition, records);
}
- Map<TopicPartition, Long> checkpointedOffsets() {
+ @Override
+ public int addRecords(final TopicPartition partition, final Iterable<ConsumerRecord<byte[], byte[]>> records) {
+ throw new UnsupportedOperationException("addRecords not supported by StandbyTask");
+ }
+
+ public Map<TopicPartition, Long> checkpointedOffsets() {
return checkpointedOffsets;
}
+ @Override
+ public boolean process() {
+ throw new UnsupportedOperationException("process not supported by StandbyTask");
+ }
+
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/kafka/blob/3e69ce80/streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java
index 842721d..1887e73 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java
@@ -186,6 +186,12 @@ public class StoreChangelogReader implements ChangelogReader {
return restoredOffsets;
}
+ @Override
+ public void clear() {
+ partitionInfo.clear();
+ stateRestorers.clear();
+ }
+
private void restorePartition(final Map<TopicPartition, Long> endOffsets,
final ConsumerRecords<byte[], byte[]> allRecords,
final Iterator<TopicPartition> partitionIterator) {
http://git-wip-us.apache.org/repos/asf/kafka/blob/3e69ce80/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamPartitionAssignor.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamPartitionAssignor.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamPartitionAssignor.java
index ebdd64d..d479a72 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamPartitionAssignor.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamPartitionAssignor.java
@@ -53,7 +53,7 @@ import static org.apache.kafka.common.utils.Utils.getHost;
import static org.apache.kafka.common.utils.Utils.getPort;
import static org.apache.kafka.streams.processor.internals.InternalTopicManager.WINDOW_CHANGE_LOG_ADDITIONAL_RETENTION_DEFAULT;
-public class StreamPartitionAssignor implements PartitionAssignor, Configurable {
+public class StreamPartitionAssignor implements PartitionAssignor, Configurable, ThreadMetadataProvider {
private static final Logger log = LoggerFactory.getLogger(StreamPartitionAssignor.class);
private Time time = Time.SYSTEM;
@@ -168,7 +168,7 @@ public class StreamPartitionAssignor implements PartitionAssignor, Configurable
}
};
- private StreamThread streamThread;
+ private ThreadDataProvider threadDataProvider;
private String userEndPoint;
private int numStandbyReplicas;
@@ -207,16 +207,16 @@ public class StreamPartitionAssignor implements PartitionAssignor, Configurable
throw ex;
}
- if (!(o instanceof StreamThread)) {
- KafkaException ex = new KafkaException(String.format("%s is not an instance of %s", o.getClass().getName(), StreamThread.class.getName()));
+ if (!(o instanceof ThreadDataProvider)) {
+ KafkaException ex = new KafkaException(String.format("%s is not an instance of %s", o.getClass().getName(), ThreadDataProvider.class.getName()));
log.error(ex.getMessage(), ex);
throw ex;
}
- streamThread = (StreamThread) o;
- streamThread.setPartitionAssignor(this);
+ threadDataProvider = (ThreadDataProvider) o;
+ threadDataProvider.setThreadMetadataProvider(this);
- logPrefix = String.format("stream-thread [%s]", streamThread.getName());
+ logPrefix = String.format("stream-thread [%s]", threadDataProvider.name());
String userEndPoint = (String) configs.get(StreamsConfig.APPLICATION_SERVER_CONFIG);
if (userEndPoint != null && !userEndPoint.isEmpty()) {
@@ -237,13 +237,13 @@ public class StreamPartitionAssignor implements PartitionAssignor, Configurable
}
internalTopicManager = new InternalTopicManager(
- StreamsKafkaClient.create(this.streamThread.config),
+ StreamsKafkaClient.create(this.threadDataProvider.config()),
configs.containsKey(StreamsConfig.REPLICATION_FACTOR_CONFIG) ? (Integer) configs.get(StreamsConfig.REPLICATION_FACTOR_CONFIG) : 1,
configs.containsKey(StreamsConfig.WINDOW_STORE_CHANGE_LOG_ADDITIONAL_RETENTION_MS_CONFIG) ?
(Long) configs.get(StreamsConfig.WINDOW_STORE_CHANGE_LOG_ADDITIONAL_RETENTION_MS_CONFIG)
: WINDOW_CHANGE_LOG_ADDITIONAL_RETENTION_DEFAULT, time);
- this.copartitionedTopicsValidator = new CopartitionedTopicsValidator(streamThread.getName());
+ this.copartitionedTopicsValidator = new CopartitionedTopicsValidator(threadDataProvider.name());
}
@Override
@@ -258,13 +258,13 @@ public class StreamPartitionAssignor implements PartitionAssignor, Configurable
// 2. Task ids of previously running tasks
// 3. Task ids of valid local states on the client's state directory.
- final Set<TaskId> previousActiveTasks = streamThread.prevActiveTasks();
- Set<TaskId> standbyTasks = streamThread.cachedTasks();
+ final Set<TaskId> previousActiveTasks = threadDataProvider.prevActiveTasks();
+ Set<TaskId> standbyTasks = threadDataProvider.cachedTasks();
standbyTasks.removeAll(previousActiveTasks);
- SubscriptionInfo data = new SubscriptionInfo(streamThread.processId, previousActiveTasks, standbyTasks, this.userEndPoint);
+ SubscriptionInfo data = new SubscriptionInfo(threadDataProvider.processId(), previousActiveTasks, standbyTasks, this.userEndPoint);
- if (streamThread.builder.sourceTopicPattern() != null &&
- !streamThread.builder.subscriptionUpdates().getUpdates().equals(topics)) {
+ if (threadDataProvider.builder().sourceTopicPattern() != null &&
+ !threadDataProvider.builder().subscriptionUpdates().getUpdates().equals(topics)) {
updateSubscribedTopics(topics);
}
@@ -276,7 +276,7 @@ public class StreamPartitionAssignor implements PartitionAssignor, Configurable
log.debug("{} found {} topics possibly matching regex", logPrefix, topics);
// update the topic groups with the returned subscription set for regex pattern subscriptions
subscriptionUpdates.updateTopics(topics);
- streamThread.builder.updateSubscriptions(subscriptionUpdates, streamThread.getName());
+ threadDataProvider.builder().updateSubscriptions(subscriptionUpdates, threadDataProvider.name());
}
/*
@@ -329,7 +329,7 @@ public class StreamPartitionAssignor implements PartitionAssignor, Configurable
// parse the topology to determine the repartition source topics,
// making sure they are created with the number of partitions as
// the maximum of the depending sub-topologies source topics' number of partitions
- Map<Integer, InternalTopologyBuilder.TopicsInfo> topicGroups = streamThread.builder.topicGroups();
+ Map<Integer, InternalTopologyBuilder.TopicsInfo> topicGroups = threadDataProvider.builder().topicGroups();
Map<String, InternalTopicMetadata> repartitionTopicMetadata = new HashMap<>();
for (InternalTopologyBuilder.TopicsInfo topicsInfo : topicGroups.values()) {
@@ -401,7 +401,7 @@ public class StreamPartitionAssignor implements PartitionAssignor, Configurable
// ensure the co-partitioning topics within the group have the same number of partitions,
// and enforce the number of partitions for those repartition topics to be the same if they
// are co-partitioned as well.
- ensureCopartitioning(streamThread.builder.copartitionGroups(), repartitionTopicMetadata, metadata);
+ ensureCopartitioning(threadDataProvider.builder().copartitionGroups(), repartitionTopicMetadata, metadata);
// make sure the repartition source topics exist with the right number of partitions,
// create these topics if necessary
@@ -421,7 +421,7 @@ public class StreamPartitionAssignor implements PartitionAssignor, Configurable
sourceTopicsByGroup.put(entry.getKey(), entry.getValue().sourceTopics);
}
- Map<TaskId, Set<TopicPartition>> partitionsForTask = streamThread.partitionGrouper.partitionGroups(
+ Map<TaskId, Set<TopicPartition>> partitionsForTask = threadDataProvider.partitionGrouper().partitionGroups(
sourceTopicsByGroup, metadataWithInternalTopics);
// check if all partitions are assigned, and there are no duplicates of partitions in multiple tasks
@@ -627,13 +627,13 @@ public class StreamPartitionAssignor implements PartitionAssignor, Configurable
}
private void checkForNewTopicAssignments(Assignment assignment) {
- if (streamThread.builder.sourceTopicPattern() != null) {
+ if (threadDataProvider.builder().sourceTopicPattern() != null) {
final Set<String> assignedTopics = new HashSet<>();
for (final TopicPartition topicPartition : assignment.partitions()) {
assignedTopics.add(topicPartition.topic());
}
- if (!streamThread.builder.subscriptionUpdates().getUpdates().containsAll(assignedTopics)) {
- assignedTopics.addAll(streamThread.builder.subscriptionUpdates().getUpdates());
+ if (!threadDataProvider.builder().subscriptionUpdates().getUpdates().containsAll(assignedTopics)) {
+ assignedTopics.addAll(threadDataProvider.builder().subscriptionUpdates().getUpdates());
updateSubscribedTopics(assignedTopics);
}
}
@@ -702,28 +702,28 @@ public class StreamPartitionAssignor implements PartitionAssignor, Configurable
}
}
- Map<HostInfo, Set<TopicPartition>> getPartitionsByHostState() {
+ public Map<HostInfo, Set<TopicPartition>> getPartitionsByHostState() {
if (partitionsByHostState == null) {
return Collections.emptyMap();
}
return Collections.unmodifiableMap(partitionsByHostState);
}
- Cluster clusterMetadata() {
+ public Cluster clusterMetadata() {
if (metadataWithInternalTopics == null) {
return Cluster.empty();
}
return metadataWithInternalTopics;
}
- Map<TaskId, Set<TopicPartition>> activeTasks() {
+ public Map<TaskId, Set<TopicPartition>> activeTasks() {
if (activeTasks == null) {
return Collections.emptyMap();
}
return Collections.unmodifiableMap(activeTasks);
}
- Map<TaskId, Set<TopicPartition>> standbyTasks() {
+ public Map<TaskId, Set<TopicPartition>> standbyTasks() {
if (standbyTasks == null) {
return Collections.emptyMap();
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/3e69ce80/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java
index d4bd668..697bda8 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java
@@ -42,6 +42,7 @@ import org.slf4j.LoggerFactory;
import java.util.Collection;
import java.util.HashMap;
+import java.util.List;
import java.util.Map;
import static java.lang.String.format;
@@ -400,7 +401,7 @@ public class StreamTask extends AbstractTask implements ProcessorNodePunctuator
}
// helper to avoid calling suspend() twice if a suspended task is not reassigned and closed
- void closeSuspended(boolean clean, RuntimeException firstException) {
+ public void closeSuspended(boolean clean, RuntimeException firstException) {
try {
closeStateManager(clean);
} catch (final RuntimeException e) {
@@ -437,6 +438,11 @@ public class StreamTask extends AbstractTask implements ProcessorNodePunctuator
}
}
+ @Override
+ public Map<TopicPartition, Long> checkpointedOffsets() {
+ throw new UnsupportedOperationException("checkpointedOffsets is not supported by StreamTasks");
+ }
+
/**
* <pre>
* - {@link #suspend(boolean) suspend(clean)}
@@ -529,7 +535,7 @@ public class StreamTask extends AbstractTask implements ProcessorNodePunctuator
* current partition group timestamp has reached the defined stamp
* Note, this is only called in the presence of new records
*/
- boolean maybePunctuateStreamTime() {
+ public boolean maybePunctuateStreamTime() {
final long timestamp = partitionGroup.timestamp();
// if the timestamp is not known yet, meaning there is not enough data accumulated
@@ -546,11 +552,17 @@ public class StreamTask extends AbstractTask implements ProcessorNodePunctuator
* current system timestamp has reached the defined stamp
* Note, this is called irrespective of the presence of new records
*/
- boolean maybePunctuateSystemTime() {
+ public boolean maybePunctuateSystemTime() {
final long timestamp = time.milliseconds();
return systemTimePunctuationQueue.mayPunctuate(timestamp, PunctuationType.SYSTEM_TIME, this);
}
+
+ @Override
+ public List<ConsumerRecord<byte[], byte[]>> update(final TopicPartition partition, final List<ConsumerRecord<byte[], byte[]>> remaining) {
+ throw new UnsupportedOperationException("update is not implemented");
+ }
+
/**
* Request committing the current task's state
*/
@@ -561,7 +573,7 @@ public class StreamTask extends AbstractTask implements ProcessorNodePunctuator
/**
* Whether or not a request has been made to commit the current state
*/
- boolean commitNeeded() {
+ public boolean commitNeeded() {
return commitRequested;
}
[2/5] kafka git commit: KAFKA-5702; extract refactor StreamThread
Posted by da...@apache.org.
http://git-wip-us.apache.org/repos/asf/kafka/blob/3e69ce80/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamPartitionAssignorTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamPartitionAssignorTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamPartitionAssignorTest.java
index 5fc3807..e638a48 100644
--- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamPartitionAssignorTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamPartitionAssignorTest.java
@@ -23,9 +23,6 @@ import org.apache.kafka.common.Node;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.config.ConfigException;
-import org.apache.kafka.common.metrics.Metrics;
-import org.apache.kafka.common.utils.MockTime;
-import org.apache.kafka.common.utils.Time;
import org.apache.kafka.common.utils.Utils;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
@@ -35,6 +32,8 @@ import org.apache.kafka.streams.kstream.JoinWindows;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.KeyValueMapper;
import org.apache.kafka.streams.kstream.ValueJoiner;
+import org.apache.kafka.streams.processor.DefaultPartitionGrouper;
+import org.apache.kafka.streams.processor.PartitionGrouper;
import org.apache.kafka.streams.processor.TaskId;
import org.apache.kafka.streams.processor.internals.assignment.AssignmentInfo;
import org.apache.kafka.streams.processor.internals.assignment.SubscriptionInfo;
@@ -44,9 +43,8 @@ import org.apache.kafka.test.MockInternalTopicManager;
import org.apache.kafka.test.MockProcessorSupplier;
import org.apache.kafka.test.MockStateStoreSupplier;
import org.apache.kafka.test.MockTimestampExtractor;
-import org.apache.kafka.test.TestUtils;
+import org.easymock.EasyMock;
import org.junit.Assert;
-import org.junit.Before;
import org.junit.Test;
import java.util.ArrayList;
@@ -105,18 +103,15 @@ public class StreamPartitionAssignorTest {
private final TaskId task1 = new TaskId(0, 1);
private final TaskId task2 = new TaskId(0, 2);
private final TaskId task3 = new TaskId(0, 3);
- private final String userEndPoint = "localhost:2171";
private final StreamPartitionAssignor partitionAssignor = new StreamPartitionAssignor();
private final MockClientSupplier mockClientSupplier = new MockClientSupplier();
private final InternalTopologyBuilder builder = new InternalTopologyBuilder();
private final StreamsConfig config = new StreamsConfig(configProps());
- private final StateDirectory stateDirectory = new StateDirectory("appId", TestUtils.tempDirectory().getPath(), new MockTime());
- private final StreamThread mockStreamThread = new StreamThread(builder, config,
- mockClientSupplier, "appID",
- "clientId", UUID.randomUUID(),
- new Metrics(), new MockTime(),
- null, 1L, stateDirectory);
+ private final ThreadDataProvider threadDataProvider = EasyMock.createNiceMock(ThreadDataProvider.class);
private final Map<String, Object> configurationMap = new HashMap<>();
+ private final DefaultPartitionGrouper defaultPartitionGrouper = new DefaultPartitionGrouper();
+ private final SingleGroupPartitionGrouperStub stubPartitionGrouper = new SingleGroupPartitionGrouperStub();
+ private final String userEndPoint = "localhost:8080";
private Properties configProps() {
return new Properties() {
@@ -129,13 +124,29 @@ public class StreamPartitionAssignorTest {
};
}
- @Before
- public void setUp() {
- configurationMap.put(StreamsConfig.InternalConfig.STREAM_THREAD_INSTANCE, mockStreamThread);
- configurationMap.put(StreamsConfig.NUM_STANDBY_REPLICAS_CONFIG, 0);
+ private void configurePartitionAssignor(final int standbyReplicas, final String endPoint) {
+ configurationMap.put(StreamsConfig.InternalConfig.STREAM_THREAD_INSTANCE, threadDataProvider);
+ configurationMap.put(StreamsConfig.NUM_STANDBY_REPLICAS_CONFIG, standbyReplicas);
+ configurationMap.put(StreamsConfig.APPLICATION_SERVER_CONFIG, endPoint);
partitionAssignor.configure(configurationMap);
}
+ private void mockThreadDataProvider(final Set<TaskId> prevTasks,
+ final Set<TaskId> cachedTasks,
+ final UUID processId,
+ final PartitionGrouper partitionGrouper,
+ final InternalTopologyBuilder builder) throws NoSuchFieldException, IllegalAccessException {
+ EasyMock.expect(threadDataProvider.name()).andReturn("name").anyTimes();
+ EasyMock.expect(threadDataProvider.prevActiveTasks()).andReturn(prevTasks).anyTimes();
+ EasyMock.expect(threadDataProvider.cachedTasks()).andReturn(cachedTasks).anyTimes();
+ EasyMock.expect(threadDataProvider.config()).andReturn(config).anyTimes();
+ EasyMock.expect(threadDataProvider.builder()).andReturn(builder).anyTimes();
+ EasyMock.expect(threadDataProvider.processId()).andReturn(processId).anyTimes();
+ EasyMock.expect(threadDataProvider.partitionGrouper()).andReturn(partitionGrouper).anyTimes();
+ EasyMock.replay(threadDataProvider);
+ }
+
+
@Test
public void testSubscription() throws Exception {
builder.addSource(null, "source1", null, null, null, "topic1");
@@ -148,34 +159,10 @@ public class StreamPartitionAssignorTest {
new TaskId(0, 1), new TaskId(1, 1), new TaskId(2, 1),
new TaskId(0, 2), new TaskId(1, 2), new TaskId(2, 2));
- String clientId = "client-id";
- UUID processId = UUID.randomUUID();
- StreamThread thread = new StreamThread(
- builder,
- config,
- new MockClientSupplier(),
- "test",
- clientId,
- processId,
- new Metrics(),
- Time.SYSTEM,
- new StreamsMetadataState(builder,
- StreamsMetadataState.UNKNOWN_HOST),
- 0,
- stateDirectory) {
-
- @Override
- public Set<TaskId> prevActiveTasks() {
- return prevTasks;
- }
- @Override
- public Set<TaskId> cachedTasks() {
- return cachedTasks;
- }
- };
-
- partitionAssignor.configure(config.getConsumerConfigs(thread, "test", clientId));
+ final UUID processId = UUID.randomUUID();
+ mockThreadDataProvider(prevTasks, cachedTasks, processId, stubPartitionGrouper, builder);
+ configurePartitionAssignor(0, null);
PartitionAssignor.Subscription subscription = partitionAssignor.subscription(Utils.mkSet("topic1", "topic2"));
Collections.sort(subscription.topics());
@@ -188,6 +175,7 @@ public class StreamPartitionAssignorTest {
assertEquals(info.encode(), subscription.userData());
}
+
@Test
public void testAssignBasic() throws Exception {
builder.addSource(null, "source1", null, null, null, "topic1");
@@ -205,24 +193,11 @@ public class StreamPartitionAssignorTest {
UUID uuid1 = UUID.randomUUID();
UUID uuid2 = UUID.randomUUID();
- String client1 = "client1";
+ mockThreadDataProvider(prevTasks10, standbyTasks10, uuid1, stubPartitionGrouper, builder);
+ configurePartitionAssignor(0, null);
- StreamThread thread10 = new StreamThread(
- builder,
- config,
- mockClientSupplier,
- "test",
- client1,
- uuid1,
- new Metrics(),
- Time.SYSTEM,
- new StreamsMetadataState(builder, StreamsMetadataState.UNKNOWN_HOST),
- 0,
- stateDirectory);
-
- partitionAssignor.configure(config.getConsumerConfigs(thread10, "test", client1));
- partitionAssignor.setInternalTopicManager(new MockInternalTopicManager(thread10.config, mockClientSupplier.restoreConsumer));
+ partitionAssignor.setInternalTopicManager(new MockInternalTopicManager(config, mockClientSupplier.restoreConsumer));
Map<String, PartitionAssignor.Subscription> subscriptions = new HashMap<>();
subscriptions.put("consumer10",
@@ -281,23 +256,11 @@ public class StreamPartitionAssignorTest {
Set<TaskId> allTasks = Utils.mkSet(task0, task1, task2);
UUID uuid1 = UUID.randomUUID();
- String client1 = "client1";
- StreamThread thread10 = new StreamThread(
- builder,
- config,
- mockClientSupplier,
- "test",
- client1,
- uuid1,
- new Metrics(),
- Time.SYSTEM,
- new StreamsMetadataState(builder, StreamsMetadataState.UNKNOWN_HOST),
- 0,
- stateDirectory);
-
- partitionAssignor.configure(config.getConsumerConfigs(thread10, "test", client1));
- partitionAssignor.setInternalTopicManager(new MockInternalTopicManager(thread10.config, mockClientSupplier.restoreConsumer));
+ mockThreadDataProvider(Collections.<TaskId>emptySet(), Collections.<TaskId>emptySet(), uuid1, stubPartitionGrouper, builder);
+ configurePartitionAssignor(0, null);
+
+ partitionAssignor.setInternalTopicManager(new MockInternalTopicManager(config, mockClientSupplier.restoreConsumer));
Map<String, PartitionAssignor.Subscription> subscriptions = new HashMap<>();
subscriptions.put("consumer10",
new PartitionAssignor.Subscription(topics, new SubscriptionInfo(uuid1, Collections.<TaskId>emptySet(), Collections.<TaskId>emptySet(), userEndPoint).encode()));
@@ -331,22 +294,9 @@ public class StreamPartitionAssignorTest {
Collections.<String>emptySet(),
Collections.<String>emptySet());
UUID uuid1 = UUID.randomUUID();
- String client1 = "client1";
- StreamThread thread10 = new StreamThread(
- builder,
- config,
- new MockClientSupplier(),
- "test",
- client1,
- uuid1,
- new Metrics(),
- Time.SYSTEM,
- new StreamsMetadataState(builder, StreamsMetadataState.UNKNOWN_HOST),
- 0,
- stateDirectory);
-
- partitionAssignor.configure(config.getConsumerConfigs(thread10, "test", client1));
+ mockThreadDataProvider(prevTasks10, standbyTasks10, uuid1, stubPartitionGrouper, builder);
+ configurePartitionAssignor(0, null);
Map<String, PartitionAssignor.Subscription> subscriptions = new HashMap<>();
subscriptions.put("consumer10",
@@ -400,23 +350,9 @@ public class StreamPartitionAssignorTest {
UUID uuid1 = UUID.randomUUID();
UUID uuid2 = UUID.randomUUID();
- String client1 = "client1";
-
- StreamThread thread10 = new StreamThread(
- builder,
- config,
- mockClientSupplier,
- "test",
- client1,
- uuid1,
- new Metrics(),
- Time.SYSTEM,
- new StreamsMetadataState(builder, StreamsMetadataState.UNKNOWN_HOST),
- 0,
- stateDirectory);
-
- partitionAssignor.configure(config.getConsumerConfigs(thread10, "test", client1));
- partitionAssignor.setInternalTopicManager(new MockInternalTopicManager(thread10.config, mockClientSupplier.restoreConsumer));
+ mockThreadDataProvider(prevTasks10, Collections.<TaskId>emptySet(), uuid1, stubPartitionGrouper, builder);
+ configurePartitionAssignor(0, null);
+ partitionAssignor.setInternalTopicManager(new MockInternalTopicManager(config, mockClientSupplier.restoreConsumer));
Map<String, PartitionAssignor.Subscription> subscriptions = new HashMap<>();
subscriptions.put("consumer10",
@@ -477,24 +413,14 @@ public class StreamPartitionAssignorTest {
UUID uuid1 = UUID.randomUUID();
UUID uuid2 = UUID.randomUUID();
- String client1 = "client1";
+ mockThreadDataProvider(Collections.<TaskId>emptySet(),
+ Collections.<TaskId>emptySet(),
+ uuid1,
+ defaultPartitionGrouper, builder);
+ configurePartitionAssignor(0, null);
- StreamThread thread10 = new StreamThread(
- builder,
- config,
- mockClientSupplier,
- applicationId,
- client1,
- uuid1,
- new Metrics(),
- Time.SYSTEM,
- new StreamsMetadataState(builder, StreamsMetadataState.UNKNOWN_HOST),
- 0,
- stateDirectory);
-
- partitionAssignor.configure(config.getConsumerConfigs(thread10, applicationId, client1));
- partitionAssignor.setInternalTopicManager(new MockInternalTopicManager(thread10.config, mockClientSupplier.restoreConsumer));
+ partitionAssignor.setInternalTopicManager(new MockInternalTopicManager(config, mockClientSupplier.restoreConsumer));
Map<String, PartitionAssignor.Subscription> subscriptions = new HashMap<>();
subscriptions.put("consumer10",
@@ -526,7 +452,7 @@ public class StreamPartitionAssignorTest {
assertEquals(new HashSet<>(tasks), allTasks);
// check tasks for state topics
- Map<Integer, InternalTopologyBuilder.TopicsInfo> topicGroups = thread10.builder.topicGroups();
+ Map<Integer, InternalTopologyBuilder.TopicsInfo> topicGroups = builder.topicGroups();
assertEquals(Utils.mkSet(task00, task01, task02), tasksForState(applicationId, "store1", tasks, topicGroups));
assertEquals(Utils.mkSet(task10, task11, task12), tasksForState(applicationId, "store2", tasks, topicGroups));
@@ -572,23 +498,11 @@ public class StreamPartitionAssignorTest {
UUID uuid1 = UUID.randomUUID();
UUID uuid2 = UUID.randomUUID();
- String client1 = "client1";
- StreamThread thread10 = new StreamThread(
- builder,
- config,
- mockClientSupplier,
- "test",
- client1,
- uuid1,
- new Metrics(),
- Time.SYSTEM,
- new StreamsMetadataState(builder, StreamsMetadataState.UNKNOWN_HOST),
- 0,
- stateDirectory);
-
- partitionAssignor.configure(config.getConsumerConfigs(thread10, "test", client1));
- partitionAssignor.setInternalTopicManager(new MockInternalTopicManager(thread10.config, mockClientSupplier.restoreConsumer));
+ mockThreadDataProvider(prevTasks00, standbyTasks01, uuid1, defaultPartitionGrouper, builder);
+
+ configurePartitionAssignor(1, null);
+ partitionAssignor.setInternalTopicManager(new MockInternalTopicManager(config, mockClientSupplier.restoreConsumer));
Map<String, PartitionAssignor.Subscription> subscriptions = new HashMap<>();
subscriptions.put("consumer10",
@@ -642,22 +556,8 @@ public class StreamPartitionAssignorTest {
builder.addProcessor("processor", new MockProcessorSupplier(), "source1", "source2");
UUID uuid = UUID.randomUUID();
- String client1 = "client1";
-
- StreamThread thread = new StreamThread(
- builder,
- config,
- mockClientSupplier,
- "test",
- client1,
- uuid,
- new Metrics(),
- Time.SYSTEM,
- new StreamsMetadataState(builder, StreamsMetadataState.UNKNOWN_HOST),
- 0,
- stateDirectory);
-
- partitionAssignor.configure(config.getConsumerConfigs(thread, "test", client1));
+ mockThreadDataProvider(Collections.<TaskId>emptySet(), Collections.<TaskId>emptySet(), uuid, defaultPartitionGrouper, builder);
+ configurePartitionAssignor(0, null);
List<TaskId> activeTaskList = Utils.mkList(task0, task3);
Map<TaskId, Set<TopicPartition>> activeTasks = new HashMap<>();
@@ -689,23 +589,9 @@ public class StreamPartitionAssignorTest {
Set<TaskId> allTasks = Utils.mkSet(task0, task1, task2);
UUID uuid1 = UUID.randomUUID();
- String client1 = "client1";
-
- StreamThread thread10 = new StreamThread(
- builder,
- config,
- mockClientSupplier,
- applicationId,
- client1,
- uuid1,
- new Metrics(),
- Time.SYSTEM,
- new StreamsMetadataState(builder, StreamsMetadataState.UNKNOWN_HOST),
- 0,
- stateDirectory);
-
- partitionAssignor.configure(config.getConsumerConfigs(thread10, applicationId, client1));
- MockInternalTopicManager internalTopicManager = new MockInternalTopicManager(thread10.config, mockClientSupplier.restoreConsumer);
+ mockThreadDataProvider(Collections.<TaskId>emptySet(), Collections.<TaskId>emptySet(), uuid1, defaultPartitionGrouper, builder);
+ configurePartitionAssignor(0, null);
+ MockInternalTopicManager internalTopicManager = new MockInternalTopicManager(config, mockClientSupplier.restoreConsumer);
partitionAssignor.setInternalTopicManager(internalTopicManager);
Map<String, PartitionAssignor.Subscription> subscriptions = new HashMap<>();
@@ -737,13 +623,10 @@ public class StreamPartitionAssignorTest {
Set<TaskId> allTasks = Utils.mkSet(task0, task1, task2);
UUID uuid1 = UUID.randomUUID();
- String client1 = "client1";
+ mockThreadDataProvider(Collections.<TaskId>emptySet(), Collections.<TaskId>emptySet(), uuid1, defaultPartitionGrouper, builder);
- StreamThread thread10 = new StreamThread(builder, config, mockClientSupplier, applicationId, client1, uuid1, new Metrics(), Time.SYSTEM, new StreamsMetadataState(builder, StreamsMetadataState.UNKNOWN_HOST),
- 0, stateDirectory);
-
- partitionAssignor.configure(config.getConsumerConfigs(thread10, applicationId, client1));
- MockInternalTopicManager internalTopicManager = new MockInternalTopicManager(thread10.config, mockClientSupplier.restoreConsumer);
+ configurePartitionAssignor(0, null);
+ MockInternalTopicManager internalTopicManager = new MockInternalTopicManager(config, mockClientSupplier.restoreConsumer);
partitionAssignor.setInternalTopicManager(internalTopicManager);
Map<String, PartitionAssignor.Subscription> subscriptions = new HashMap<>();
@@ -760,9 +643,6 @@ public class StreamPartitionAssignorTest {
@Test
public void shouldAddUserDefinedEndPointToSubscription() throws Exception {
- final Properties properties = configProps();
- properties.put(StreamsConfig.APPLICATION_SERVER_CONFIG, "localhost:8080");
- final StreamsConfig config = new StreamsConfig(properties);
final String applicationId = "application-id";
builder.setApplicationId(applicationId);
builder.addSource(null, "source", null, null, null, "input");
@@ -770,12 +650,11 @@ public class StreamPartitionAssignorTest {
builder.addSink("sink", "output", null, null, null, "processor");
final UUID uuid1 = UUID.randomUUID();
- final String client1 = "client1";
-
- final StreamThread streamThread = new StreamThread(builder, config, mockClientSupplier, applicationId, client1, uuid1, new Metrics(), Time.SYSTEM, new StreamsMetadataState(builder, StreamsMetadataState.UNKNOWN_HOST),
- 0, stateDirectory);
-
- partitionAssignor.configure(config.getConsumerConfigs(streamThread, applicationId, client1));
+ mockThreadDataProvider(Collections.<TaskId>emptySet(),
+ Collections.<TaskId>emptySet(),
+ uuid1,
+ defaultPartitionGrouper, builder);
+ configurePartitionAssignor(0, userEndPoint);
final PartitionAssignor.Subscription subscription = partitionAssignor.subscription(Utils.mkSet("input"));
final SubscriptionInfo subscriptionInfo = SubscriptionInfo.decode(subscription.userData());
assertEquals("localhost:8080", subscriptionInfo.userEndPoint);
@@ -783,10 +662,6 @@ public class StreamPartitionAssignorTest {
@Test
public void shouldMapUserEndPointToTopicPartitions() throws Exception {
- final Properties properties = configProps();
- final String myEndPoint = "localhost:8080";
- properties.put(StreamsConfig.APPLICATION_SERVER_CONFIG, myEndPoint);
- final StreamsConfig config = new StreamsConfig(properties);
final String applicationId = "application-id";
builder.setApplicationId(applicationId);
builder.addSource(null, "source", null, null, null, "topic1");
@@ -796,29 +671,15 @@ public class StreamPartitionAssignorTest {
final List<String> topics = Utils.mkList("topic1");
final UUID uuid1 = UUID.randomUUID();
- final String client1 = "client1";
- final StreamThread streamThread = new StreamThread(
- builder,
- config,
- mockClientSupplier,
- applicationId,
- client1,
- uuid1,
- new Metrics(),
- Time.SYSTEM,
- new StreamsMetadataState(builder, StreamsMetadataState.UNKNOWN_HOST),
- 0,
- stateDirectory);
-
- final StreamPartitionAssignor partitionAssignor = new StreamPartitionAssignor();
- partitionAssignor.configure(config.getConsumerConfigs(streamThread, applicationId, client1));
- partitionAssignor.setInternalTopicManager(new MockInternalTopicManager(streamThread.config, mockClientSupplier.restoreConsumer));
+ mockThreadDataProvider(Collections.<TaskId>emptySet(), Collections.<TaskId>emptySet(), uuid1, defaultPartitionGrouper, builder);
+ configurePartitionAssignor(0, userEndPoint);
+ partitionAssignor.setInternalTopicManager(new MockInternalTopicManager(config, mockClientSupplier.restoreConsumer));
final Map<String, PartitionAssignor.Subscription> subscriptions = new HashMap<>();
final Set<TaskId> emptyTasks = Collections.emptySet();
subscriptions.put("consumer1",
- new PartitionAssignor.Subscription(topics, new SubscriptionInfo(uuid1, emptyTasks, emptyTasks, myEndPoint).encode()));
+ new PartitionAssignor.Subscription(topics, new SubscriptionInfo(uuid1, emptyTasks, emptyTasks, userEndPoint).encode()));
final Map<String, PartitionAssignor.Assignment> assignments = partitionAssignor.assign(metadata, subscriptions);
final PartitionAssignor.Assignment consumerAssignment = assignments.get("consumer1");
@@ -831,32 +692,15 @@ public class StreamPartitionAssignorTest {
@Test
public void shouldThrowExceptionIfApplicationServerConfigIsNotHostPortPair() throws Exception {
- final Properties properties = configProps();
final String myEndPoint = "localhost";
- properties.put(StreamsConfig.APPLICATION_SERVER_CONFIG, myEndPoint);
- final StreamsConfig config = new StreamsConfig(properties);
- final UUID uuid1 = UUID.randomUUID();
- final String client1 = "client1";
final String applicationId = "application-id";
builder.setApplicationId(applicationId);
- final StreamThread streamThread = new StreamThread(
- builder,
- config,
- mockClientSupplier,
- applicationId,
- client1,
- uuid1,
- new Metrics(),
- Time.SYSTEM,
- new StreamsMetadataState(builder, StreamsMetadataState.UNKNOWN_HOST),
- 0,
- stateDirectory);
-
- partitionAssignor.setInternalTopicManager(new MockInternalTopicManager(streamThread.config, mockClientSupplier.restoreConsumer));
+ mockThreadDataProvider(Collections.<TaskId>emptySet(), Collections.<TaskId>emptySet(), UUID.randomUUID(), defaultPartitionGrouper, builder);
+ partitionAssignor.setInternalTopicManager(new MockInternalTopicManager(config, mockClientSupplier.restoreConsumer));
try {
- partitionAssignor.configure(config.getConsumerConfigs(streamThread, applicationId, client1));
+ configurePartitionAssignor(0, myEndPoint);
Assert.fail("expected to an exception due to invalid config");
} catch (ConfigException e) {
// pass
@@ -865,30 +709,12 @@ public class StreamPartitionAssignorTest {
@Test
public void shouldThrowExceptionIfApplicationServerConfigPortIsNotAnInteger() throws Exception {
- final Properties properties = configProps();
final String myEndPoint = "localhost:j87yhk";
- properties.put(StreamsConfig.APPLICATION_SERVER_CONFIG, myEndPoint);
- final StreamsConfig config = new StreamsConfig(properties);
- final UUID uuid1 = UUID.randomUUID();
- final String client1 = "client1";
final String applicationId = "application-id";
builder.setApplicationId(applicationId);
- final StreamThread streamThread = new StreamThread(
- builder,
- config,
- mockClientSupplier,
- applicationId,
- client1,
- uuid1,
- new Metrics(),
- Time.SYSTEM,
- new StreamsMetadataState(builder, StreamsMetadataState.UNKNOWN_HOST),
- 0,
- stateDirectory);
-
try {
- partitionAssignor.configure(config.getConsumerConfigs(streamThread, applicationId, client1));
+ configurePartitionAssignor(0, myEndPoint);
Assert.fail("expected to an exception due to invalid config");
} catch (ConfigException e) {
// pass
@@ -904,6 +730,9 @@ public class StreamPartitionAssignorTest {
AssignmentInfo assignmentInfo = new AssignmentInfo(Collections.singletonList(new TaskId(0, 0)),
Collections.<TaskId, Set<TopicPartition>>emptyMap(),
hostState);
+ mockThreadDataProvider(Collections.<TaskId>emptySet(), Collections.<TaskId>emptySet(), UUID.randomUUID(), defaultPartitionGrouper, builder);
+ configurePartitionAssignor(0, null);
+
partitionAssignor.onAssignment(new PartitionAssignor.Assignment(topic, assignmentInfo.encode()));
assertEquals(hostState, partitionAssignor.getPartitionsByHostState());
}
@@ -919,6 +748,8 @@ public class StreamPartitionAssignorTest {
hostState);
+ mockThreadDataProvider(Collections.<TaskId>emptySet(), Collections.<TaskId>emptySet(), UUID.randomUUID(), defaultPartitionGrouper, builder);
+ configurePartitionAssignor(0, null);
partitionAssignor.onAssignment(new PartitionAssignor.Assignment(topic, assignmentInfo.encode()));
final Cluster cluster = partitionAssignor.clusterMetadata();
final List<PartitionInfo> partitionInfos = cluster.partitionsForTopic("topic");
@@ -999,22 +830,15 @@ public class StreamPartitionAssignorTest {
final UUID uuid = UUID.randomUUID();
final String client = "client1";
- final StreamThread streamThread = new StreamThread(
- internalTopologyBuilder,
- config,
- mockClientSupplier,
- applicationId,
- client,
- uuid,
- new Metrics(),
- Time.SYSTEM,
- new StreamsMetadataState(internalTopologyBuilder, StreamsMetadataState.UNKNOWN_HOST),
- 0,
- stateDirectory);
-
- partitionAssignor.configure(config.getConsumerConfigs(streamThread, applicationId, client));
+ mockThreadDataProvider(Collections.<TaskId>emptySet(),
+ Collections.<TaskId>emptySet(),
+ UUID.randomUUID(),
+ defaultPartitionGrouper,
+ internalTopologyBuilder);
+ configurePartitionAssignor(0, null);
+
final MockInternalTopicManager mockInternalTopicManager = new MockInternalTopicManager(
- streamThread.config,
+ config,
mockClientSupplier.restoreConsumer);
partitionAssignor.setInternalTopicManager(mockInternalTopicManager);
@@ -1027,6 +851,7 @@ public class StreamPartitionAssignorTest {
new SubscriptionInfo(uuid, emptyTasks, emptyTasks, userEndPoint).encode()
)
);
+
final Map<String, PartitionAssignor.Assignment> assignment = partitionAssignor.assign(metadata, subscriptions);
final Map<String, Integer> expectedCreatedInternalTopics = new HashMap<>();
@@ -1042,7 +867,7 @@ public class StreamPartitionAssignorTest {
new TopicPartition(applicationId + "-count-repartition", 1),
new TopicPartition(applicationId + "-count-repartition", 2)
);
- assertThat(new HashSet(assignment.get(client).partitions()), equalTo(new HashSet(expectedAssignment)));
+ assertThat(new HashSet<>(assignment.get(client).partitions()), equalTo(new HashSet<>(expectedAssignment)));
}
@Test
@@ -1056,6 +881,12 @@ public class StreamPartitionAssignorTest {
secondHostState.put(new HostInfo("localhost", 9090), Utils.mkSet(partitionOne));
secondHostState.put(new HostInfo("other", 9090), Utils.mkSet(partitionTwo));
+ mockThreadDataProvider(Collections.<TaskId>emptySet(),
+ Collections.<TaskId>emptySet(),
+ UUID.randomUUID(),
+ defaultPartitionGrouper,
+ builder);
+ configurePartitionAssignor(0, null);
partitionAssignor.onAssignment(createAssignment(firstHostState));
assertEquals(firstHostState, partitionAssignor.getPartitionsByHostState());
partitionAssignor.onAssignment(createAssignment(secondHostState));
@@ -1072,6 +903,12 @@ public class StreamPartitionAssignorTest {
final Map<HostInfo, Set<TopicPartition>> secondHostState = Collections.singletonMap(
new HostInfo("localhost", 9090), Utils.mkSet(topicOne, topicTwo));
+ mockThreadDataProvider(Collections.<TaskId>emptySet(),
+ Collections.<TaskId>emptySet(),
+ UUID.randomUUID(),
+ defaultPartitionGrouper,
+ builder);
+ configurePartitionAssignor(0, null);
partitionAssignor.onAssignment(createAssignment(firstHostState));
assertEquals(Utils.mkSet("topic"), partitionAssignor.clusterMetadata().topics());
partitionAssignor.onAssignment(createAssignment(secondHostState));
@@ -1081,9 +918,6 @@ public class StreamPartitionAssignorTest {
@Test
public void shouldNotAddStandbyTaskPartitionsToPartitionsForHost() throws Exception {
final String applicationId = "appId";
- final Properties props = configProps();
- props.setProperty(StreamsConfig.NUM_STANDBY_REPLICAS_CONFIG, "1");
- final StreamsConfig config = new StreamsConfig(props);
final StreamsBuilder builder = new StreamsBuilder();
final InternalTopologyBuilder internalTopologyBuilder = StreamsBuilderTest.internalTopologyBuilder(builder);
@@ -1092,24 +926,15 @@ public class StreamPartitionAssignorTest {
builder.stream("topic1").groupByKey().count("count");
final UUID uuid = UUID.randomUUID();
- final String client = "client1";
+ mockThreadDataProvider(Collections.<TaskId>emptySet(),
+ Collections.<TaskId>emptySet(),
+ uuid,
+ defaultPartitionGrouper,
+ internalTopologyBuilder);
- final StreamThread streamThread = new StreamThread(
- internalTopologyBuilder,
- config,
- mockClientSupplier,
- applicationId,
- client,
- uuid,
- new Metrics(),
- Time.SYSTEM,
- new StreamsMetadataState(internalTopologyBuilder, StreamsMetadataState.UNKNOWN_HOST),
- 0,
- stateDirectory);
-
- partitionAssignor.configure(config.getConsumerConfigs(streamThread, applicationId, client));
+ configurePartitionAssignor(1, userEndPoint);
partitionAssignor.setInternalTopicManager(new MockInternalTopicManager(
- streamThread.config,
+ config,
mockClientSupplier.restoreConsumer));
final Map<String, PartitionAssignor.Subscription> subscriptions = new HashMap<>();
@@ -1133,7 +958,7 @@ public class StreamPartitionAssignorTest {
final Map<String, PartitionAssignor.Assignment> assign = partitionAssignor.assign(metadata, subscriptions);
final PartitionAssignor.Assignment consumer1Assignment = assign.get("consumer1");
final AssignmentInfo assignmentInfo = AssignmentInfo.decode(consumer1Assignment.userData());
- final Set<TopicPartition> consumer1partitions = assignmentInfo.partitionsByHost.get(new HostInfo("localhost", 2171));
+ final Set<TopicPartition> consumer1partitions = assignmentInfo.partitionsByHost.get(new HostInfo("localhost", 8080));
final Set<TopicPartition> consumer2Partitions = assignmentInfo.partitionsByHost.get(new HostInfo("other", 9090));
final HashSet<TopicPartition> allAssignedPartitions = new HashSet<>(consumer1partitions);
allAssignedPartitions.addAll(consumer2Partitions);
@@ -1148,7 +973,7 @@ public class StreamPartitionAssignorTest {
}
@Test(expected = KafkaException.class)
- public void shouldThrowKafkaExceptionIfStreamThreadConfigIsNotStreamThreadInstance() throws Exception {
+ public void shouldThrowKafkaExceptionIfStreamThreadConfigIsNotThreadDataProviderInstance() throws Exception {
final Map<String, Object> config = new HashMap<>();
config.put(StreamsConfig.NUM_STANDBY_REPLICAS_CONFIG, 1);
config.put(StreamsConfig.InternalConfig.STREAM_THREAD_INSTANCE, "i am not a stream thread");
http://git-wip-us.apache.org/repos/asf/kafka/blob/3e69ce80/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java
index f3222d0..648a15d 100644
--- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java
@@ -16,8 +16,10 @@
*/
package org.apache.kafka.streams.processor.internals;
+import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.MockConsumer;
+import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.clients.consumer.OffsetResetStrategy;
import org.apache.kafka.clients.producer.MockProducer;
import org.apache.kafka.common.KafkaException;
@@ -36,6 +38,7 @@ import org.apache.kafka.common.utils.Time;
import org.apache.kafka.common.utils.Utils;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.StreamsMetrics;
+import org.apache.kafka.streams.errors.ProcessorStateException;
import org.apache.kafka.streams.errors.StreamsException;
import org.apache.kafka.streams.processor.AbstractProcessor;
import org.apache.kafka.streams.processor.Processor;
@@ -54,6 +57,7 @@ import org.apache.kafka.test.MockTimestampExtractor;
import org.apache.kafka.test.NoOpProcessorContext;
import org.apache.kafka.test.NoOpRecordCollector;
import org.apache.kafka.test.TestUtils;
+import org.easymock.EasyMock;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
@@ -121,7 +125,6 @@ public class StreamTaskTest {
private final MockTime time = new MockTime();
private File baseDir = TestUtils.tempDirectory();
private StateDirectory stateDirectory;
- private final RecordCollectorImpl recordCollector = new RecordCollectorImpl(producer, "taskId");
private StreamsConfig config;
private StreamsConfig eosConfig;
private StreamTask task;
@@ -903,6 +906,93 @@ public class StreamTaskTest {
}
@SuppressWarnings("unchecked")
+ @Test
+ public void shouldNotViolateAtLeastOnceWhenExceptionOccursDuringFlushStateWhenCommitting() {
+ final MockProducer producer = new MockProducer();
+ final Consumer<byte[], byte[]> consumer = EasyMock.createStrictMock(Consumer.class);
+ EasyMock.expect(consumer.committed(EasyMock.anyObject(TopicPartition.class)))
+ .andStubReturn(new OffsetAndMetadata(1L));
+ EasyMock.replay(consumer);
+ final StreamTask task = new StreamTask(taskId00, applicationId, partitions, topology, consumer,
+ changelogReader, eosConfig, streamsMetrics, stateDirectory, null, time, producer) {
+
+ @Override
+ protected void flushState() {
+ throw new RuntimeException("KABOOM!");
+ }
+ };
+
+ try {
+ task.commit();
+ fail("should have thrown an exception");
+ } catch (Exception e) {
+ // all good
+ }
+ EasyMock.verify(consumer);
+ }
+
+ @SuppressWarnings("unchecked")
+ @Test
+ public void shouldNotViolateAtLeastOnceWhenExceptionOccursDuringTaskSuspension() {
+ final MockProducer producer = new MockProducer();
+ final Consumer<byte[], byte[]> consumer = EasyMock.createStrictMock(Consumer.class);
+ EasyMock.expect(consumer.committed(EasyMock.anyObject(TopicPartition.class)))
+ .andStubReturn(new OffsetAndMetadata(1L));
+ EasyMock.replay(consumer);
+ MockSourceNode sourceNode = new MockSourceNode(topic1, intDeserializer, intDeserializer) {
+ @Override
+ public void close() {
+ throw new RuntimeException("KABOOM!");
+ }
+ };
+
+ final ProcessorTopology topology = new ProcessorTopology(Collections.<ProcessorNode>singletonList(sourceNode),
+ Collections.<String, SourceNode>singletonMap(topic1[0], sourceNode),
+ Collections.<String, SinkNode>emptyMap(),
+ Collections.<StateStore>emptyList(),
+ Collections.<String, String>emptyMap(),
+ Collections.<StateStore>emptyList());
+ final StreamTask task = new StreamTask(taskId00, applicationId, Utils.mkSet(partition1), topology, consumer,
+ changelogReader, eosConfig, streamsMetrics, stateDirectory, null, time, producer);
+
+
+ try {
+ task.suspend();
+ fail("should have thrown an exception");
+ } catch (Exception e) {
+ // all good
+ }
+ EasyMock.verify(consumer);
+ }
+
+ @Test
+ public void shouldCloseStateManagerIfFailureOnTaskClose() {
+ final AtomicBoolean stateManagerCloseCalled = new AtomicBoolean(false);
+ final StreamTask streamTask = new StreamTask(taskId00, applicationId, partitions, topology, consumer,
+ changelogReader, eosConfig, streamsMetrics, stateDirectory, null,
+ time, new MockProducer<byte[], byte[]>()) {
+
+ @Override
+ void suspend(boolean val) {
+ throw new RuntimeException("KABOOM!");
+ }
+
+ @Override
+ void closeStateManager(final boolean writeCheckpoint) throws ProcessorStateException {
+ stateManagerCloseCalled.set(true);
+ }
+ };
+
+ try {
+ streamTask.close(true);
+ fail("should have thrown an exception");
+ } catch (Exception e) {
+ // all good
+ }
+ assertTrue(stateManagerCloseCalled.get());
+ }
+
+ @SuppressWarnings("unchecked")
private StreamTask createTaskThatThrowsExceptionOnClose() {
final MockSourceNode processorNode = new MockSourceNode(topic1, intDeserializer, intDeserializer) {
@Override
[4/5] kafka git commit: KAFKA-5702; extract refactor StreamThread
Posted by da...@apache.org.
http://git-wip-us.apache.org/repos/asf/kafka/blob/3e69ce80/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java
----------------------------------------------------------------------
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 c157531..099682b 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
@@ -40,20 +40,18 @@ import org.apache.kafka.common.metrics.stats.Sum;
import org.apache.kafka.common.utils.Time;
import org.apache.kafka.streams.KafkaClientSupplier;
import org.apache.kafka.streams.StreamsConfig;
+import org.apache.kafka.streams.StreamsMetrics;
import org.apache.kafka.streams.errors.LockException;
import org.apache.kafka.streams.errors.StreamsException;
import org.apache.kafka.streams.errors.TaskIdFormatException;
import org.apache.kafka.streams.processor.PartitionGrouper;
import org.apache.kafka.streams.processor.StateRestoreListener;
-import org.apache.kafka.streams.processor.StateStore;
import org.apache.kafka.streams.processor.TaskId;
-import org.apache.kafka.streams.state.HostInfo;
import org.apache.kafka.streams.state.internals.ThreadCache;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.File;
-import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
@@ -64,14 +62,11 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
-import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;
-import java.util.concurrent.atomic.AtomicReference;
-import java.util.regex.Pattern;
import static java.util.Collections.singleton;
-public class StreamThread extends Thread {
+public class StreamThread extends Thread implements ThreadDataProvider {
private static final Logger log = LoggerFactory.getLogger(StreamThread.class);
private static final AtomicInteger STREAM_THREAD_ID_SEQUENCE = new AtomicInteger(1);
@@ -157,60 +152,48 @@ public class StreamThread extends Thread {
void onChange(final Thread thread, final ThreadStateTransitionValidator newState, final ThreadStateTransitionValidator oldState);
}
- private class RebalanceListener implements ConsumerRebalanceListener {
- private final Time time;
- private final int requestTimeOut;
- RebalanceListener(final Time time, final int requestTimeOut) {
+ static class RebalanceListener implements ConsumerRebalanceListener {
+ private final Time time;
+ private final TaskManager taskManager;
+ private final StreamThread streamThread;
+ private final String logPrefix;
+
+ RebalanceListener(final Time time,
+ final TaskManager taskManager,
+ final StreamThread streamThread,
+ final String logPrefix) {
this.time = time;
- this.requestTimeOut = requestTimeOut;
+ this.taskManager = taskManager;
+ this.streamThread = streamThread;
+ this.logPrefix = logPrefix;
}
@Override
public void onPartitionsAssigned(final Collection<TopicPartition> assignment) {
- log.debug("{} at state {}: new partitions {} assigned at the end of consumer rebalance.\n" +
- "\tassigned active tasks: {}\n" +
- "\tassigned standby tasks: {}\n" +
- "\tcurrent suspended active tasks: {}\n" +
- "\tcurrent suspended standby tasks: {}\n" +
- "\tprevious active tasks: {}",
- logPrefix,
- state,
- assignment,
- partitionAssignor.activeTasks().keySet(),
- partitionAssignor.standbyTasks().keySet(),
- suspendedTasks.keySet(),
- suspendedStandbyTasks.keySet(),
- prevActiveTasks);
-
final long start = time.milliseconds();
try {
- storeChangelogReader =
- new StoreChangelogReader(getName(), restoreConsumer, time, requestTimeOut,
- globalStateRestoreListener);
- setState(State.ASSIGNING_PARTITIONS);
- // do this first as we may have suspended standby tasks that
- // will become active or vice versa
- closeNonAssignedSuspendedStandbyTasks();
- closeNonAssignedSuspendedTasks();
- addStreamTasks(assignment, start);
- storeChangelogReader.restore();
- addStandbyTasks(start);
- streamsMetadataState.onChange(partitionAssignor.getPartitionsByHostState(), partitionAssignor.clusterMetadata());
- setState(State.RUNNING);
+ streamThread.setState(State.ASSIGNING_PARTITIONS);
+ taskManager.createTasks(assignment);
+ final RuntimeException exception = streamThread.unAssignChangeLogPartitions();
+ if (exception != null) {
+ throw exception;
+ }
+ streamThread.refreshMetadataState();
+ streamThread.setState(State.RUNNING);
} catch (final Throwable t) {
- rebalanceException = t;
+ streamThread.setRebalanceException(t);
throw t;
} finally {
log.info("{} partition assignment took {} ms.\n" +
- "\tcurrent active tasks: {}\n" +
- "\tcurrent standby tasks: {}\n" +
- "\tprevious active tasks: {}\n",
- logPrefix,
- time.milliseconds() - start,
- activeTasks.keySet(),
- standbyTasks.keySet(),
- prevActiveTasks);
+ "\tcurrent active tasks: {}\n" +
+ "\tcurrent standby tasks: {}\n" +
+ "\tprevious active tasks: {}\n",
+ logPrefix,
+ time.milliseconds() - start,
+ taskManager.activeTaskIds(),
+ taskManager.standbyTaskIds(),
+ taskManager.prevActiveTaskIds());
}
}
@@ -220,39 +203,74 @@ public class StreamThread extends Thread {
"\tcurrent assigned active tasks: {}\n" +
"\tcurrent assigned standby tasks: {}\n",
logPrefix,
- state,
+ streamThread.state,
assignment,
- activeTasks.keySet(), standbyTasks.keySet());
+ taskManager.activeTaskIds(),
+ taskManager.standbyTaskIds());
final long start = time.milliseconds();
try {
- setState(State.PARTITIONS_REVOKED);
+ streamThread.setState(State.PARTITIONS_REVOKED);
// suspend active tasks
- suspendTasksAndState();
+ taskManager.suspendTasksAndState();
} catch (final Throwable t) {
- rebalanceException = t;
+ streamThread.setRebalanceException(t);
throw t;
} finally {
- streamsMetadataState.onChange(Collections.<HostInfo, Set<TopicPartition>>emptyMap(), partitionAssignor.clusterMetadata());
- removeStreamTasks();
- removeStandbyTasks();
+ streamThread.refreshMetadataState();
+ taskManager.removeTasks();
+ streamThread.clearStandbyRecords();
log.info("{} partition revocation took {} ms.\n" +
"\tsuspended active tasks: {}\n" +
"\tsuspended standby tasks: {}",
logPrefix,
time.milliseconds() - start,
- suspendedTasks.keySet(),
- suspendedStandbyTasks.keySet());
+ taskManager.suspendedActiveTaskIds(),
+ taskManager.suspendedStandbyTaskIds());
}
}
}
- abstract class AbstractTaskCreator {
+
+ static abstract class AbstractTaskCreator {
final static long MAX_BACKOFF_TIME_MS = 1000L;
- void retryWithBackoff(final Map<TaskId, Set<TopicPartition>> tasksToBeCreated, final long start) {
+ private final long rebalanceTimeoutMs;
+ final String applicationId;
+ final InternalTopologyBuilder builder;
+ final StreamsConfig config;
+ final StreamsMetrics streamsMetrics;
+ final StateDirectory stateDirectory;
+ final Sensor taskCreatedSensor;
+ final ChangelogReader storeChangelogReader;
+ final Time time;
+ final String logPrefix;
+
+ AbstractTaskCreator(final InternalTopologyBuilder builder,
+ final StreamsConfig config,
+ final StreamsMetrics streamsMetrics,
+ final StateDirectory stateDirectory,
+ final Sensor taskCreatedSensor,
+ final ChangelogReader storeChangelogReader,
+ final Time time,
+ final long rebalanceTimeoutMs,
+ final String logPrefix) {
+ this.applicationId = config.getString(StreamsConfig.APPLICATION_ID_CONFIG);
+ this.builder = builder;
+ this.config = config;
+ this.streamsMetrics = streamsMetrics;
+ this.stateDirectory = stateDirectory;
+ this.taskCreatedSensor = taskCreatedSensor;
+ this.storeChangelogReader = storeChangelogReader;
+ this.time = time;
+ this.rebalanceTimeoutMs = rebalanceTimeoutMs;
+ this.logPrefix = logPrefix;
+ }
+
+ Map<Task, Set<TopicPartition>> retryWithBackoff(final Consumer<byte[], byte[]> consumer, final Map<TaskId, Set<TopicPartition>> tasksToBeCreated, final long start) {
long backoffTimeMs = 50L;
final Set<TaskId> retryingTasks = new HashSet<>();
+ final Map<Task, Set<TopicPartition>> createdTasks = new HashMap<>();
while (true) {
final Iterator<Map.Entry<TaskId, Set<TopicPartition>>> it = tasksToBeCreated.entrySet().iterator();
while (it.hasNext()) {
@@ -261,7 +279,11 @@ public class StreamThread extends Thread {
final Set<TopicPartition> partitions = newTaskAndPartitions.getValue();
try {
- createTask(taskId, partitions);
+ final Task task = createTask(consumer, taskId, partitions);
+ if (task != null) {
+ log.trace("{} Created task {} with assigned partitions {}", logPrefix, taskId, partitions);
+ createdTasks.put(task, partitions);
+ }
it.remove();
backoffTimeMs = 50L;
retryingTasks.remove(taskId);
@@ -286,48 +308,138 @@ public class StreamThread extends Thread {
// ignore
}
}
+ return createdTasks;
}
- abstract void createTask(final TaskId id, final Set<TopicPartition> partitions);
+ abstract Task createTask(final Consumer<byte[], byte[]> consumer, final TaskId id, final Set<TopicPartition> partitions);
+
+ public void close() {}
}
- class TaskCreator extends AbstractTaskCreator {
+ static class TaskCreator extends AbstractTaskCreator {
+ private final ThreadCache cache;
+ private final KafkaClientSupplier clientSupplier;
+ private final String threadClientId;
+ private final Producer<byte[], byte[]> threadProducer;
+
+
+ TaskCreator(final InternalTopologyBuilder builder,
+ final StreamsConfig config,
+ final StreamsMetrics streamsMetrics,
+ final StateDirectory stateDirectory,
+ final Sensor taskCreatedSensor,
+ final ChangelogReader storeChangelogReader,
+ final ThreadCache cache,
+ final Time time,
+ final KafkaClientSupplier clientSupplier,
+ final Producer<byte[], byte[]> threadProducer,
+ final String threadClientId,
+ final long rebalanceTimeoutMs,
+ final String logPrefix) {
+ super(
+ builder,
+ config,
+ streamsMetrics,
+ stateDirectory,
+ taskCreatedSensor,
+ storeChangelogReader,
+ time,
+ rebalanceTimeoutMs,
+ logPrefix);
+ this.cache = cache;
+ this.clientSupplier = clientSupplier;
+ this.threadProducer = threadProducer;
+ this.threadClientId = threadClientId;
+ }
+
@Override
- void createTask(final TaskId taskId, final Set<TopicPartition> partitions) {
- final StreamTask task = createStreamTask(taskId, partitions);
+ StreamTask createTask(final Consumer<byte[], byte[]> consumer, final TaskId taskId, final Set<TopicPartition> partitions) {
+ taskCreatedSensor.record();
+
+ return new StreamTask(
+ taskId,
+ applicationId,
+ partitions,
+ builder.build(taskId.topicGroupId),
+ consumer,
+ storeChangelogReader,
+ config,
+ streamsMetrics,
+ stateDirectory,
+ cache,
+ time,
+ createProducer(taskId));
- activeTasks.put(taskId, task);
+ }
- for (final TopicPartition partition : partitions) {
- activeTasksByPartition.put(partition, task);
+ @Override
+ public void close() {
+ if (threadProducer != null) {
+ try {
+ threadProducer.close();
+ } catch (final Throwable e) {
+ log.error("{} Failed to close producer due to the following error:", logPrefix, e);
+ }
}
}
- }
- class StandbyTaskCreator extends AbstractTaskCreator {
- private final Map<TopicPartition, Long> checkpointedOffsets;
+ private Producer<byte[], byte[]> createProducer(final TaskId id) {
+ // eos
+ if (threadProducer == null) {
+ final Map<String, Object> producerConfigs = config.getProducerConfigs(threadClientId + "-" + id);
+ log.info("{} Creating producer client for task {}", logPrefix, id);
+ producerConfigs.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, applicationId + "-" + id);
+ return clientSupplier.getProducer(producerConfigs);
+ }
+
+ return threadProducer;
- StandbyTaskCreator(final Map<TopicPartition, Long> checkpointedOffsets) {
- this.checkpointedOffsets = checkpointedOffsets;
+ }
+ }
+
+ static class StandbyTaskCreator extends AbstractTaskCreator {
+ StandbyTaskCreator(final InternalTopologyBuilder builder,
+ final StreamsConfig config,
+ final StreamsMetrics streamsMetrics,
+ final StateDirectory stateDirectory,
+ final Sensor taskCreatedSensor,
+ final ChangelogReader storeChangelogReader,
+ final Time time,
+ final long rebalanceTimeoutMs,
+ final String logPrefix) {
+ super(builder,
+ config,
+ streamsMetrics,
+ stateDirectory,
+ taskCreatedSensor,
+ storeChangelogReader,
+ time,
+ rebalanceTimeoutMs,
+ logPrefix);
}
@Override
- void createTask(final TaskId taskId, final Set<TopicPartition> partitions) {
- final StandbyTask task = createStandbyTask(taskId, partitions);
- updateStandByTaskMaps(checkpointedOffsets, taskId, partitions, task);
+ StandbyTask createTask(final Consumer<byte[], byte[]> consumer, final TaskId taskId, final Set<TopicPartition> partitions) {
+ taskCreatedSensor.record();
+
+ final ProcessorTopology topology = builder.build(taskId.topicGroupId);
+
+ if (!topology.stateStores().isEmpty()) {
+ return new StandbyTask(taskId, applicationId, partitions, topology, consumer, storeChangelogReader, config, streamsMetrics, stateDirectory);
+ } else {
+ log.trace("{} Skipped standby task {} with assigned partitions {} since it does not have any state stores to materialize", logPrefix, taskId, partitions);
+
+ return null;
+ }
}
- }
- interface StreamTaskAction {
- String name();
- void apply(final StreamTask task);
}
/**
* This class extends {@link StreamsMetricsImpl(Metrics, String, String, Map)} and
* overrides one of its functions for efficiency
*/
- private class StreamsMetricsThreadImpl extends StreamsMetricsImpl {
+ static class StreamsMetricsThreadImpl extends StreamsMetricsImpl {
final Sensor commitTimeSensor;
final Sensor pollTimeSensor;
final Sensor processTimeSensor;
@@ -369,12 +481,6 @@ public class StreamThread extends Thread {
}
-
- @Override
- public void recordLatency(final Sensor sensor, final long startNs, final long endNs) {
- sensor.record(endNs - startNs, timerStartedMs);
- }
-
void removeAllSensors() {
removeSensor(commitTimeSensor);
removeSensor(pollTimeSensor);
@@ -388,93 +494,69 @@ public class StreamThread extends Thread {
}
- private volatile State state = State.CREATED;
private final Object stateLock = new Object();
-
- private StreamThread.StateListener stateListener = null;
- final PartitionGrouper partitionGrouper;
private final StreamsMetadataState streamsMetadataState;
- public final String applicationId;
- public final String clientId;
- public final UUID processId;
-
- protected final StreamsConfig config;
- protected final InternalTopologyBuilder builder;
- Producer<byte[], byte[]> threadProducer;
- private final KafkaClientSupplier clientSupplier;
- protected final Consumer<byte[], byte[]> consumer;
- final Consumer<byte[], byte[]> restoreConsumer;
-
private final String logPrefix;
- private final String threadClientId;
- private final Pattern sourceTopicPattern;
- private final Map<TaskId, StreamTask> activeTasks;
- private final Map<TaskId, StandbyTask> standbyTasks;
- private final Map<TopicPartition, StreamTask> activeTasksByPartition;
- private final Map<TopicPartition, StandbyTask> standbyTasksByPartition;
- private final Set<TaskId> prevActiveTasks;
- private final Map<TaskId, StreamTask> suspendedTasks;
- private final Map<TaskId, StandbyTask> suspendedStandbyTasks;
+ private final TaskManager taskManager;
private final Time time;
- private final int rebalanceTimeoutMs;
private final long pollTimeMs;
private final long commitTimeMs;
+ private final PartitionGrouper partitionGrouper;
+ private final UUID processId;
+ private final StateDirectory stateDirectory;
private final StreamsMetricsThreadImpl streamsMetrics;
- // TODO: this is not private only for tests, should be better refactored
- final StateDirectory stateDirectory;
+
+ private long lastCommitMs;
private String originalReset;
- private StreamPartitionAssignor partitionAssignor;
+ private ThreadMetadataProvider metadataProvider;
+ private boolean processStandbyRecords = false;
+ private Throwable rebalanceException;
+ private Map<TopicPartition, List<ConsumerRecord<byte[], byte[]>>> standbyRecords = new HashMap<>();
+ private StreamThread.StateListener stateListener;
+ private volatile State state = State.CREATED;
private long timerStartedMs;
- private long lastCommitMs;
- private Throwable rebalanceException = null;
- private final boolean eosEnabled;
- private Map<TopicPartition, List<ConsumerRecord<byte[], byte[]>>> standbyRecords;
- private boolean processStandbyRecords = false;
+ final StreamsConfig config;
+ final ConsumerRebalanceListener rebalanceListener;
+ final Consumer<byte[], byte[]> restoreConsumer;
- private final ThreadCache cache;
- private StoreChangelogReader storeChangelogReader;
+ protected final Consumer<byte[], byte[]> consumer;
+ protected final InternalTopologyBuilder builder;
- private final TaskCreator taskCreator = new TaskCreator();
+ public final String applicationId;
+ public final String clientId;
- final ConsumerRebalanceListener rebalanceListener;
- private StateRestoreListener globalStateRestoreListener;
private final static int UNLIMITED_RECORDS = -1;
public StreamThread(final InternalTopologyBuilder builder,
- final StreamsConfig config,
- final KafkaClientSupplier clientSupplier,
- final String applicationId,
final String clientId,
+ final String threadClientId,
+ final StreamsConfig config,
final UUID processId,
- final Metrics metrics,
final Time time,
final StreamsMetadataState streamsMetadataState,
- final long cacheSizeBytes,
+ final TaskManager taskManager,
+ final StreamsMetricsThreadImpl streamsMetrics,
+ final KafkaClientSupplier clientSupplier,
+ final Consumer<byte[], byte[]> restoreConsumer,
final StateDirectory stateDirectory) {
- super(clientId + "-StreamThread-" + STREAM_THREAD_ID_SEQUENCE.getAndIncrement());
- this.applicationId = applicationId;
- this.config = config;
+ super(threadClientId);
this.builder = builder;
- this.clientSupplier = clientSupplier;
- sourceTopicPattern = builder.sourceTopicPattern();
this.clientId = clientId;
+ this.applicationId = config.getString(StreamsConfig.APPLICATION_ID_CONFIG);
+ this.pollTimeMs = config.getLong(StreamsConfig.POLL_MS_CONFIG);
+ this.commitTimeMs = config.getLong(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG);
this.processId = processId;
- partitionGrouper = config.getConfiguredInstance(StreamsConfig.PARTITION_GROUPER_CLASS_CONFIG, PartitionGrouper.class);
+ this.time = time;
this.streamsMetadataState = streamsMetadataState;
- threadClientId = getName();
- logPrefix = String.format("stream-thread [%s]", threadClientId);
-
- streamsMetrics = new StreamsMetricsThreadImpl(metrics, "stream-metrics", "thread." + threadClientId,
- Collections.singletonMap("client-id", threadClientId));
- if (config.getLong(StreamsConfig.CACHE_MAX_BYTES_BUFFERING_CONFIG) < 0) {
- log.warn("{} Negative cache size passed in thread. Reverting to cache size of 0 bytes", logPrefix);
- }
- cache = new ThreadCache(threadClientId, cacheSizeBytes, streamsMetrics);
- eosEnabled = StreamsConfig.EXACTLY_ONCE.equals(config.getString(StreamsConfig.PROCESSING_GUARANTEE_CONFIG));
-
-
- // set the consumer clients
+ this.taskManager = taskManager;
+ this.logPrefix = logPrefix(threadClientId);
+ this.streamsMetrics = streamsMetrics;
+ this.restoreConsumer = restoreConsumer;
+ this.stateDirectory = stateDirectory;
+ this.rebalanceListener = new RebalanceListener(time, taskManager, this, logPrefix);
+ this.config = config;
+ this.partitionGrouper = config.getConfiguredInstance(StreamsConfig.PARTITION_GROUPER_CLASS_CONFIG, PartitionGrouper.class);
log.info("{} Creating consumer client", logPrefix);
final Map<String, Object> consumerConfigs = config.getConsumerConfigs(this, applicationId, threadClientId);
@@ -483,34 +565,101 @@ public class StreamThread extends Thread {
log.info("{} Custom offset resets specified updating configs original auto offset reset {}", logPrefix, originalReset);
consumerConfigs.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "none");
}
+ this.consumer = clientSupplier.getConsumer(consumerConfigs);
+ taskManager.setConsumer(consumer);
+ }
+
+ @SuppressWarnings("ConstantConditions")
+ public static StreamThread create(final InternalTopologyBuilder builder,
+ final StreamsConfig config,
+ final KafkaClientSupplier clientSupplier,
+ final UUID processId,
+ final String clientId,
+ final Metrics metrics,
+ final Time time,
+ final StreamsMetadataState streamsMetadataState,
+ final long cacheSizeBytes,
+ final StateDirectory stateDirectory,
+ final StateRestoreListener stateRestoreListener) {
+
+ final String threadClientId = clientId + "-StreamThread-" + STREAM_THREAD_ID_SEQUENCE.getAndIncrement();
+ final StreamsMetricsThreadImpl streamsMetrics = new StreamsMetricsThreadImpl(metrics,
+ "stream-metrics",
+ "thread." + threadClientId,
+ Collections.singletonMap("client-id",
+ threadClientId));
+
+ final String logPrefix = logPrefix(threadClientId);
+ if (config.getLong(StreamsConfig.CACHE_MAX_BYTES_BUFFERING_CONFIG) < 0) {
+ log.warn("{} Negative cache size passed in thread. Reverting to cache size of 0 bytes", logPrefix);
+ }
+ final ThreadCache cache = new ThreadCache(threadClientId, cacheSizeBytes, streamsMetrics);
+
+ final boolean eosEnabled = StreamsConfig.EXACTLY_ONCE.equals(config.getString(StreamsConfig.PROCESSING_GUARANTEE_CONFIG));
- consumer = clientSupplier.getConsumer(consumerConfigs);
log.info("{} Creating restore consumer client", logPrefix);
- restoreConsumer = clientSupplier.getRestoreConsumer(config.getRestoreConsumerConfigs(threadClientId));
- // initialize the task list
- // activeTasks needs to be concurrent as it can be accessed
- // by QueryableState
- activeTasks = new ConcurrentHashMap<>();
- standbyTasks = new HashMap<>();
- activeTasksByPartition = new HashMap<>();
- standbyTasksByPartition = new HashMap<>();
- prevActiveTasks = new HashSet<>();
- suspendedTasks = new HashMap<>();
- suspendedStandbyTasks = new HashMap<>();
-
- // standby KTables
- standbyRecords = new HashMap<>();
+ final Map<String, Object> consumerConfigs = config.getRestoreConsumerConfigs(threadClientId);
+ final Consumer<byte[], byte[]> restoreConsumer = clientSupplier.getRestoreConsumer(consumerConfigs);
+
- this.stateDirectory = stateDirectory;
final Object maxPollInterval = consumerConfigs.get(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG);
- rebalanceTimeoutMs = (Integer) ConfigDef.parseType(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, maxPollInterval, Type.INT);
- pollTimeMs = config.getLong(StreamsConfig.POLL_MS_CONFIG);
- commitTimeMs = config.getLong(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG);
+ final int rebalanceTimeoutMs = (Integer) ConfigDef.parseType(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, maxPollInterval, Type.INT);
+
+ final StoreChangelogReader changelogReader = new StoreChangelogReader(threadClientId,
+ restoreConsumer,
+ time,
+ config.getInt(ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG),
+ stateRestoreListener);
+
+ Producer<byte[], byte[]> threadProducer = null;
+ if (!eosEnabled) {
+ final Map<String, Object> producerConfigs = config.getProducerConfigs(threadClientId);
+ log.info("{} Creating shared producer client", logPrefix);
+ threadProducer = clientSupplier.getProducer(producerConfigs);
+ }
+
+ final AbstractTaskCreator activeTaskCreator = new TaskCreator(builder,
+ config,
+ streamsMetrics,
+ stateDirectory,
+ streamsMetrics.taskCreatedSensor,
+ changelogReader,
+ cache,
+ time,
+ clientSupplier,
+ threadProducer,
+ threadClientId,
+ rebalanceTimeoutMs,
+ logPrefix);
+ final AbstractTaskCreator standbyTaskCreator = new StandbyTaskCreator(builder,
+ config,
+ streamsMetrics,
+ stateDirectory,
+ streamsMetrics.taskCreatedSensor,
+ changelogReader,
+ time,
+ rebalanceTimeoutMs,
+ logPrefix);
+ final TaskManager taskManager = new TaskManager(changelogReader, time, logPrefix, restoreConsumer, activeTaskCreator, standbyTaskCreator);
+
+ return new StreamThread(builder,
+ clientId,
+ threadClientId,
+ config,
+ processId,
+ time,
+ streamsMetadataState,
+ taskManager,
+ streamsMetrics,
+ clientSupplier,
+ restoreConsumer,
+ stateDirectory);
- this.time = time;
- timerStartedMs = time.milliseconds();
- lastCommitMs = timerStartedMs;
- rebalanceListener = new RebalanceListener(time, config.getInt(ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG));
+
+ }
+
+ private static String logPrefix(final String threadClientId) {
+ return String.format("stream-thread [%s]", threadClientId);
}
/**
@@ -540,22 +689,26 @@ public class StreamThread extends Thread {
}
}
+ void setRebalanceException(final Throwable rebalanceException) {
+ this.rebalanceException = rebalanceException;
+ }
+
/**
* Main event loop for polling, and processing records through topologies.
*/
private void runLoop() {
long recordsProcessedBeforeCommit = UNLIMITED_RECORDS;
- consumer.subscribe(sourceTopicPattern, rebalanceListener);
-
+ consumer.subscribe(builder.sourceTopicPattern(), rebalanceListener);
+ lastCommitMs = time.milliseconds();
while (stillRunning()) {
timerStartedMs = time.milliseconds();
// try to fetch some records if necessary
final ConsumerRecords<byte[], byte[]> records = pollRequests();
- if (records != null && !records.isEmpty() && !activeTasks.isEmpty()) {
+ if (records != null && !records.isEmpty() && taskManager.hasActiveTasks()) {
streamsMetrics.pollTimeSensor.record(computeLatency(), timerStartedMs);
addRecordsToTasks(records);
- final long totalProcessed = processAndPunctuateStreamTime(activeTasks, recordsProcessedBeforeCommit);
+ final long totalProcessed = processAndPunctuateStreamTime(taskManager.activeTasks(), recordsProcessedBeforeCommit);
if (totalProcessed > 0) {
final long processLatency = computeLatency();
streamsMetrics.processTimeSensor.record(processLatency / (double) totalProcessed,
@@ -646,7 +799,7 @@ public class StreamThread extends Thread {
int numAddedRecords = 0;
for (final TopicPartition partition : records.partitions()) {
- final StreamTask task = activeTasksByPartition.get(partition);
+ final Task task = taskManager.activeTask(partition);
numAddedRecords += task.addRecords(partition, records.records(partition));
}
streamsMetrics.skippedRecordsSensor.record(records.count() - numAddedRecords, timerStartedMs);
@@ -661,7 +814,7 @@ public class StreamThread extends Thread {
* if UNLIMITED_RECORDS, then commit is never called
* @return Number of records processed since last commit.
*/
- private long processAndPunctuateStreamTime(final Map<TaskId, StreamTask> tasks,
+ private long processAndPunctuateStreamTime(final Map<TaskId, Task> tasks,
final long recordsProcessedBeforeCommit) {
long totalProcessedEachRound;
@@ -670,9 +823,9 @@ public class StreamThread extends Thread {
// until no task has any records left
do {
totalProcessedEachRound = 0;
- final Iterator<Map.Entry<TaskId, StreamTask>> it = tasks.entrySet().iterator();
+ final Iterator<Map.Entry<TaskId, Task>> it = tasks.entrySet().iterator();
while (it.hasNext()) {
- final StreamTask task = it.next().getValue();
+ final Task task = it.next().getValue();
try {
// we processed one record,
// if more are buffered waiting for the next round
@@ -684,7 +837,7 @@ public class StreamThread extends Thread {
totalProcessedSinceLastMaybeCommit++;
}
} catch (final ProducerFencedException e) {
- closeZombieTask(task);
+ taskManager.closeZombieTask(task);
it.remove();
}
}
@@ -700,7 +853,7 @@ public class StreamThread extends Thread {
} while (totalProcessedEachRound != 0);
// go over the tasks again to punctuate or commit
- final RuntimeException e = performOnStreamTasks(new StreamTaskAction() {
+ final RuntimeException e = taskManager.performOnActiveTasks(new TaskManager.TaskAction() {
private String name;
@Override
public String name() {
@@ -708,7 +861,7 @@ public class StreamThread extends Thread {
}
@Override
- public void apply(final StreamTask task) {
+ public void apply(final Task task) {
name = "punctuate";
maybePunctuateStreamTime(task);
if (task.commitNeeded()) {
@@ -733,7 +886,7 @@ public class StreamThread extends Thread {
return totalProcessedSinceLastMaybeCommit;
}
- private void maybePunctuateStreamTime(final StreamTask task) {
+ private void maybePunctuateStreamTime(final Task task) {
try {
// check whether we should punctuate based on the task's partition group timestamp;
// which are essentially based on record timestamp.
@@ -747,14 +900,14 @@ public class StreamThread extends Thread {
}
private void maybePunctuateSystemTime() {
- final RuntimeException e = performOnStreamTasks(new StreamTaskAction() {
+ final RuntimeException e = taskManager.performOnActiveTasks(new TaskManager.TaskAction() {
@Override
public String name() {
return "punctuate";
}
@Override
- public void apply(final StreamTask task) {
+ public void apply(final Task task) {
try {
// check whether we should punctuate based on system timestamp
if (task.maybePunctuateSystemTime()) {
@@ -810,14 +963,14 @@ public class StreamThread extends Thread {
if (commitTimeMs >= 0 && lastCommitMs + commitTimeMs < now) {
if (log.isTraceEnabled()) {
log.trace("{} Committing all active tasks {} and standby tasks {} since {}ms has elapsed (commit interval is {}ms)",
- logPrefix, activeTasks.keySet(), standbyTasks.keySet(), now - lastCommitMs, commitTimeMs);
+ logPrefix, taskManager.activeTaskIds(), taskManager.standbyTaskIds(), now - lastCommitMs, commitTimeMs);
}
commitAll();
if (log.isDebugEnabled()) {
log.info("{} Committed all active tasks {} and standby tasks {} in {}ms",
- logPrefix, activeTasks.keySet(), standbyTasks.keySet(), timerStartedMs - now);
+ logPrefix, taskManager.activeTaskIds(), taskManager.standbyTaskIds(), timerStartedMs - now);
}
lastCommitMs = now;
@@ -830,30 +983,32 @@ public class StreamThread extends Thread {
* Commit the states of all its tasks
*/
private void commitAll() {
- final RuntimeException e = performOnStreamTasks(new StreamTaskAction() {
+ final TaskManager.TaskAction commitAction = new TaskManager.TaskAction() {
@Override
public String name() {
return "commit";
}
@Override
- public void apply(final StreamTask task) {
+ public void apply(final Task task) {
commitOne(task);
}
- });
+ };
+ final RuntimeException e = taskManager.performOnActiveTasks(commitAction);
if (e != null) {
throw e;
}
- for (final StandbyTask task : standbyTasks.values()) {
- commitOne(task);
+ final RuntimeException standbyTaskCommitException = taskManager.performOnStandbyTasks(commitAction);
+ if (standbyTaskCommitException != null) {
+ throw standbyTaskCommitException;
}
}
/**
* Commit the state of a task
*/
- private void commitOne(final AbstractTask task) {
+ private void commitOne(final Task task) {
try {
task.commit();
} catch (final CommitFailedException e) {
@@ -869,7 +1024,7 @@ public class StreamThread extends Thread {
}
private void maybeUpdateStandbyTasks(final long now) {
- if (!standbyTasks.isEmpty()) {
+ if (taskManager.hasStandbyTasks()) {
if (processStandbyRecords) {
if (!standbyRecords.isEmpty()) {
final Map<TopicPartition, List<ConsumerRecord<byte[], byte[]>>> remainingStandbyRecords = new HashMap<>();
@@ -878,7 +1033,7 @@ public class StreamThread extends Thread {
final TopicPartition partition = entry.getKey();
List<ConsumerRecord<byte[], byte[]>> remaining = entry.getValue();
if (remaining != null) {
- final StandbyTask task = standbyTasksByPartition.get(partition);
+ final Task task = taskManager.standbyTask(partition);
remaining = task.update(partition, remaining);
if (remaining != null) {
remainingStandbyRecords.put(partition, remaining);
@@ -890,7 +1045,7 @@ public class StreamThread extends Thread {
standbyRecords = remainingStandbyRecords;
- log.debug("{} Updated standby tasks {} in {}ms", logPrefix, standbyTasks.keySet(), time.milliseconds() - now);
+ log.debug("{} Updated standby tasks {} in {}ms", logPrefix, taskManager.standbyTaskIds(), time.milliseconds() - now);
}
processStandbyRecords = false;
}
@@ -899,7 +1054,7 @@ public class StreamThread extends Thread {
if (!records.isEmpty()) {
for (final TopicPartition partition : records.partitions()) {
- final StandbyTask task = standbyTasksByPartition.get(partition);
+ final Task task = taskManager.standbyTask(partition);
if (task == null) {
throw new StreamsException(logPrefix + " Missing standby task for partition " + partition);
@@ -950,15 +1105,25 @@ public class StreamThread extends Thread {
}
}
- public Map<TaskId, StreamTask> tasks() {
- return Collections.unmodifiableMap(activeTasks);
+ public Map<TaskId, Task> tasks() {
+ return Collections.unmodifiableMap(taskManager.activeTasks());
}
/**
* Returns ids of tasks that were being executed before the rebalance.
*/
public Set<TaskId> prevActiveTasks() {
- return Collections.unmodifiableSet(prevActiveTasks);
+ return taskManager.prevActiveTaskIds();
+ }
+
+ @Override
+ public InternalTopologyBuilder builder() {
+ return builder;
+ }
+
+ @Override
+ public String name() {
+ return getName();
}
/**
@@ -991,6 +1156,21 @@ public class StreamThread extends Thread {
return tasks;
}
+ @Override
+ public UUID processId() {
+ return processId;
+ }
+
+ @Override
+ public StreamsConfig config() {
+ return config;
+ }
+
+ @Override
+ public PartitionGrouper partitionGrouper() {
+ return partitionGrouper;
+ }
+
/**
* Set the {@link StreamThread.StateListener} to be notified when state changes. Note this API is internal to
* Kafka Streams and is not intended to be used by an external application.
@@ -1000,16 +1180,6 @@ public class StreamThread extends Thread {
}
/**
- * Set the listener invoked at the beginning, end of batch updates and the conclusion of
- * restoring a {@link StateStore}.
- *
- * @param globalStateRestoreListener listener for capturing state store restoration status.
- */
- public void setGlobalStateRestoreListener(final StateRestoreListener globalStateRestoreListener) {
- this.globalStateRestoreListener = globalStateRestoreListener;
- }
-
- /**
* @return The state this instance is in
*/
public State state() {
@@ -1065,6 +1235,7 @@ public class StreamThread extends Thread {
* This is useful in debugging scenarios.
* @return A string representation of the StreamThread instance.
*/
+ @SuppressWarnings("ThrowableNotThrown")
public String toString(final String indent) {
final StringBuilder sb = new StringBuilder()
.append(indent).append("StreamsThread appId: ").append(applicationId).append("\n")
@@ -1072,46 +1243,41 @@ public class StreamThread extends Thread {
.append(indent).append("\tStreamsThread threadId: ").append(getName()).append("\n");
// iterate and print active tasks
- if (activeTasks != null) {
- sb.append(indent).append("\tActive tasks:\n");
- for (final Map.Entry<TaskId, StreamTask> entry : activeTasks.entrySet()) {
- final StreamTask task = entry.getValue();
- sb.append(indent).append(task.toString(indent + "\t\t"));
+ final TaskManager.TaskAction printAction = new TaskManager.TaskAction() {
+ @Override
+ public String name() {
+ return "print";
}
- }
- // iterate and print standby tasks
- if (standbyTasks != null) {
- sb.append(indent).append("\tStandby tasks:\n");
- for (final StandbyTask task : standbyTasks.values()) {
+ @Override
+ public void apply(final Task task) {
sb.append(indent).append(task.toString(indent + "\t\t"));
}
- sb.append("\n");
- }
+ };
+ sb.append(indent).append("\tActive tasks:\n");
+ taskManager.performOnActiveTasks(printAction);
+ sb.append(indent).append("\tStandby tasks:\n");
+ taskManager.performOnStandbyTasks(printAction);
return sb.toString();
}
String threadClientId() {
- return threadClientId;
+ return getName();
}
- void setPartitionAssignor(final StreamPartitionAssignor partitionAssignor) {
- this.partitionAssignor = partitionAssignor;
+ public void setThreadMetadataProvider(final ThreadMetadataProvider metadataProvider) {
+ this.metadataProvider = metadataProvider;
+ taskManager.setThreadMetadataProvider(metadataProvider);
}
private void shutdown(final boolean cleanRun) {
log.info("{} Shutting down", logPrefix);
- shutdownTasksAndState(cleanRun);
+ taskManager.shutdown(cleanRun);
// close all embedded clients
- if (threadProducer != null) {
- try {
- threadProducer.close();
- } catch (final Throwable e) {
- log.error("{} Failed to close producer due to the following error:", logPrefix, e);
- }
- }
+ taskManager.closeProducer();
+
try {
consumer.close();
} catch (final Throwable e) {
@@ -1122,105 +1288,13 @@ public class StreamThread extends Thread {
} catch (final Throwable e) {
log.error("{} Failed to close restore consumer due to the following error:", logPrefix, e);
}
- try {
- partitionAssignor.close();
- } catch (final Throwable e) {
- log.error("{} Failed to close KafkaStreamClient due to the following error:", logPrefix, e);
- }
-
- removeStreamTasks();
- removeStandbyTasks();
-
- // clean up global tasks
+ taskManager.removeTasks();
log.info("{} Stream thread shutdown complete", logPrefix);
setState(State.DEAD);
streamsMetrics.removeAllSensors();
}
- private void shutdownTasksAndState(final boolean cleanRun) {
- log.debug("{} Shutting down all active tasks {}, standby tasks {}, suspended tasks {}, and suspended standby tasks {}",
- logPrefix, activeTasks.keySet(), standbyTasks.keySet(),
- suspendedTasks.keySet(), suspendedStandbyTasks.keySet());
-
- for (final AbstractTask task : allTasks()) {
- try {
- task.close(cleanRun);
- } catch (final RuntimeException e) {
- log.error("{} Failed while closing {} {} due to the following error:",
- logPrefix,
- task.getClass().getSimpleName(),
- task.id(),
- e);
- }
- }
-
- // remove the changelog partitions from restore consumer
- unAssignChangeLogPartitions();
- }
-
- /**
- * Similar to shutdownTasksAndState, however does not close the task managers, in the hope that
- * soon the tasks will be assigned again
- */
- private void suspendTasksAndState() {
- log.debug("{} Suspending all active tasks {} and standby tasks {}",
- logPrefix, activeTasks.keySet(), standbyTasks.keySet());
-
- final AtomicReference<RuntimeException> firstException = new AtomicReference<>(null);
-
- firstException.compareAndSet(null, performOnStreamTasks(new StreamTaskAction() {
- @Override
- public String name() {
- return "suspend";
- }
-
- @Override
- public void apply(final StreamTask task) {
- try {
- task.suspend();
- } catch (final CommitFailedException e) {
- // commit failed during suspension. Just log it.
- log.warn("{} Failed to commit task {} state when suspending due to CommitFailedException", logPrefix, task.id);
- } catch (final Exception e) {
- log.error("{} Suspending task {} failed due to the following error:", logPrefix, task.id, e);
- try {
- task.close(false);
- } catch (final Exception f) {
- log.error("{} After suspending failed, closing the same task {} failed again due to the following error:", logPrefix, task.id, f);
- }
- throw e;
- }
- }
- }));
-
- for (final StandbyTask task : standbyTasks.values()) {
- try {
- try {
- task.suspend();
- } catch (final Exception e) {
- log.error("{} Suspending standby task {} failed due to the following error:", logPrefix, task.id, e);
- try {
- task.close(false);
- } catch (final Exception f) {
- log.error("{} After suspending failed, closing the same standby task {} failed again due to the following error:", logPrefix, task.id, f);
- }
- throw e;
- }
- } catch (final RuntimeException e) {
- firstException.compareAndSet(null, e);
- }
- }
-
- // remove the changelog partitions from restore consumer
- firstException.compareAndSet(null, unAssignChangeLogPartitions());
-
- updateSuspendedTasks();
-
- if (firstException.get() != null) {
- throw new StreamsException(logPrefix + " failed to suspend stream tasks", firstException.get());
- }
- }
private RuntimeException unAssignChangeLogPartitions() {
try {
@@ -1233,313 +1307,11 @@ public class StreamThread extends Thread {
return null;
}
- private List<AbstractTask> allTasks() {
- final List<AbstractTask> tasks = activeAndStandbytasks();
- tasks.addAll(suspendedAndSuspendedStandbytasks());
- return tasks;
- }
-
- private List<AbstractTask> activeAndStandbytasks() {
- final List<AbstractTask> tasks = new ArrayList<AbstractTask>(activeTasks.values());
- tasks.addAll(standbyTasks.values());
- return tasks;
- }
-
- private List<AbstractTask> suspendedAndSuspendedStandbytasks() {
- final List<AbstractTask> tasks = new ArrayList<AbstractTask>(suspendedTasks.values());
- tasks.addAll(suspendedStandbyTasks.values());
- return tasks;
- }
-
- private StreamTask findMatchingSuspendedTask(final TaskId taskId, final Set<TopicPartition> partitions) {
- if (suspendedTasks.containsKey(taskId)) {
- final StreamTask task = suspendedTasks.get(taskId);
- if (task.partitions().equals(partitions)) {
- return task;
- }
- }
- return null;
- }
-
- private StandbyTask findMatchingSuspendedStandbyTask(final TaskId taskId, final Set<TopicPartition> partitions) {
- if (suspendedStandbyTasks.containsKey(taskId)) {
- final StandbyTask task = suspendedStandbyTasks.get(taskId);
- if (task.partitions().equals(partitions)) {
- return task;
- }
- }
- return null;
- }
-
- private void closeNonAssignedSuspendedTasks() {
- final Map<TaskId, Set<TopicPartition>> newTaskAssignment = partitionAssignor.activeTasks();
- final Iterator<Map.Entry<TaskId, StreamTask>> suspendedTaskIterator = suspendedTasks.entrySet().iterator();
- while (suspendedTaskIterator.hasNext()) {
- final Map.Entry<TaskId, StreamTask> next = suspendedTaskIterator.next();
- final StreamTask task = next.getValue();
- final Set<TopicPartition> assignedPartitionsForTask = newTaskAssignment.get(next.getKey());
- if (!task.partitions().equals(assignedPartitionsForTask)) {
- log.debug("{} Closing suspended and not re-assigned task {}", logPrefix, task.id());
- try {
- task.closeSuspended(true, null);
- } catch (final Exception e) {
- log.error("{} Failed to close suspended task {} due to the following error:", logPrefix, next.getKey(), e);
- } finally {
- suspendedTaskIterator.remove();
- }
- }
- }
- }
-
- private void closeNonAssignedSuspendedStandbyTasks() {
- final Set<TaskId> currentSuspendedTaskIds = partitionAssignor.standbyTasks().keySet();
- final Iterator<Map.Entry<TaskId, StandbyTask>> standByTaskIterator = suspendedStandbyTasks.entrySet().iterator();
- while (standByTaskIterator.hasNext()) {
- final Map.Entry<TaskId, StandbyTask> suspendedTask = standByTaskIterator.next();
- if (!currentSuspendedTaskIds.contains(suspendedTask.getKey())) {
- final StandbyTask task = suspendedTask.getValue();
- log.debug("{} Closing suspended and not re-assigned standby task {}", logPrefix, task.id());
- try {
- task.close(true);
- } catch (final Exception e) {
- log.error("{} Failed to remove suspended standby task {} due to the following error:", logPrefix, task.id(), e);
- } finally {
- standByTaskIterator.remove();
- }
- }
- }
- }
-
- // visible for testing
- protected StreamTask createStreamTask(final TaskId id, final Collection<TopicPartition> partitions) {
- streamsMetrics.taskCreatedSensor.record();
-
- try {
- return new StreamTask(
- id,
- applicationId,
- partitions,
- builder.build(id.topicGroupId),
- consumer,
- storeChangelogReader,
- config,
- streamsMetrics,
- stateDirectory,
- cache,
- time,
- createProducer(id));
- } finally {
- log.trace("{} Created active task {} with assigned partitions {}", logPrefix, id, partitions);
- }
- }
-
- private Producer<byte[], byte[]> createProducer(final TaskId id) {
-
- final Producer<byte[], byte[]> producer;
- if (eosEnabled) {
- final Map<String, Object> producerConfigs = config.getProducerConfigs(threadClientId + "-" + id);
- log.info("{} Creating producer client for task {}", logPrefix, id);
- producerConfigs.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, applicationId + "-" + id);
- producer = clientSupplier.getProducer(producerConfigs);
- } else {
- if (threadProducer == null) {
- final Map<String, Object> producerConfigs = config.getProducerConfigs(threadClientId);
- log.info("{} Creating shared producer client", logPrefix);
- threadProducer = clientSupplier.getProducer(producerConfigs);
- }
- producer = threadProducer;
- }
-
- return producer;
- }
-
- private void addStreamTasks(final Collection<TopicPartition> assignment, final long start) {
- if (partitionAssignor == null) {
- throw new IllegalStateException(logPrefix + " Partition assignor has not been initialized while adding stream tasks: this should not happen.");
- }
-
- final Map<TaskId, Set<TopicPartition>> newTasks = new HashMap<>();
-
- // collect newly assigned tasks and reopen re-assigned tasks
- log.debug("{} Adding assigned tasks as active: {}", logPrefix, partitionAssignor.activeTasks());
- for (final Map.Entry<TaskId, Set<TopicPartition>> entry : partitionAssignor.activeTasks().entrySet()) {
- final TaskId taskId = entry.getKey();
- final Set<TopicPartition> partitions = entry.getValue();
-
- if (assignment.containsAll(partitions)) {
- try {
- final StreamTask task = findMatchingSuspendedTask(taskId, partitions);
- if (task != null) {
- suspendedTasks.remove(taskId);
- task.resume();
-
- activeTasks.put(taskId, task);
-
- for (final TopicPartition partition : partitions) {
- activeTasksByPartition.put(partition, task);
- }
- } else {
- newTasks.put(taskId, partitions);
- }
- } catch (final StreamsException e) {
- log.error("{} Failed to create an active task {} due to the following error:", logPrefix, taskId, e);
- throw e;
- }
- } else {
- log.warn("{} Task {} owned partitions {} are not contained in the assignment {}", logPrefix, taskId, partitions, assignment);
- }
- }
-
- // create all newly assigned tasks (guard against race condition with other thread via backoff and retry)
- // -> other thread will call removeSuspendedTasks(); eventually
- log.trace("{} New active tasks to be created: {}", logPrefix, newTasks);
-
- taskCreator.retryWithBackoff(newTasks, start);
- }
-
- // visible for testing
- protected StandbyTask createStandbyTask(final TaskId id, final Collection<TopicPartition> partitions) {
- streamsMetrics.taskCreatedSensor.record();
-
- final ProcessorTopology topology = builder.build(id.topicGroupId);
-
- if (!topology.stateStores().isEmpty()) {
- try {
- return new StandbyTask(id, applicationId, partitions, topology, consumer, storeChangelogReader, config, streamsMetrics, stateDirectory);
- } finally {
- log.trace("{} Created standby task {} with assigned partitions {}", logPrefix, id, partitions);
- }
- } else {
- log.trace("{} Skipped standby task {} with assigned partitions {} since it does not have any state stores to materialize", logPrefix, id, partitions);
-
- return null;
- }
- }
-
- private void addStandbyTasks(final long start) {
- if (partitionAssignor == null) {
- throw new IllegalStateException(logPrefix + " Partition assignor has not been initialized while adding standby tasks: this should not happen.");
- }
-
- final Map<TopicPartition, Long> checkpointedOffsets = new HashMap<>();
-
- final Map<TaskId, Set<TopicPartition>> newStandbyTasks = new HashMap<>();
-
- log.debug("{} Adding assigned standby tasks {}", logPrefix, partitionAssignor.standbyTasks());
- // collect newly assigned standby tasks and reopen re-assigned standby tasks
- for (final Map.Entry<TaskId, Set<TopicPartition>> entry : partitionAssignor.standbyTasks().entrySet()) {
- final TaskId taskId = entry.getKey();
- final Set<TopicPartition> partitions = entry.getValue();
- final StandbyTask task = findMatchingSuspendedStandbyTask(taskId, partitions);
-
- if (task != null) {
- suspendedStandbyTasks.remove(taskId);
- task.resume();
- } else {
- newStandbyTasks.put(taskId, partitions);
- }
-
- updateStandByTaskMaps(checkpointedOffsets, taskId, partitions, task);
- }
-
- // create all newly assigned standby tasks (guard against race condition with other thread via backoff and retry)
- // -> other thread will call removeSuspendedStandbyTasks(); eventually
- log.trace("{} New standby tasks to be created: {}", logPrefix, newStandbyTasks);
-
- new StandbyTaskCreator(checkpointedOffsets).retryWithBackoff(newStandbyTasks, start);
-
- restoreConsumer.assign(new ArrayList<>(checkpointedOffsets.keySet()));
-
- for (final Map.Entry<TopicPartition, Long> entry : checkpointedOffsets.entrySet()) {
- final TopicPartition partition = entry.getKey();
- final long offset = entry.getValue();
- if (offset >= 0) {
- restoreConsumer.seek(partition, offset);
- } else {
- restoreConsumer.seekToBeginning(singleton(partition));
- }
- }
- }
-
- private void updateStandByTaskMaps(final Map<TopicPartition, Long> checkpointedOffsets,
- final TaskId taskId,
- final Set<TopicPartition> partitions,
- final StandbyTask task) {
- if (task != null) {
- standbyTasks.put(taskId, task);
- for (final TopicPartition partition : partitions) {
- standbyTasksByPartition.put(partition, task);
- }
- // collect checked pointed offsets to position the restore consumer
- // this include all partitions from which we restore states
- for (final TopicPartition partition : task.checkpointedOffsets().keySet()) {
- standbyTasksByPartition.put(partition, task);
- }
- checkpointedOffsets.putAll(task.checkpointedOffsets());
- }
- }
-
- private void updateSuspendedTasks() {
- suspendedTasks.clear();
- suspendedTasks.putAll(activeTasks);
- suspendedStandbyTasks.putAll(standbyTasks);
- }
-
- private void removeStreamTasks() {
- log.debug("{} Removing all active tasks {}", logPrefix, activeTasks.keySet());
-
- try {
- prevActiveTasks.clear();
- prevActiveTasks.addAll(activeTasks.keySet());
-
- activeTasks.clear();
- activeTasksByPartition.clear();
- } catch (final Exception e) {
- log.error("{} Failed to remove stream tasks due to the following error:", logPrefix, e);
- }
- }
-
- private void removeStandbyTasks() {
- log.debug("{} Removing all standby tasks {}", logPrefix, standbyTasks.keySet());
-
- standbyTasks.clear();
- standbyTasksByPartition.clear();
+ private void clearStandbyRecords() {
standbyRecords.clear();
}
- private void closeZombieTask(final StreamTask task) {
- log.warn("{} Producer of task {} fenced; closing zombie task", logPrefix, task.id);
- try {
- task.close(false);
- } catch (final Exception e) {
- log.warn("{} Failed to close zombie task due to {}, ignore and proceed", logPrefix, e);
- }
- activeTasks.remove(task.id);
- }
-
-
- private RuntimeException performOnStreamTasks(final StreamTaskAction action) {
- RuntimeException firstException = null;
- final Iterator<Map.Entry<TaskId, StreamTask>> it = activeTasks.entrySet().iterator();
- while (it.hasNext()) {
- final StreamTask task = it.next().getValue();
- try {
- action.apply(task);
- } catch (final ProducerFencedException e) {
- closeZombieTask(task);
- it.remove();
- } catch (final RuntimeException t) {
- log.error("{} Failed to {} stream task {} due to the following error:",
- logPrefix,
- action.name(),
- task.id(),
- t);
- if (firstException == null) {
- firstException = t;
- }
- }
- }
-
- return firstException;
+ private void refreshMetadataState() {
+ streamsMetadataState.onChange(metadataProvider.getPartitionsByHostState(), metadataProvider.clusterMetadata());
}
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/3e69ce80/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsMetadataState.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsMetadataState.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsMetadataState.java
index 3fd1613..f470b74 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsMetadataState.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsMetadataState.java
@@ -195,7 +195,7 @@ public class StreamsMetadataState {
* @param currentState the current mapping of {@link HostInfo} -> {@link TopicPartition}s
* @param clusterMetadata the current clusterMetadata {@link Cluster}
*/
- public synchronized void onChange(final Map<HostInfo, Set<TopicPartition>> currentState, final Cluster clusterMetadata) {
+ synchronized void onChange(final Map<HostInfo, Set<TopicPartition>> currentState, final Cluster clusterMetadata) {
this.clusterMetadata = clusterMetadata;
rebuildMetadata(currentState);
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/3e69ce80/streams/src/main/java/org/apache/kafka/streams/processor/internals/Task.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/Task.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/Task.java
new file mode 100644
index 0000000..177306a
--- /dev/null
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/Task.java
@@ -0,0 +1,67 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.streams.processor.internals;
+
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.streams.processor.ProcessorContext;
+import org.apache.kafka.streams.processor.StateStore;
+import org.apache.kafka.streams.processor.TaskId;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+public interface Task {
+ void resume();
+
+ void commit();
+
+ void suspend();
+
+ void close(boolean clean);
+
+ TaskId id();
+
+ String applicationId();
+
+ Set<TopicPartition> partitions();
+
+ ProcessorTopology topology();
+
+ ProcessorContext context();
+
+ StateStore getStore(String name);
+
+ void closeSuspended(boolean clean, RuntimeException e);
+
+ Map<TopicPartition, Long> checkpointedOffsets();
+
+ boolean process();
+
+ boolean commitNeeded();
+
+ boolean maybePunctuateStreamTime();
+
+ boolean maybePunctuateSystemTime();
+
+ List<ConsumerRecord<byte[], byte[]>> update(TopicPartition partition, List<ConsumerRecord<byte[], byte[]>> remaining);
+
+ String toString(String indent);
+
+ int addRecords(TopicPartition partition, final Iterable<ConsumerRecord<byte[], byte[]>> records);
+}
[3/5] kafka git commit: KAFKA-5702; extract refactor StreamThread
Posted by da...@apache.org.
http://git-wip-us.apache.org/repos/asf/kafka/blob/3e69ce80/streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java
new file mode 100644
index 0000000..4bdad9a
--- /dev/null
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java
@@ -0,0 +1,524 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.streams.processor.internals;
+
+import org.apache.kafka.clients.consumer.CommitFailedException;
+import org.apache.kafka.clients.consumer.Consumer;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.errors.ProducerFencedException;
+import org.apache.kafka.common.utils.Time;
+import org.apache.kafka.streams.errors.StreamsException;
+import org.apache.kafka.streams.processor.TaskId;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.TreeSet;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static java.util.Collections.singleton;
+
+class TaskManager {
+ // initialize the task list
+ // activeTasks needs to be concurrent as it can be accessed
+ // by QueryableState
+ private static final Logger log = LoggerFactory.getLogger(TaskManager.class);
+ private final Map<TaskId, Task> activeTasks = new ConcurrentHashMap<>();
+ private final Map<TaskId, Task> standbyTasks = new HashMap<>();
+ private final Map<TopicPartition, Task> activeTasksByPartition = new HashMap<>();
+ private final Map<TopicPartition, Task> standbyTasksByPartition = new HashMap<>();
+ private final Set<TaskId> prevActiveTasks = new TreeSet<>();
+ private final Map<TaskId, Task> suspendedTasks = new HashMap<>();
+ private final Map<TaskId, Task> suspendedStandbyTasks = new HashMap<>();
+ private final ChangelogReader changelogReader;
+ private final Time time;
+ private final String logPrefix;
+ private final Consumer<byte[], byte[]> restoreConsumer;
+ private final StreamThread.AbstractTaskCreator taskCreator;
+ private final StreamThread.AbstractTaskCreator standbyTaskCreator;
+ private ThreadMetadataProvider threadMetadataProvider;
+ private Consumer<byte[], byte[]> consumer;
+
+ TaskManager(final ChangelogReader changelogReader,
+ final Time time,
+ final String logPrefix,
+ final Consumer<byte[], byte[]> restoreConsumer,
+ final StreamThread.AbstractTaskCreator taskCreator,
+ final StreamThread.AbstractTaskCreator standbyTaskCreator) {
+ this.changelogReader = changelogReader;
+ this.time = time;
+ this.logPrefix = logPrefix;
+ this.restoreConsumer = restoreConsumer;
+ this.taskCreator = taskCreator;
+ this.standbyTaskCreator = standbyTaskCreator;
+ }
+
+ void createTasks(final Collection<TopicPartition> assignment) {
+ if (threadMetadataProvider == null) {
+ throw new IllegalStateException(logPrefix + " taskIdProvider has not been initialized while adding stream tasks. This should not happen.");
+ }
+ if (consumer == null) {
+ throw new IllegalStateException(logPrefix + " consumer has not been initialized while adding stream tasks. This should not happen.");
+ }
+
+ final long start = time.milliseconds();
+ changelogReader.clear();
+ // do this first as we may have suspended standby tasks that
+ // will become active or vice versa
+ closeNonAssignedSuspendedStandbyTasks();
+ Map<TaskId, Set<TopicPartition>> assignedActiveTasks = threadMetadataProvider.activeTasks();
+ closeNonAssignedSuspendedTasks(assignedActiveTasks);
+ addStreamTasks(assignment, assignedActiveTasks, start);
+ changelogReader.restore();
+ addStandbyTasks(start);
+ }
+
+ void setThreadMetadataProvider(final ThreadMetadataProvider threadMetadataProvider) {
+ this.threadMetadataProvider = threadMetadataProvider;
+ }
+
+ private void closeNonAssignedSuspendedStandbyTasks() {
+ final Set<TaskId> currentSuspendedTaskIds = threadMetadataProvider.standbyTasks().keySet();
+ final Iterator<Map.Entry<TaskId, Task>> standByTaskIterator = suspendedStandbyTasks.entrySet().iterator();
+ while (standByTaskIterator.hasNext()) {
+ final Map.Entry<TaskId, Task> suspendedTask = standByTaskIterator.next();
+ if (!currentSuspendedTaskIds.contains(suspendedTask.getKey())) {
+ final Task task = suspendedTask.getValue();
+ log.debug("{} Closing suspended and not re-assigned standby task {}", logPrefix, task.id());
+ try {
+ task.close(true);
+ } catch (final Exception e) {
+ log.error("{} Failed to remove suspended standby task {} due to the following error:", logPrefix, task.id(), e);
+ } finally {
+ standByTaskIterator.remove();
+ }
+ }
+ }
+ }
+
+ private void closeNonAssignedSuspendedTasks(final Map<TaskId, Set<TopicPartition>> newTaskAssignment) {
+ final Iterator<Map.Entry<TaskId, Task>> suspendedTaskIterator = suspendedTasks.entrySet().iterator();
+ while (suspendedTaskIterator.hasNext()) {
+ final Map.Entry<TaskId, Task> next = suspendedTaskIterator.next();
+ final Task task = next.getValue();
+ final Set<TopicPartition> assignedPartitionsForTask = newTaskAssignment.get(next.getKey());
+ if (!task.partitions().equals(assignedPartitionsForTask)) {
+ log.debug("{} Closing suspended and not re-assigned task {}", logPrefix, task.id());
+ try {
+ task.closeSuspended(true, null);
+ } catch (final Exception e) {
+ log.error("{} Failed to close suspended task {} due to the following error:", logPrefix, next.getKey(), e);
+ } finally {
+ suspendedTaskIterator.remove();
+ }
+ }
+ }
+ }
+
+ private void addStreamTasks(final Collection<TopicPartition> assignment, final Map<TaskId, Set<TopicPartition>> assignedTasks, final long start) {
+ final Map<TaskId, Set<TopicPartition>> newTasks = new HashMap<>();
+
+ // collect newly assigned tasks and reopen re-assigned tasks
+ log.debug("{} Adding assigned tasks as active: {}", logPrefix, assignedTasks);
+ for (final Map.Entry<TaskId, Set<TopicPartition>> entry : assignedTasks.entrySet()) {
+ final TaskId taskId = entry.getKey();
+ final Set<TopicPartition> partitions = entry.getValue();
+
+ if (assignment.containsAll(partitions)) {
+ try {
+ final Task task = findMatchingSuspendedTask(taskId, partitions);
+ if (task != null) {
+ suspendedTasks.remove(taskId);
+ task.resume();
+
+ activeTasks.put(taskId, task);
+
+ for (final TopicPartition partition : partitions) {
+ activeTasksByPartition.put(partition, task);
+ }
+ } else {
+ newTasks.put(taskId, partitions);
+ }
+ } catch (final StreamsException e) {
+ log.error("{} Failed to create an active task {} due to the following error:", logPrefix, taskId, e);
+ throw e;
+ }
+ } else {
+ log.warn("{} Task {} owned partitions {} are not contained in the assignment {}", logPrefix, taskId, partitions, assignment);
+ }
+ }
+
+ // create all newly assigned tasks (guard against race condition with other thread via backoff and retry)
+ // -> other thread will call removeSuspendedTasks(); eventually
+ log.trace("{} New active tasks to be created: {}", logPrefix, newTasks);
+
+ if (!newTasks.isEmpty()) {
+ final Map<Task, Set<TopicPartition>> createdTasks = taskCreator.retryWithBackoff(consumer, newTasks, start);
+ for (final Map.Entry<Task, Set<TopicPartition>> entry : createdTasks.entrySet()) {
+ final Task task = entry.getKey();
+ activeTasks.put(task.id(), task);
+ for (final TopicPartition partition : entry.getValue()) {
+ activeTasksByPartition.put(partition, task);
+ }
+ }
+ }
+ }
+
+ private void addStandbyTasks(final long start) {
+ final Map<TopicPartition, Long> checkpointedOffsets = new HashMap<>();
+
+ final Map<TaskId, Set<TopicPartition>> newStandbyTasks = new HashMap<>();
+
+ Map<TaskId, Set<TopicPartition>> assignedStandbyTasks = threadMetadataProvider.standbyTasks();
+ log.debug("{} Adding assigned standby tasks {}", logPrefix, assignedStandbyTasks);
+ // collect newly assigned standby tasks and reopen re-assigned standby tasks
+ for (final Map.Entry<TaskId, Set<TopicPartition>> entry : assignedStandbyTasks.entrySet()) {
+ final TaskId taskId = entry.getKey();
+ final Set<TopicPartition> partitions = entry.getValue();
+ final Task task = findMatchingSuspendedStandbyTask(taskId, partitions);
+
+ if (task != null) {
+ suspendedStandbyTasks.remove(taskId);
+ task.resume();
+ } else {
+ newStandbyTasks.put(taskId, partitions);
+ }
+
+ updateStandByTasks(checkpointedOffsets, taskId, partitions, task);
+ }
+
+ // create all newly assigned standby tasks (guard against race condition with other thread via backoff and retry)
+ // -> other thread will call removeSuspendedStandbyTasks(); eventually
+ log.trace("{} New standby tasks to be created: {}", logPrefix, newStandbyTasks);
+ if (!newStandbyTasks.isEmpty()) {
+ final Map<Task, Set<TopicPartition>> createdStandbyTasks = standbyTaskCreator.retryWithBackoff(consumer, newStandbyTasks, start);
+ for (Map.Entry<Task, Set<TopicPartition>> entry : createdStandbyTasks.entrySet()) {
+ final Task task = entry.getKey();
+ updateStandByTasks(checkpointedOffsets, task.id(), entry.getValue(), task);
+ }
+ }
+
+ restoreConsumer.assign(checkpointedOffsets.keySet());
+
+ for (final Map.Entry<TopicPartition, Long> entry : checkpointedOffsets.entrySet()) {
+ final TopicPartition partition = entry.getKey();
+ final long offset = entry.getValue();
+ if (offset >= 0) {
+ restoreConsumer.seek(partition, offset);
+ } else {
+ restoreConsumer.seekToBeginning(singleton(partition));
+ }
+ }
+ }
+
+ private void updateStandByTasks(final Map<TopicPartition, Long> checkpointedOffsets,
+ final TaskId taskId,
+ final Set<TopicPartition> partitions,
+ final Task task) {
+ if (task != null) {
+ standbyTasks.put(taskId, task);
+ for (final TopicPartition partition : partitions) {
+ standbyTasksByPartition.put(partition, task);
+ }
+ // collect checked pointed offsets to position the restore consumer
+ // this include all partitions from which we restore states
+ for (final TopicPartition partition : task.checkpointedOffsets().keySet()) {
+ standbyTasksByPartition.put(partition, task);
+ }
+ checkpointedOffsets.putAll(task.checkpointedOffsets());
+ }
+ }
+
+ List<Task> allTasks() {
+ final List<Task> tasks = activeAndStandbytasks();
+ tasks.addAll(suspendedAndSuspendedStandbytasks());
+ return tasks;
+ }
+
+ private List<Task> activeAndStandbytasks() {
+ final List<Task> tasks = new ArrayList<>(activeTasks.values());
+ tasks.addAll(standbyTasks.values());
+ return tasks;
+ }
+
+ private List<Task> suspendedAndSuspendedStandbytasks() {
+ final List<Task> tasks = new ArrayList<>(suspendedTasks.values());
+ tasks.addAll(suspendedStandbyTasks.values());
+ return tasks;
+ }
+
+ private Task findMatchingSuspendedTask(final TaskId taskId, final Set<TopicPartition> partitions) {
+ if (suspendedTasks.containsKey(taskId)) {
+ final Task task = suspendedTasks.get(taskId);
+ if (task.partitions().equals(partitions)) {
+ return task;
+ }
+ }
+ return null;
+ }
+
+ private Task findMatchingSuspendedStandbyTask(final TaskId taskId, final Set<TopicPartition> partitions) {
+ if (suspendedStandbyTasks.containsKey(taskId)) {
+ final Task task = suspendedStandbyTasks.get(taskId);
+ if (task.partitions().equals(partitions)) {
+ return task;
+ }
+ }
+ return null;
+ }
+
+ Set<TaskId> activeTaskIds() {
+ return Collections.unmodifiableSet(activeTasks.keySet());
+ }
+
+ Set<TaskId> standbyTaskIds() {
+ return Collections.unmodifiableSet(standbyTasks.keySet());
+ }
+
+ Set<TaskId> prevActiveTaskIds() {
+ return Collections.unmodifiableSet(prevActiveTasks);
+ }
+
+ /**
+ * Similar to shutdownTasksAndState, however does not close the task managers, in the hope that
+ * soon the tasks will be assigned again
+ */
+ void suspendTasksAndState() {
+ log.debug("{} Suspending all active tasks {} and standby tasks {}",
+ logPrefix, activeTasks.keySet(), standbyTasks.keySet());
+
+ final AtomicReference<RuntimeException> firstException = new AtomicReference<>(null);
+
+ firstException.compareAndSet(null, performOnActiveTasks(new TaskAction() {
+ @Override
+ public String name() {
+ return "suspend";
+ }
+
+ @Override
+ public void apply(final Task task) {
+ try {
+ task.suspend();
+ } catch (final CommitFailedException e) {
+ // commit failed during suspension. Just log it.
+ log.warn("{} Failed to commit task {} state when suspending due to CommitFailedException", logPrefix, task.id());
+ } catch (final Exception e) {
+ log.error("{} Suspending task {} failed due to the following error:", logPrefix, task.id(), e);
+ try {
+ task.close(false);
+ } catch (final Exception f) {
+ log.error("{} After suspending failed, closing the same task {} failed again due to the following error:", logPrefix, task.id(), f);
+ }
+ throw e;
+ }
+ }
+ }));
+
+ for (final Task task : standbyTasks.values()) {
+ try {
+ try {
+ task.suspend();
+ } catch (final Exception e) {
+ log.error("{} Suspending standby task {} failed due to the following error:", logPrefix, task.id(), e);
+ try {
+ task.close(false);
+ } catch (final Exception f) {
+ log.error("{} After suspending failed, closing the same standby task {} failed again due to the following error:", logPrefix, task.id(), f);
+ }
+ throw e;
+ }
+ } catch (final RuntimeException e) {
+ firstException.compareAndSet(null, e);
+ }
+ }
+
+ // remove the changelog partitions from restore consumer
+ firstException.compareAndSet(null, unAssignChangeLogPartitions());
+
+ updateSuspendedTasks();
+
+ if (firstException.get() != null) {
+ throw new StreamsException(logPrefix + " failed to suspend stream tasks", firstException.get());
+ }
+ }
+
+ private RuntimeException unAssignChangeLogPartitions() {
+ try {
+ // un-assign the change log partitions
+ restoreConsumer.assign(Collections.<TopicPartition>emptyList());
+ } catch (final RuntimeException e) {
+ log.error("{} Failed to un-assign change log partitions due to the following error:", logPrefix, e);
+ return e;
+ }
+ return null;
+ }
+
+ private void updateSuspendedTasks() {
+ suspendedTasks.clear();
+ suspendedTasks.putAll(activeTasks);
+ suspendedStandbyTasks.putAll(standbyTasks);
+ }
+
+ private void removeStreamTasks() {
+ log.debug("{} Removing all active tasks {}", logPrefix, activeTasks.keySet());
+
+ try {
+ prevActiveTasks.clear();
+ prevActiveTasks.addAll(activeTasks.keySet());
+
+ activeTasks.clear();
+ activeTasksByPartition.clear();
+ } catch (final Exception e) {
+ log.error("{} Failed to remove stream tasks due to the following error:", logPrefix, e);
+ }
+ }
+
+ void closeZombieTask(final Task task) {
+ log.warn("{} Producer of task {} fenced; closing zombie task", logPrefix, task.id());
+ try {
+ task.close(false);
+ } catch (final Exception e) {
+ log.warn("{} Failed to close zombie task due to {}, ignore and proceed", logPrefix, e);
+ }
+ activeTasks.remove(task.id());
+ }
+
+
+ RuntimeException performOnActiveTasks(final TaskAction action) {
+ return performOnTasks(action, activeTasks, "stream task");
+ }
+
+ RuntimeException performOnStandbyTasks(final TaskAction action) {
+ return performOnTasks(action, standbyTasks, "standby task");
+ }
+
+ private RuntimeException performOnTasks(final TaskAction action, final Map<TaskId, Task> tasks, final String taskType) {
+ RuntimeException firstException = null;
+ final Iterator<Map.Entry<TaskId, Task>> it = tasks.entrySet().iterator();
+ while (it.hasNext()) {
+ final Task task = it.next().getValue();
+ try {
+ action.apply(task);
+ } catch (final ProducerFencedException e) {
+ closeZombieTask(task);
+ it.remove();
+ } catch (final RuntimeException t) {
+ log.error("{} Failed to {} " + taskType + " {} due to the following error:",
+ logPrefix,
+ action.name(),
+ task.id(),
+ t);
+ if (firstException == null) {
+ firstException = t;
+ }
+ }
+ }
+
+ return firstException;
+ }
+
+
+
+ void shutdown(final boolean clean) {
+ log.debug("{} Shutting down all active tasks {}, standby tasks {}, suspended tasks {}, and suspended standby tasks {}",
+ logPrefix, activeTasks.keySet(), standbyTasks.keySet(),
+ suspendedTasks.keySet(), suspendedStandbyTasks.keySet());
+
+ for (final Task task : allTasks()) {
+ try {
+ task.close(clean);
+ } catch (final RuntimeException e) {
+ log.error("{} Failed while closing {} {} due to the following error:",
+ logPrefix,
+ task.getClass().getSimpleName(),
+ task.id(),
+ e);
+ }
+ }
+ try {
+ threadMetadataProvider.close();
+ } catch (final Throwable e) {
+ log.error("{} Failed to close KafkaStreamClient due to the following error:", logPrefix, e);
+ }
+ // remove the changelog partitions from restore consumer
+ unAssignChangeLogPartitions();
+
+ }
+
+ Set<TaskId> suspendedActiveTaskIds() {
+ return Collections.unmodifiableSet(suspendedTasks.keySet());
+ }
+
+ Set<TaskId> suspendedStandbyTaskIds() {
+ return Collections.unmodifiableSet(suspendedStandbyTasks.keySet());
+ }
+
+ void removeTasks() {
+ removeStreamTasks();
+ removeStandbyTasks();
+ }
+
+ private void removeStandbyTasks() {
+ log.debug("{} Removing all standby tasks {}", logPrefix, standbyTasks.keySet());
+ standbyTasks.clear();
+ standbyTasksByPartition.clear();
+ }
+
+ Task activeTask(final TopicPartition partition) {
+ return activeTasksByPartition.get(partition);
+ }
+
+ boolean hasStandbyTasks() {
+ return !standbyTasks.isEmpty();
+ }
+
+ Task standbyTask(final TopicPartition partition) {
+ return standbyTasksByPartition.get(partition);
+ }
+
+ public Map<TaskId, Task> activeTasks() {
+ return activeTasks;
+ }
+
+ boolean hasActiveTasks() {
+ return !activeTasks.isEmpty();
+ }
+
+ void setConsumer(final Consumer<byte[], byte[]> consumer) {
+ this.consumer = consumer;
+ }
+
+ public void closeProducer() {
+ taskCreator.close();
+ }
+
+
+
+
+ interface TaskAction {
+ String name();
+ void apply(final Task task);
+ }
+}
http://git-wip-us.apache.org/repos/asf/kafka/blob/3e69ce80/streams/src/main/java/org/apache/kafka/streams/processor/internals/ThreadDataProvider.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/ThreadDataProvider.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/ThreadDataProvider.java
new file mode 100644
index 0000000..ded98f7
--- /dev/null
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/ThreadDataProvider.java
@@ -0,0 +1,36 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.streams.processor.internals;
+
+import org.apache.kafka.streams.StreamsConfig;
+import org.apache.kafka.streams.processor.PartitionGrouper;
+import org.apache.kafka.streams.processor.TaskId;
+
+import java.util.Set;
+import java.util.UUID;
+
+// interface to get info about the StreamThread
+interface ThreadDataProvider {
+ InternalTopologyBuilder builder();
+ String name();
+ Set<TaskId> prevActiveTasks();
+ Set<TaskId> cachedTasks();
+ UUID processId();
+ StreamsConfig config();
+ PartitionGrouper partitionGrouper();
+ void setThreadMetadataProvider(final ThreadMetadataProvider provider);
+}
http://git-wip-us.apache.org/repos/asf/kafka/blob/3e69ce80/streams/src/main/java/org/apache/kafka/streams/processor/internals/ThreadMetadataProvider.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/ThreadMetadataProvider.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/ThreadMetadataProvider.java
new file mode 100644
index 0000000..f185045
--- /dev/null
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/ThreadMetadataProvider.java
@@ -0,0 +1,36 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.streams.processor.internals;
+
+import org.apache.kafka.common.Cluster;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.streams.processor.TaskId;
+import org.apache.kafka.streams.state.HostInfo;
+
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * Interface used by a <code>StreamThread</code> to get metadata from the <code>StreamPartitionAssignor</code>
+ */
+public interface ThreadMetadataProvider {
+ Map<TaskId, Set<TopicPartition>> standbyTasks();
+ Map<TaskId, Set<TopicPartition>> activeTasks();
+ Map<HostInfo, Set<TopicPartition>> getPartitionsByHostState();
+ Cluster clusterMetadata();
+ void close();
+}
http://git-wip-us.apache.org/repos/asf/kafka/blob/3e69ce80/streams/src/main/java/org/apache/kafka/streams/state/HostInfo.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/HostInfo.java b/streams/src/main/java/org/apache/kafka/streams/state/HostInfo.java
index 6f48543..8fcdd03 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/HostInfo.java
+++ b/streams/src/main/java/org/apache/kafka/streams/state/HostInfo.java
@@ -19,6 +19,7 @@ package org.apache.kafka.streams.state;
import org.apache.kafka.common.serialization.Serializer;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.processor.StreamPartitioner;
+import org.apache.kafka.streams.processor.internals.StreamPartitionAssignor;
/**
* Represents a user defined endpoint in a {@link org.apache.kafka.streams.KafkaStreams} application.
@@ -29,7 +30,7 @@ import org.apache.kafka.streams.processor.StreamPartitioner;
* {@link KafkaStreams#metadataForKey(String, Object, Serializer)}
*
* The HostInfo is constructed during Partition Assignment
- * see {@link org.apache.kafka.streams.processor.internals.StreamPartitionAssignor}
+ * see {@link StreamPartitionAssignor}
* It is extracted from the config {@link org.apache.kafka.streams.StreamsConfig#APPLICATION_SERVER_CONFIG}
*
* If developers wish to expose an endpoint in their KafkaStreams applications they should provide the above
http://git-wip-us.apache.org/repos/asf/kafka/blob/3e69ce80/streams/src/main/java/org/apache/kafka/streams/state/internals/StreamThreadStateStoreProvider.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/StreamThreadStateStoreProvider.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/StreamThreadStateStoreProvider.java
index 45d9898..19a898e 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/internals/StreamThreadStateStoreProvider.java
+++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/StreamThreadStateStoreProvider.java
@@ -18,8 +18,8 @@ package org.apache.kafka.streams.state.internals;
import org.apache.kafka.streams.errors.InvalidStateStoreException;
import org.apache.kafka.streams.processor.StateStore;
-import org.apache.kafka.streams.processor.internals.StreamTask;
import org.apache.kafka.streams.processor.internals.StreamThread;
+import org.apache.kafka.streams.processor.internals.Task;
import org.apache.kafka.streams.state.QueryableStoreType;
import java.util.ArrayList;
@@ -48,7 +48,7 @@ public class StreamThreadStateStoreProvider implements StateStoreProvider {
throw new InvalidStateStoreException("the state store, " + storeName + ", may have migrated to another instance.");
}
final List<T> stores = new ArrayList<>();
- for (StreamTask streamTask : streamThread.tasks().values()) {
+ for (Task streamTask : streamThread.tasks().values()) {
final StateStore store = streamTask.getStore(storeName);
if (store != null && queryableStoreType.accepts(store)) {
if (!store.isOpen()) {
http://git-wip-us.apache.org/repos/asf/kafka/blob/3e69ce80/streams/src/test/java/org/apache/kafka/streams/integration/RegexSourceIntegrationTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/integration/RegexSourceIntegrationTest.java b/streams/src/test/java/org/apache/kafka/streams/integration/RegexSourceIntegrationTest.java
index b5b6e4f..d67588a 100644
--- a/streams/src/test/java/org/apache/kafka/streams/integration/RegexSourceIntegrationTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/integration/RegexSourceIntegrationTest.java
@@ -17,31 +17,25 @@
package org.apache.kafka.streams.integration;
import kafka.utils.MockTime;
+import org.apache.kafka.clients.consumer.Consumer;
+import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;
-import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.serialization.ByteArrayDeserializer;
import org.apache.kafka.common.serialization.Serde;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
-import org.apache.kafka.common.utils.Time;
-import org.apache.kafka.streams.KafkaClientSupplier;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
-import org.apache.kafka.streams.StreamsBuilderTest;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.integration.utils.EmbeddedKafkaCluster;
import org.apache.kafka.streams.integration.utils.IntegrationTestUtils;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.processor.ProcessorSupplier;
-import org.apache.kafka.streams.processor.TaskId;
import org.apache.kafka.streams.processor.TopologyBuilder;
import org.apache.kafka.streams.processor.internals.DefaultKafkaClientSupplier;
-import org.apache.kafka.streams.processor.internals.InternalTopologyBuilder;
-import org.apache.kafka.streams.processor.internals.StateDirectory;
-import org.apache.kafka.streams.processor.internals.StreamTask;
-import org.apache.kafka.streams.processor.internals.StreamThread;
-import org.apache.kafka.streams.processor.internals.StreamsMetadataState;
import org.apache.kafka.test.IntegrationTest;
import org.apache.kafka.test.MockProcessorSupplier;
import org.apache.kafka.test.MockStateStoreSupplier;
@@ -55,7 +49,6 @@ import org.junit.ClassRule;
import org.junit.Test;
import org.junit.experimental.categories.Category;
-import java.lang.reflect.Field;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
@@ -63,7 +56,6 @@ import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Properties;
-import java.util.UUID;
import java.util.regex.Pattern;
import static org.hamcrest.CoreMatchers.equalTo;
@@ -96,6 +88,7 @@ public class RegexSourceIntegrationTest {
private static final String STRING_SERDE_CLASSNAME = Serdes.String().getClass().getName();
private Properties streamsConfiguration;
private static final String STREAM_TASKS_NOT_UPDATED = "Stream tasks not updated";
+ private KafkaStreams streams;
@BeforeClass
@@ -127,6 +120,9 @@ public class RegexSourceIntegrationTest {
@After
public void tearDown() throws Exception {
+ if (streams != null) {
+ streams.close();
+ }
// Remove any state from previous test runs
IntegrationTestUtils.purgeLocalStreamsState(streamsConfiguration);
}
@@ -147,48 +143,38 @@ public class RegexSourceIntegrationTest {
final KStream<String, String> pattern1Stream = builder.stream(Pattern.compile("TEST-TOPIC-\\d"));
pattern1Stream.to(stringSerde, stringSerde, DEFAULT_OUTPUT_TOPIC);
+ final List<String> assignedTopics = new ArrayList<>();
+ streams = new KafkaStreams(builder.build(), streamsConfig, new DefaultKafkaClientSupplier() {
+ @Override
+ public Consumer<byte[], byte[]> getConsumer(final Map<String, Object> config) {
+ return new KafkaConsumer<byte[], byte[]>(config, new ByteArrayDeserializer(), new ByteArrayDeserializer()) {
+ @Override
+ public void subscribe(final Pattern topics, final ConsumerRebalanceListener listener) {
+ super.subscribe(topics, new TheConsumerRebalanceListener(assignedTopics, listener));
+ }
+ };
- final KafkaStreams streams = new KafkaStreams(builder.build(), streamsConfiguration);
-
- final Field streamThreadsField = streams.getClass().getDeclaredField("threads");
- streamThreadsField.setAccessible(true);
- final StreamThread[] streamThreads = (StreamThread[]) streamThreadsField.get(streams);
- final StreamThread originalThread = streamThreads[0];
+ }
+ });
- final TestStreamThread testStreamThread = new TestStreamThread(
- StreamsBuilderTest.internalTopologyBuilder(builder),
- streamsConfig,
- new DefaultKafkaClientSupplier(),
- originalThread.applicationId,
- originalThread.clientId,
- originalThread.processId,
- new Metrics(),
- Time.SYSTEM);
- final TestCondition oneTopicAdded = new TestCondition() {
+ streams.start();
+ TestUtils.waitForCondition(new TestCondition() {
@Override
public boolean conditionMet() {
- return testStreamThread.assignedTopicPartitions.equals(expectedFirstAssignment);
+ return assignedTopics.equals(expectedFirstAssignment);
}
- };
-
- streamThreads[0] = testStreamThread;
- streams.start();
-
- TestUtils.waitForCondition(oneTopicAdded, STREAM_TASKS_NOT_UPDATED);
+ }, STREAM_TASKS_NOT_UPDATED);
CLUSTER.createTopic("TEST-TOPIC-2");
- final TestCondition secondTopicAdded = new TestCondition() {
+ TestUtils.waitForCondition(new TestCondition() {
@Override
public boolean conditionMet() {
- return testStreamThread.assignedTopicPartitions.equals(expectedSecondAssignment);
+ return assignedTopics.equals(expectedSecondAssignment);
}
- };
-
- TestUtils.waitForCondition(secondTopicAdded, STREAM_TASKS_NOT_UPDATED);
+ }, STREAM_TASKS_NOT_UPDATED);
- streams.close();
}
@Test
@@ -208,49 +194,40 @@ public class RegexSourceIntegrationTest {
pattern1Stream.to(stringSerde, stringSerde, DEFAULT_OUTPUT_TOPIC);
- final KafkaStreams streams = new KafkaStreams(builder.build(), streamsConfiguration);
-
- final Field streamThreadsField = streams.getClass().getDeclaredField("threads");
- streamThreadsField.setAccessible(true);
- final StreamThread[] streamThreads = (StreamThread[]) streamThreadsField.get(streams);
- final StreamThread originalThread = streamThreads[0];
+ final List<String> assignedTopics = new ArrayList<>();
+ streams = new KafkaStreams(builder.build(), streamsConfig, new DefaultKafkaClientSupplier() {
+ @Override
+ public Consumer<byte[], byte[]> getConsumer(final Map<String, Object> config) {
+ return new KafkaConsumer<byte[], byte[]>(config, new ByteArrayDeserializer(), new ByteArrayDeserializer()) {
+ @Override
+ public void subscribe(final Pattern topics, final ConsumerRebalanceListener listener) {
+ super.subscribe(topics, new TheConsumerRebalanceListener(assignedTopics, listener));
+ }
+ };
- final TestStreamThread testStreamThread = new TestStreamThread(
- StreamsBuilderTest.internalTopologyBuilder(builder),
- streamsConfig,
- new DefaultKafkaClientSupplier(),
- originalThread.applicationId,
- originalThread.clientId,
- originalThread.processId,
- new Metrics(),
- Time.SYSTEM);
+ }
+ });
- streamThreads[0] = testStreamThread;
- final TestCondition bothTopicsAdded = new TestCondition() {
+ streams.start();
+ TestUtils.waitForCondition(new TestCondition() {
@Override
public boolean conditionMet() {
- return testStreamThread.assignedTopicPartitions.equals(expectedFirstAssignment);
+ return assignedTopics.equals(expectedFirstAssignment);
}
- };
- streams.start();
-
- TestUtils.waitForCondition(bothTopicsAdded, STREAM_TASKS_NOT_UPDATED);
+ }, STREAM_TASKS_NOT_UPDATED);
CLUSTER.deleteTopic("TEST-TOPIC-A");
- final TestCondition oneTopicRemoved = new TestCondition() {
+ TestUtils.waitForCondition(new TestCondition() {
@Override
public boolean conditionMet() {
- return testStreamThread.assignedTopicPartitions.equals(expectedSecondAssignment);
+ return assignedTopics.equals(expectedSecondAssignment);
}
- };
-
- TestUtils.waitForCondition(oneTopicRemoved, STREAM_TASKS_NOT_UPDATED);
-
- streams.close();
+ }, STREAM_TASKS_NOT_UPDATED);
}
+ @SuppressWarnings("deprecation")
@Test
public void shouldAddStateStoreToRegexDefinedSource() throws Exception {
@@ -264,7 +241,7 @@ public class RegexSourceIntegrationTest {
.addStateStore(stateStoreSupplier, "my-processor");
- final KafkaStreams streams = new KafkaStreams(builder, streamsConfiguration);
+ streams = new KafkaStreams(builder, streamsConfiguration);
try {
streams.start();
@@ -308,7 +285,7 @@ public class RegexSourceIntegrationTest {
pattern2Stream.to(stringSerde, stringSerde, DEFAULT_OUTPUT_TOPIC);
namedTopicsStream.to(stringSerde, stringSerde, DEFAULT_OUTPUT_TOPIC);
- final KafkaStreams streams = new KafkaStreams(builder.build(), streamsConfiguration);
+ streams = new KafkaStreams(builder.build(), streamsConfiguration);
streams.start();
final Properties producerConfig = TestUtils.producerConfig(CLUSTER.bootstrapServers(), StringSerializer.class, StringSerializer.class);
@@ -330,7 +307,6 @@ public class RegexSourceIntegrationTest {
actualValues.add(receivedKeyValue.value);
}
- streams.close();
Collections.sort(actualValues);
Collections.sort(expectedReceivedValues);
assertThat(actualValues, equalTo(expectedReceivedValues));
@@ -339,84 +315,67 @@ public class RegexSourceIntegrationTest {
@Test
public void testMultipleConsumersCanReadFromPartitionedTopic() throws Exception {
- final Serde<String> stringSerde = Serdes.String();
- final StreamsBuilder builderLeader = new StreamsBuilder();
- final StreamsBuilder builderFollower = new StreamsBuilder();
- final List<String> expectedAssignment = Arrays.asList(PARTITIONED_TOPIC_1, PARTITIONED_TOPIC_2);
-
- final KStream<String, String> partitionedStreamLeader = builderLeader.stream(Pattern.compile("partitioned-\\d"));
- final KStream<String, String> partitionedStreamFollower = builderFollower.stream(Pattern.compile("partitioned-\\d"));
+ KafkaStreams partitionedStreamsLeader = null;
+ KafkaStreams partitionedStreamsFollower = null;
+ try {
+ final Serde<String> stringSerde = Serdes.String();
+ final StreamsBuilder builderLeader = new StreamsBuilder();
+ final StreamsBuilder builderFollower = new StreamsBuilder();
+ final List<String> expectedAssignment = Arrays.asList(PARTITIONED_TOPIC_1, PARTITIONED_TOPIC_2);
+ final KStream<String, String> partitionedStreamLeader = builderLeader.stream(Pattern.compile("partitioned-\\d"));
+ final KStream<String, String> partitionedStreamFollower = builderFollower.stream(Pattern.compile("partitioned-\\d"));
- partitionedStreamLeader.to(stringSerde, stringSerde, DEFAULT_OUTPUT_TOPIC);
- partitionedStreamFollower.to(stringSerde, stringSerde, DEFAULT_OUTPUT_TOPIC);
- final KafkaStreams partitionedStreamsLeader = new KafkaStreams(builderLeader.build(), streamsConfiguration);
- final KafkaStreams partitionedStreamsFollower = new KafkaStreams(builderFollower.build(), streamsConfiguration);
+ partitionedStreamLeader.to(stringSerde, stringSerde, DEFAULT_OUTPUT_TOPIC);
+ partitionedStreamFollower.to(stringSerde, stringSerde, DEFAULT_OUTPUT_TOPIC);
- final StreamsConfig streamsConfig = new StreamsConfig(streamsConfiguration);
+ final List<String> leaderAssignment = new ArrayList<>();
+ final List<String> followerAssignment = new ArrayList<>();
+ StreamsConfig config = new StreamsConfig(streamsConfiguration);
+ partitionedStreamsLeader = new KafkaStreams(builderLeader.build(), config, new DefaultKafkaClientSupplier() {
+ @Override
+ public Consumer<byte[], byte[]> getConsumer(final Map<String, Object> config) {
+ return new KafkaConsumer<byte[], byte[]>(config, new ByteArrayDeserializer(), new ByteArrayDeserializer()) {
+ @Override
+ public void subscribe(final Pattern topics, final ConsumerRebalanceListener listener) {
+ super.subscribe(topics, new TheConsumerRebalanceListener(leaderAssignment, listener));
+ }
+ };
- final Field leaderStreamThreadsField = partitionedStreamsLeader.getClass().getDeclaredField("threads");
- leaderStreamThreadsField.setAccessible(true);
- final StreamThread[] leaderStreamThreads = (StreamThread[]) leaderStreamThreadsField.get(partitionedStreamsLeader);
- final StreamThread originalLeaderThread = leaderStreamThreads[0];
+ }
+ });
+ partitionedStreamsFollower = new KafkaStreams(builderFollower.build(), config, new DefaultKafkaClientSupplier() {
+ @Override
+ public Consumer<byte[], byte[]> getConsumer(final Map<String, Object> config) {
+ return new KafkaConsumer<byte[], byte[]>(config, new ByteArrayDeserializer(), new ByteArrayDeserializer()) {
+ @Override
+ public void subscribe(final Pattern topics, final ConsumerRebalanceListener listener) {
+ super.subscribe(topics, new TheConsumerRebalanceListener(followerAssignment, listener));
+ }
+ };
- final TestStreamThread leaderTestStreamThread = new TestStreamThread(
- StreamsBuilderTest.internalTopologyBuilder(builderLeader),
- streamsConfig,
- new DefaultKafkaClientSupplier(),
- originalLeaderThread.applicationId,
- originalLeaderThread.clientId,
- originalLeaderThread.processId,
- new Metrics(),
- Time.SYSTEM);
+ }
+ });
- leaderStreamThreads[0] = leaderTestStreamThread;
- final TestCondition bothTopicsAddedToLeader = new TestCondition() {
- @Override
- public boolean conditionMet() {
- return leaderTestStreamThread.assignedTopicPartitions.equals(expectedAssignment);
+ partitionedStreamsLeader.start();
+ partitionedStreamsFollower.start();
+ TestUtils.waitForCondition(new TestCondition() {
+ @Override
+ public boolean conditionMet() {
+ return followerAssignment.equals(expectedAssignment) && leaderAssignment.equals(expectedAssignment);
+ }
+ }, "topic assignment not completed");
+ } finally {
+ if (partitionedStreamsLeader != null) {
+ partitionedStreamsLeader.close();
}
- };
-
-
-
- final Field followerStreamThreadsField = partitionedStreamsFollower.getClass().getDeclaredField("threads");
- followerStreamThreadsField.setAccessible(true);
- final StreamThread[] followerStreamThreads = (StreamThread[]) followerStreamThreadsField.get(partitionedStreamsFollower);
- final StreamThread originalFollowerThread = followerStreamThreads[0];
-
- final TestStreamThread followerTestStreamThread = new TestStreamThread(
- StreamsBuilderTest.internalTopologyBuilder(builderFollower),
- streamsConfig,
- new DefaultKafkaClientSupplier(),
- originalFollowerThread.applicationId,
- originalFollowerThread.clientId,
- originalFollowerThread.processId,
- new Metrics(),
- Time.SYSTEM);
-
- followerStreamThreads[0] = followerTestStreamThread;
-
-
- final TestCondition bothTopicsAddedToFollower = new TestCondition() {
- @Override
- public boolean conditionMet() {
- return followerTestStreamThread.assignedTopicPartitions.equals(expectedAssignment);
+ if (partitionedStreamsFollower != null) {
+ partitionedStreamsFollower.close();
}
- };
-
- partitionedStreamsLeader.start();
- TestUtils.waitForCondition(bothTopicsAddedToLeader, "Topics never assigned to leader stream");
-
-
- partitionedStreamsFollower.start();
- TestUtils.waitForCondition(bothTopicsAddedToFollower, "Topics never assigned to follower stream");
-
- partitionedStreamsLeader.close();
- partitionedStreamsFollower.close();
+ }
}
@@ -443,7 +402,7 @@ public class RegexSourceIntegrationTest {
pattern1Stream.to(stringSerde, stringSerde, DEFAULT_OUTPUT_TOPIC);
pattern2Stream.to(stringSerde, stringSerde, DEFAULT_OUTPUT_TOPIC);
- final KafkaStreams streams = new KafkaStreams(builder.build(), streamsConfiguration);
+ streams = new KafkaStreams(builder.build(), streamsConfiguration);
streams.start();
final Properties producerConfig = TestUtils.producerConfig(CLUSTER.bootstrapServers(), StringSerializer.class, StringSerializer.class);
@@ -453,34 +412,33 @@ public class RegexSourceIntegrationTest {
final Properties consumerConfig = TestUtils.consumerConfig(CLUSTER.bootstrapServers(), StringDeserializer.class, StringDeserializer.class);
- try {
- IntegrationTestUtils.waitUntilMinKeyValueRecordsReceived(consumerConfig, DEFAULT_OUTPUT_TOPIC, 2, 5000);
- fail("Should not get here");
- } finally {
- streams.close();
- }
-
+ IntegrationTestUtils.waitUntilMinKeyValueRecordsReceived(consumerConfig, DEFAULT_OUTPUT_TOPIC, 2, 5000);
+ fail("Should not get here");
}
- private class TestStreamThread extends StreamThread {
- public volatile List<String> assignedTopicPartitions = new ArrayList<>();
+ private static class TheConsumerRebalanceListener implements ConsumerRebalanceListener {
+ private final List<String> assignedTopics;
+ private final ConsumerRebalanceListener listener;
+
+ TheConsumerRebalanceListener(final List<String> assignedTopics, final ConsumerRebalanceListener listener) {
+ this.assignedTopics = assignedTopics;
+ this.listener = listener;
+ }
- public TestStreamThread(final InternalTopologyBuilder builder, final StreamsConfig config, final KafkaClientSupplier clientSupplier, final String applicationId, final String clientId, final UUID processId, final Metrics metrics, final Time time) {
- super(builder, config, clientSupplier, applicationId, clientId, processId, metrics, time, new StreamsMetadataState(builder, StreamsMetadataState.UNKNOWN_HOST),
- 0, new StateDirectory(applicationId, config.getString(StreamsConfig.STATE_DIR_CONFIG), time));
+ @Override
+ public void onPartitionsRevoked(final Collection<TopicPartition> partitions) {
+ assignedTopics.clear();
+ listener.onPartitionsRevoked(partitions);
}
@Override
- public StreamTask createStreamTask(final TaskId id, final Collection<TopicPartition> partitions) {
- final List<String> topicPartitions = new ArrayList<>();
+ public void onPartitionsAssigned(final Collection<TopicPartition> partitions) {
for (final TopicPartition partition : partitions) {
- topicPartitions.add(partition.topic());
+ assignedTopics.add(partition.topic());
}
- Collections.sort(topicPartitions);
-
- assignedTopicPartitions = topicPartitions;
- return super.createStreamTask(id, partitions);
+ Collections.sort(assignedTopics);
+ listener.onPartitionsAssigned(partitions);
}
-
}
+
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/3e69ce80/streams/src/test/java/org/apache/kafka/streams/processor/internals/AbstractTaskTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/AbstractTaskTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/AbstractTaskTest.java
index f78b43a..353f740 100644
--- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/AbstractTaskTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/AbstractTaskTest.java
@@ -17,6 +17,7 @@
package org.apache.kafka.streams.processor.internals;
import org.apache.kafka.clients.consumer.Consumer;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.MockConsumer;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.clients.consumer.OffsetResetStrategy;
@@ -35,6 +36,8 @@ import org.apache.kafka.test.TestUtils;
import org.junit.Test;
import java.util.Collections;
+import java.util.List;
+import java.util.Map;
import java.util.Properties;
public class AbstractTaskTest {
@@ -91,6 +94,46 @@ public class AbstractTaskTest {
@Override
public void close(final boolean clean) {}
+
+ @Override
+ public void closeSuspended(final boolean clean, final RuntimeException e) {
+
+ }
+
+ @Override
+ public Map<TopicPartition, Long> checkpointedOffsets() {
+ return null;
+ }
+
+ @Override
+ public boolean process() {
+ return false;
+ }
+
+ @Override
+ public boolean maybePunctuateStreamTime() {
+ return false;
+ }
+
+ @Override
+ public boolean maybePunctuateSystemTime() {
+ return false;
+ }
+
+ @Override
+ public List<ConsumerRecord<byte[], byte[]>> update(final TopicPartition partition, final List<ConsumerRecord<byte[], byte[]>> remaining) {
+ return null;
+ }
+
+ @Override
+ public int addRecords(final TopicPartition partition, final Iterable<ConsumerRecord<byte[], byte[]>> records) {
+ return 0;
+ }
+
+ @Override
+ public boolean commitNeeded() {
+ return false;
+ }
};
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/3e69ce80/streams/src/test/java/org/apache/kafka/streams/processor/internals/StandbyTaskTest.java
----------------------------------------------------------------------
diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StandbyTaskTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StandbyTaskTest.java
index fad7116..e232316 100644
--- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StandbyTaskTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StandbyTaskTest.java
@@ -32,6 +32,7 @@ import org.apache.kafka.common.utils.MockTime;
import org.apache.kafka.common.utils.Time;
import org.apache.kafka.common.utils.Utils;
import org.apache.kafka.streams.StreamsConfig;
+import org.apache.kafka.streams.errors.ProcessorStateException;
import org.apache.kafka.streams.kstream.internals.InternalStreamsBuilder;
import org.apache.kafka.streams.kstream.internals.InternalStreamsBuilderTest;
import org.apache.kafka.streams.processor.StateStore;
@@ -57,12 +58,15 @@ import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.Set;
+import java.util.concurrent.atomic.AtomicBoolean;
import static java.util.Collections.singleton;
import static org.hamcrest.CoreMatchers.equalTo;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
public class StandbyTaskTest {
@@ -383,6 +387,48 @@ public class StandbyTaskTest {
}
+ @Test
+ public void shouldCloseStateMangerOnTaskCloseWhenCommitFailed() throws Exception {
+ consumer.assign(Utils.mkList(ktable));
+ final Map<TopicPartition, OffsetAndMetadata> committedOffsets = new HashMap<>();
+ committedOffsets.put(new TopicPartition(ktable.topic(), ktable.partition()), new OffsetAndMetadata(100L));
+ consumer.commitSync(committedOffsets);
+
+ restoreStateConsumer.updatePartitions("ktable1", Utils.mkList(
+ new PartitionInfo("ktable1", 0, Node.noNode(), new Node[0], new Node[0])));
+
+ final StreamsConfig config = createConfig(baseDir);
+ final AtomicBoolean closedStateManager = new AtomicBoolean(false);
+ final StandbyTask task = new StandbyTask(taskId,
+ applicationId,
+ ktablePartitions,
+ ktableTopology,
+ consumer,
+ changelogReader,
+ config,
+ null,
+ stateDirectory
+ ) {
+ @Override
+ public void commit() {
+ throw new RuntimeException("KABOOM!");
+ }
+
+ @Override
+ void closeStateManager(final boolean writeCheckpoint) throws ProcessorStateException {
+ closedStateManager.set(true);
+ }
+ };
+
+ try {
+ task.close(true);
+ fail("should have thrown exception");
+ } catch (Exception e) {
+ // expected
+ }
+ assertTrue(closedStateManager.get());
+ }
+
private List<ConsumerRecord<byte[], byte[]>> records(ConsumerRecord<byte[], byte[]>... recs) {
return Arrays.asList(recs);
}