kafka-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From damian...@apache.org
Subject [5/5] kafka git commit: KAFKA-5702; extract refactor StreamThread
Date Fri, 11 Aug 2017 11:14:11 GMT
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 <damian.guy@gmail.com>

Reviewers: Bill Bejeck <bill@confluent.io>, Guozhang Wang <wangguoz@gmail.com>,
Eno Thereska <eno.thereska@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 <damian.guy@gmail.com>
Authored: Fri Aug 11 12:14:01 2017 +0100
Committer: Damian Guy <damian.guy@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;
     }
 


Mime
View raw message