Fix some metrics

streams/src/main/java/org/apache/kafka/streams/StreamingConfig.java

@@ -68,6 +68,9 @@ public class StreamingConfig extends AbstractConfig {
     public static final String TIMESTAMP_EXTRACTOR_CLASS_CONFIG = "timestamp.extractor";
     private static final String TIMESTAMP_EXTRACTOR_CLASS_DOC = "Timestamp extractor class that implements the <code>TimestampExtractor</code> interface.";
 
+    /** <code>client.id</code> */
+    public static final String CLIENT_ID_CONFIG = CommonClientConfigs.CLIENT_ID_CONFIG;
+
     /** <code>key.serializer</code> */
     public static final String KEY_SERIALIZER_CLASS_CONFIG = ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG;
 
@@ -88,7 +91,12 @@ public class StreamingConfig extends AbstractConfig {
     private static final String SYSTEM_TEMP_DIRECTORY = System.getProperty("java.io.tmpdir");
 
     static {
-        CONFIG = new ConfigDef().define(STATE_DIR_CONFIG,
+        CONFIG = new ConfigDef().define(CLIENT_ID_CONFIG,
+                                        Type.STRING,
+                                        "",
+                                        Importance.MEDIUM,
+                                        CommonClientConfigs.CLIENT_ID_DOC)
+                                .define(STATE_DIR_CONFIG,
                                         Type.STRING,
                                         SYSTEM_TEMP_DIRECTORY,
                                         Importance.MEDIUM,

streams/src/main/java/org/apache/kafka/streams/examples/KStreamJob.java

@@ -34,6 +34,7 @@ public class KStreamJob {
 
     public static void main(String[] args) throws Exception {
         Properties props = new Properties();
+        props.put(StreamingConfig.CLIENT_ID_CONFIG, "Example-KStream-Job");
         props.put(StreamingConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
         props.put(StreamingConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
         props.put(StreamingConfig.VALUE_SERIALIZER_CLASS_CONFIG, IntegerSerializer.class);

streams/src/main/java/org/apache/kafka/streams/examples/ProcessorJob.java

@@ -65,7 +65,7 @@ public void process(String key, String value) {
                 }
 
                 @Override
-                public void punctuate(long streamTime) {
+                public void punctuate(long timestamp) {
                     KeyValueIterator<String, Integer> iter = this.kvStore.all();
 
                     while (iter.hasNext()) {
@@ -87,6 +87,7 @@ public void close() {
 
     public static void main(String[] args) throws Exception {
         Properties props = new Properties();
+        props.put(StreamingConfig.CLIENT_ID_CONFIG, "Example-Processor-Job");
         props.put(StreamingConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
         props.put(StreamingConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
         props.put(StreamingConfig.VALUE_SERIALIZER_CLASS_CONFIG, IntegerSerializer.class);

streams/src/main/java/org/apache/kafka/streams/processor/Processor.java

@@ -23,7 +23,7 @@ public interface Processor<K, V> {
 
     void process(K key, V value);
 
-    void punctuate(long streamTime);
+    void punctuate(long timestamp);
 
     void close();
 }

streams/src/main/java/org/apache/kafka/streams/processor/internals/PunctuationQueue.java

@@ -35,17 +35,21 @@ public void close() {
         }
     }
 
-    public void mayPunctuate(long streamTime, Punctuator punctuator) {
+    public boolean mayPunctuate(long timestamp, Punctuator punctuator) {
         synchronized (pq) {
+            boolean punctuated = false;
             PunctuationSchedule top = pq.peek();
-            while (top != null && top.timestamp <= streamTime) {
+            while (top != null && top.timestamp <= timestamp) {
                 PunctuationSchedule sched = top;
                 pq.poll();
-                punctuator.punctuate(sched.node(), streamTime);
+                punctuator.punctuate(sched.node(), timestamp);
                 pq.add(sched.next());
+                punctuated = true;
 
                 top = pq.peek();
             }
+
+            return punctuated;
         }
     }
 

streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java

@@ -164,7 +164,7 @@ public void addRecords(TopicPartition partition, Iterable<ConsumerRecord<byte[],
     }
 
     /**
-     * Processes one record
+     * Process one record
      *
      * @return number of records left in the buffer of this task's partition group after the processing is done
      */
@@ -194,11 +194,6 @@ public int process() {
                 consumedOffsets.put(partition, currRecord.offset());
                 commitOffsetNeeded = true;
 
-                // commit the current task state if requested during the processing
-                if (commitRequested) {
-                    commit();
-                }
-
                 // after processing this record, if its partition queue's buffered size has been
                 // decreased to the threshold, we can then resume the consumption on this partition
                 if (partitionGroup.numBuffered(partition) == this.maxBufferedSize) {
@@ -209,23 +204,28 @@ public int process() {
                 this.currNode = null;
             }
 
-            // possibly trigger registered punctuation functions if
-            // partition group's time has reached the defined stamp
-            long timestamp = partitionGroup.timestamp();
-            punctuationQueue.mayPunctuate(timestamp, this);
-
             return partitionGroup.numBuffered();
         }
     }
 
+    /**
+     * Possibly trigger registered punctuation functions if
+     * current time has reached the defined stamp
+     *
+     * @param timestamp
+     */
+    public boolean maybePunctuate(long timestamp) {
+        return punctuationQueue.mayPunctuate(timestamp, this);
+    }
+
     @Override
-    public void punctuate(ProcessorNode node, long streamTime) {
+    public void punctuate(ProcessorNode node, long timestamp) {
         if (currNode != null)
             throw new IllegalStateException("Current node is not null");
 
         currNode = node;
         try {
-            node.processor().punctuate(streamTime);
+            node.processor().punctuate(timestamp);
         } finally {
             currNode = null;
         }
@@ -259,9 +259,17 @@ public void commit() {
             consumer.commitSync(consumedOffsets);
             commitOffsetNeeded = false;
         }
+
         commitRequested = false;
     }
 
+    /**
+     * Whether or not a request has been made to commit the current state
+     */
+    public boolean commitNeeded() {
+        return this.commitRequested;
+    }
+
     /**
      * Request committing the current task's state
      */