KAFKA-13945: add bytes/records consumed and produced metrics

checkstyle/suppressions.xml

@@ -213,7 +213,7 @@
 
     <!-- Streams tests -->
     <suppress checks="ClassFanOutComplexity"
-              files="(StreamsPartitionAssignorTest|StreamThreadTest|StreamTaskTest|TaskManagerTest|TopologyTestDriverTest).java"/>
+              files="(RecordCollectorTest|StreamsPartitionAssignorTest|StreamThreadTest|StreamTaskTest|TaskManagerTest|TopologyTestDriverTest).java"/>
 
     <suppress checks="MethodLength"
               files="(EosIntegrationTest|EosV2UpgradeIntegrationTest|KStreamKStreamJoinTest|RocksDBWindowStoreTest).java"/>

docs/ops.html

@@ -2308,7 +2308,7 @@ <h4 class="anchor-heading"><a id="connect_monitoring" class="anchor-link"></a><a
   <h4 class="anchor-heading"><a id="kafka_streams_monitoring" class="anchor-link"></a><a href="#kafka_streams_monitoring">Streams Monitoring</a></h4>
 
   A Kafka Streams instance contains all the producer and consumer metrics as well as additional metrics specific to Streams.
-  By default Kafka Streams has metrics with three recording levels: <code>info</code>, <code>debug</code>, and <code>trace</code>.
+  The metrics have three recording levels: <code>info</code>, <code>debug</code>, and <code>trace</code>.
 
   <p>
     Note that the metrics have a 4-layer hierarchy. At the top level there are client-level metrics for each started
@@ -2617,7 +2617,7 @@ <h5 class="anchor-heading"><a id="kafka_streams_node_monitoring" class="anchor-l
  </table>
 
  <h5 class="anchor-heading"><a id="kafka_streams_store_monitoring" class="anchor-link"></a><a href="#kafka_streams_store_monitoring">State Store Metrics</a></h5>
-All of the following metrics have a recording level of <code>debug</code>, except for the record-e2e-latency-* metrics which have a recording level <code>trace></code>.
+All of the following metrics have a recording level of <code>debug</code>, except for the record-e2e-latency-* metrics which have a recording level <code>trace</code>.
 Note that the <code>store-scope</code> value is specified in <code>StoreSupplier#metricsScope()</code> for user's customized state stores;
 for built-in state stores, currently we have:
   <ul>

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

@@ -260,7 +260,8 @@ private StreamTask createActiveTask(final TaskId taskId,
             taskId,
             streamsProducer,
             applicationConfig.defaultProductionExceptionHandler(),
-            streamsMetrics
+            streamsMetrics,
+            topology
         );
 
         final StreamTask task = new StreamTask(

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

@@ -21,12 +21,17 @@
 import org.apache.kafka.clients.admin.OffsetSpec;
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.producer.ProducerRecord;
 import org.apache.kafka.common.KafkaException;
 import org.apache.kafka.common.KafkaFuture;
 import org.apache.kafka.common.Metric;
 import org.apache.kafka.common.MetricName;
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.errors.TimeoutException;
+import org.apache.kafka.common.header.Header;
+import org.apache.kafka.common.header.Headers;
+import org.apache.kafka.common.utils.Utils;
 import org.apache.kafka.streams.StreamsConfig;
 import org.apache.kafka.streams.errors.StreamsException;
 import org.apache.kafka.streams.processor.TaskId;
@@ -166,4 +171,46 @@ public static String extractThreadId(final String fullThreadName) {
         final int index = fullThreadName.indexOf("StreamThread-");
         return fullThreadName.substring(index);
     }
+
+    public static long producerRecordSizeInBytes(final ProducerRecord<byte[], byte[]> record) {
+        return recordSizeInBytes(
+            record.key().length,
+            record.value() == null ? 0 : record.value().length,
+            record.topic(),
+            record.headers()
+        );
+    }
+
+    public static long consumerRecordSizeInBytes(final ConsumerRecord<byte[], byte[]> record) {
+        return recordSizeInBytes(
+            record.serializedKeySize(),
+            record.serializedValueSize(),
+            record.topic(),
+            record.headers()
+        );
+    }
+
+    public static long recordSizeInBytes(final long keyBytes,
+                                         final long valueBytes,
+                                         final String topic,
+                                         final Headers headers) {
+        long headerSizeInBytes = 0L;
+
+        if (headers != null) {
+            for (final Header header : headers.toArray()) {
+                headerSizeInBytes += Utils.utf8(header.key()).length;
+                if (header.value() != null) {
+                    headerSizeInBytes += header.value().length;
+                }
+            }
+        }
+
+        return keyBytes +
+            valueBytes +
+            8L + // timestamp
+            8L + // offset
+            Utils.utf8(topic).length +
+            4L + // partition
+            headerSizeInBytes;
+    }
 }

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

@@ -268,7 +268,7 @@ StampedRecord nextRecord(final RecordInfo info, final long wallClockTime) {
             // get the buffer size of queue before poll
             final long oldBufferSize = queue.getTotalBytesBuffered();
             // get the first record from this queue.
-            record = queue.poll();
+            record = queue.poll(wallClockTime);
             // After polling, the buffer size would have reduced.
             final long newBufferSize = queue.getTotalBytesBuffered();
 
@@ -392,6 +392,12 @@ void clear() {
         streamTime = RecordQueue.UNKNOWN;
     }
 
+    void close() {
+        for (final RecordQueue queue : partitionQueues.values()) {
+            queue.close();
+        }
+    }
+
     // Below methods are for only testing.
 
     boolean allPartitionsBufferedLocally() {

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

@@ -147,8 +147,9 @@ public void logChange(final String storeName,
             changelogPartition.partition(),
             timestamp,
             BYTES_KEY_SERIALIZER,
-            BYTEARRAY_VALUE_SERIALIZER
-        );
+            BYTEARRAY_VALUE_SERIALIZER,
+            null,
+            null);
     }
 
     /**

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

@@ -33,7 +33,9 @@ <K, V> void send(final String topic,
                      final Integer partition,
                      final Long timestamp,
                      final Serializer<K> keySerializer,
-                     final Serializer<V> valueSerializer);
+                     final Serializer<V> valueSerializer,
+                     final String processorNodeId,
+                     final InternalProcessorContext<Void, Void> context);
 
     <K, V> void send(final String topic,
                      final K key,
@@ -42,6 +44,8 @@ <K, V> void send(final String topic,
                      final Long timestamp,
                      final Serializer<K> keySerializer,
                      final Serializer<V> valueSerializer,
+                     final String processorNodeId,
+                     final InternalProcessorContext<Void, Void> context,
                      final StreamPartitioner<? super K, ? super V> partitioner);
 
     /**

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

@@ -46,6 +46,8 @@
 import org.apache.kafka.streams.processor.TaskId;
 import org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl;
 import org.apache.kafka.streams.processor.internals.metrics.TaskMetrics;
+import org.apache.kafka.streams.processor.internals.metrics.TopicMetrics;
+
 import org.slf4j.Logger;
 
 import java.util.Collections;
@@ -54,17 +56,22 @@
 import java.util.Map;
 import java.util.concurrent.atomic.AtomicReference;
 
+import static org.apache.kafka.streams.processor.internals.ClientUtils.producerRecordSizeInBytes;
+
 public class RecordCollectorImpl implements RecordCollector {
     private final static String SEND_EXCEPTION_MESSAGE = "Error encountered sending record to topic %s for task %s due to:%n%s";
 
     private final Logger log;
     private final TaskId taskId;
     private final StreamsProducer streamsProducer;
     private final ProductionExceptionHandler productionExceptionHandler;
-    private final Sensor droppedRecordsSensor;
     private final boolean eosEnabled;
     private final Map<TopicPartition, Long> offsets;
 
+    private final StreamsMetricsImpl streamsMetrics;
+    private final Sensor droppedRecordsSensor;
+    private final Map<String, Map<String, Sensor>> sinkNodeToProducedSensorByTopic = new HashMap<>();
+
     private final AtomicReference<KafkaException> sendException = new AtomicReference<>(null);
 
     /**
@@ -74,15 +81,29 @@ public RecordCollectorImpl(final LogContext logContext,
                                final TaskId taskId,
                                final StreamsProducer streamsProducer,
                                final ProductionExceptionHandler productionExceptionHandler,
-                               final StreamsMetricsImpl streamsMetrics) {
+                               final StreamsMetricsImpl streamsMetrics,
+                               final ProcessorTopology topology) {
         this.log = logContext.logger(getClass());
         this.taskId = taskId;
         this.streamsProducer = streamsProducer;
         this.productionExceptionHandler = productionExceptionHandler;
         this.eosEnabled = streamsProducer.eosEnabled();
+        this.streamsMetrics = streamsMetrics;
 
         final String threadId = Thread.currentThread().getName();
         this.droppedRecordsSensor = TaskMetrics.droppedRecordsSensor(threadId, taskId.toString(), streamsMetrics);
+        for (final String topic : topology.sinkTopics()) {
+            final String processorNodeId = topology.sink(topic).name();
+            sinkNodeToProducedSensorByTopic.computeIfAbsent(processorNodeId, t -> new HashMap<>()).put(
+                topic,
+                TopicMetrics.producedSensor(
+                    threadId,
+                    taskId.toString(),
+                    processorNodeId,
+                    topic,
+                    streamsMetrics
+                ));
+        }
 
         this.offsets = new HashMap<>();
     }
@@ -106,6 +127,8 @@ public <K, V> void send(final String topic,
                             final Long timestamp,
                             final Serializer<K> keySerializer,
                             final Serializer<V> valueSerializer,
+                            final String processorNodeId,
+                            final InternalProcessorContext<Void, Void> context,
                             final StreamPartitioner<? super K, ? super V> partitioner) {
         final Integer partition;
 
@@ -122,7 +145,7 @@ public <K, V> void send(final String topic,
                 // here we cannot drop the message on the floor even if it is a transient timeout exception,
                 // so we treat everything the same as a fatal exception
                 throw new StreamsException("Could not determine the number of partitions for topic '" + topic +
-                    "' for task " + taskId + " due to " + fatal.toString(),
+                    "' for task " + taskId + " due to " + fatal,
                     fatal
                 );
             }
@@ -136,7 +159,7 @@ public <K, V> void send(final String topic,
             partition = null;
         }
 
-        send(topic, key, value, headers, partition, timestamp, keySerializer, valueSerializer);
+        send(topic, key, value, headers, partition, timestamp, keySerializer, valueSerializer, processorNodeId, context);
     }
 
     @Override
@@ -147,7 +170,9 @@ public <K, V> void send(final String topic,
                             final Integer partition,
                             final Long timestamp,
                             final Serializer<K> keySerializer,
-                            final Serializer<V> valueSerializer) {
+                            final Serializer<V> valueSerializer,
+                            final String processorNodeId,
+                            final InternalProcessorContext<Void, Void> context) {
         checkForException();
 
         final byte[] keyBytes;
@@ -173,7 +198,7 @@ public <K, V> void send(final String topic,
                     valueClass),
                 exception);
         } catch (final RuntimeException exception) {
-            final String errorMessage = String.format(SEND_EXCEPTION_MESSAGE, topic, taskId, exception.toString());
+            final String errorMessage = String.format(SEND_EXCEPTION_MESSAGE, topic, taskId, exception);
             throw new StreamsException(errorMessage, exception);
         }
 
@@ -192,6 +217,28 @@ public <K, V> void send(final String topic,
                 } else {
                     log.warn("Received offset={} in produce response for {}", metadata.offset(), tp);
                 }
+
+                if (!topic.endsWith("-changelog")) {
+                    // we may not have created a sensor yet if the node uses dynamic topic routing
+                    final Map<String, Sensor> producedSensorByTopic = sinkNodeToProducedSensorByTopic.get(processorNodeId);
+                    if (producedSensorByTopic == null) {
+                        log.error("Unable to records bytes produced to topic {} by sink node {} as the node is not recognized.\n"
+                                      + "Known sink nodes are {}.", topic, processorNodeId, sinkNodeToProducedSensorByTopic.keySet());
+                    } else {
+                        final Sensor topicProducedSensor = producedSensorByTopic.computeIfAbsent(
+                            topic,
+                            t -> TopicMetrics.producedSensor(
+                                Thread.currentThread().getName(),
+                                taskId.toString(),
+                                processorNodeId,
+                                topic,
+                                context.metrics()
+                            )
+                        );
+                        final long bytesProduced = producerRecordSizeInBytes(serializedRecord);
+                        topicProducedSensor.record(bytesProduced, context.currentSystemTimeMs());
+                    }
+                }
             } else {
                 recordSendError(topic, exception, serializedRecord);
 
@@ -267,6 +314,8 @@ public void flush() {
     public void closeClean() {
         log.info("Closing record collector clean");
 
+        removeAllProducedSensors();
+
         // No need to abort transaction during a clean close: either we have successfully committed the ongoing
         // transaction during handleRevocation and thus there is no transaction in flight, or else none of the revoked
         // tasks had any data in the current transaction and therefore there is no need to commit or abort it.
@@ -290,6 +339,14 @@ public void closeDirty() {
         checkForException();
     }
 
+    private void removeAllProducedSensors() {
+        for (final Map<String, Sensor> nodeMap : sinkNodeToProducedSensorByTopic.values()) {
+            for (final Sensor sensor : nodeMap.values()) {
+                streamsMetrics.removeSensor(sensor);
+            }
+        }
+    }
+
     @Override
     public Map<TopicPartition, Long> offsets() {
         return Collections.unmodifiableMap(new HashMap<>(offsets));