Add support for observability

ci/publish-documentation-to-github-pages.sh

@@ -2,8 +2,13 @@
 
 . $(pwd)/release-versions.txt
 
+./mvnw clean test-compile exec:java \
+  -Dexec.mainClass=io.micrometer.docs.DocsGeneratorCommand \
+  -Dexec.classpathScope="test" \
+  -Dexec.args='src/main/java/com/rabbitmq/stream/observation/micrometer .* target/micrometer-observation-docs'
+
 MESSAGE=$(git log -1 --pretty=%B)
-./mvnw clean buildnumber:create pre-site --no-transfer-progress
+./mvnw buildnumber:create pre-site --no-transfer-progress
 
 ./mvnw javadoc:javadoc -Dmaven.javadoc.skip=false --no-transfer-progress
 

pom.xml

@@ -74,6 +74,8 @@
     <gson.version>2.10.1</gson.version>
     <vavr.version>0.10.4</vavr.version>
     <paho.version>1.2.5</paho.version>
+    <micrometer-tracing-test.version>1.1.3</micrometer-tracing-test.version>
+    <micrometer-docs-generator.version>1.0.2</micrometer-docs-generator.version>
     <maven.compiler.plugin.version>3.11.0</maven.compiler.plugin.version>
     <maven-surefire-plugin.version>3.1.2</maven-surefire-plugin.version>
     <spring-boot-maven-plugin.version>2.7.6</spring-boot-maven-plugin.version>
@@ -281,7 +283,7 @@
       <groupId>org.eclipse.paho</groupId>
       <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
       <version>${paho.version}</version>
-    <scope>test</scope>
+      <scope>test</scope>
     </dependency>
 
     <dependency>
@@ -312,6 +314,20 @@
       <scope>test</scope>
     </dependency>
 
+    <dependency>
+      <groupId>io.micrometer</groupId>
+      <artifactId>micrometer-tracing-integration-test</artifactId>
+      <version>${micrometer-tracing-test.version}</version>
+      <scope>test</scope>
+    </dependency>
+
+    <dependency>
+      <groupId>io.micrometer</groupId>
+      <artifactId>micrometer-docs-generator</artifactId>
+      <version>${micrometer-docs-generator.version}</version>
+      <scope>test</scope>
+    </dependency>
+
     <dependency>
       <groupId>org.openjdk.jmh</groupId>
       <artifactId>jmh-core</artifactId>
@@ -393,6 +409,9 @@
         <groupId>org.apache.maven.plugins</groupId>
         <artifactId>maven-resources-plugin</artifactId>
         <version>${maven-resources-plugin.version}</version>
+        <configuration>
+          <propertiesEncoding>${project.build.sourceEncoding}</propertiesEncoding>
+        </configuration>
       </plugin>
 
       <plugin>
@@ -457,6 +476,7 @@
             <idseparator>-</idseparator>
             <source-highlighter>coderay</source-highlighter>
             <test-examples>../../test/java/com/rabbitmq/stream/docs</test-examples>
+            <build-directory>${project.build.directory}</build-directory>
           </attributes>
 
         </configuration>

src/docs/asciidoc/appendixes.adoc

@@ -0,0 +1,26 @@
+ifndef::build-directory[:build-directory: ../../../target]
+:test-examples: ../../test/java/com/rabbitmq/stream/docs
+
+[appendix]
+== Micrometer Observation
+
+It is possible to use https://micrometer.io/docs/observation[Micrometer Observation] to instrument publishing and consuming in the stream Java client.
+Micrometer Observation provides https://spring.io/blog/2022/10/12/observability-with-spring-boot-3[metrics, tracing, and log correlation with one single API].
+
+The stream Java client provides an `ObservationCollector` abstraction and an implementation for Micrometer Observation.
+The following snippet shows how to create and set up the Micrometer `ObservationCollector` implementation with an existing `ObservationRegistry`:
+
+.Configuring Micrometer Observation
+[source,java,indent=0]
+--------
+include::{test-examples}/EnvironmentUsage.java[tag=micrometer-observation]
+--------
+<1> Configure Micrometer `ObservationCollector` with builder
+<2> Set Micrometer `ObservationRegistry`
+
+The next sections document the conventions, spans, and metrics made available by the instrumentation.
+They are automatically generated from the source code with the https://github.com/micrometer-metrics/micrometer-docs-generator[Micrometer documentation generator].
+
+include::{build-directory}/micrometer-observation-docs/_conventions.adoc[]
+include::{build-directory}/micrometer-observation-docs/_spans.adoc[]
+include::{build-directory}/micrometer-observation-docs/_metrics.adoc[]

src/docs/asciidoc/index.adoc

@@ -1,6 +1,7 @@
 = RabbitMQ Stream Java Client
 :revnumber: {project-version}
 :revremark: ({build-number})
+:appendix-caption: Appendix
 ifndef::imagesdir[:imagesdir: images]
 ifndef::sourcedir[:sourcedir: ../../main/java]
 :source-highlighter: prettify
@@ -28,4 +29,6 @@ include::advanced-topics.adoc[]
 
 include::building.adoc[]
 
-include::performance-tool.adoc[]
+include::performance-tool.adoc[]
+
+include::appendixes.adoc[]

src/main/java/com/rabbitmq/stream/Codec.java

@@ -1,4 +1,4 @@
-// Copyright (c) 2020-2021 VMware, Inc. or its affiliates.  All rights reserved.
+// Copyright (c) 2020-2023 VMware, Inc. or its affiliates.  All rights reserved.
 //
 // This software, the RabbitMQ Stream Java client library, is dual-licensed under the
 // Mozilla Public License 2.0 ("MPL"), and the Apache License version 2 ("ASL").

src/main/java/com/rabbitmq/stream/EnvironmentBuilder.java

@@ -222,6 +222,8 @@ public interface EnvironmentBuilder {
    */
   EnvironmentBuilder metricsCollector(MetricsCollector metricsCollector);
 
+  EnvironmentBuilder observationCollector(ObservationCollector<?> observationCollector);
+
   /**
    * The maximum number of producers allocated to a single connection.
    *

src/main/java/com/rabbitmq/stream/Message.java

@@ -1,4 +1,4 @@
-// Copyright (c) 2020-2022 VMware, Inc. or its affiliates.  All rights reserved.
+// Copyright (c) 2020-2023 VMware, Inc. or its affiliates.  All rights reserved.
 //
 // This software, the RabbitMQ Stream Java client library, is dual-licensed under the
 // Mozilla Public License 2.0 ("MPL"), and the Apache License version 2 ("ASL").
@@ -85,4 +85,31 @@ public interface Message {
    * @return the message annotations
    */
   Map<String, Object> getMessageAnnotations();
+
+  /**
+   * Add a message annotation to the message.
+   *
+   * @param key the message annotation key
+   * @param value the message annotation value
+   * @return the modified message
+   * @since 0.12.0
+   */
+  default Message annotate(String key, Object value) {
+    this.getMessageAnnotations().put(key, value);
+    return this;
+  }
+
+  /**
+   * Create a copy of the message.
+   *
+   * <p>The message copy contains the exact same instances of the original bare message (body,
+   * properties, application properties), only the message annotations are actually copied and can
+   * be modified independently.
+   *
+   * @return the message copy
+   * @since 0.12.0
+   */
+  default Message copy() {
+    return this;
+  }
 }

src/main/java/com/rabbitmq/stream/ObservationCollector.java

@@ -0,0 +1,83 @@
+// Copyright (c) 2023 VMware, Inc. or its affiliates.  All rights reserved.
+//
+// This software, the RabbitMQ Stream Java client library, is dual-licensed under the
+// Mozilla Public License 2.0 ("MPL"), and the Apache License version 2 ("ASL").
+// For the MPL, please see LICENSE-MPL-RabbitMQ. For the ASL,
+// please see LICENSE-APACHE2.
+//
+// This software is distributed on an "AS IS" basis, WITHOUT WARRANTY OF ANY KIND,
+// either express or implied. See the LICENSE file for specific language governing
+// rights and limitations of this software.
+//
+// If you have any questions regarding licensing, please contact us at
+// [email protected].
+package com.rabbitmq.stream;
+
+/**
+ * API to instrument operations in the stream client. The supported operations are publishing, and
+ * asynchronous delivery.
+ *
+ * <p>Implementations can gather information and send it to tracing backends. This allows e.g.
+ * following the processing steps of a given message through different systems.
+ *
+ * <p>This is considered an SPI and is susceptible to change at any time.
+ *
+ * @since 0.12.0
+ * @see EnvironmentBuilder#observationCollector(ObservationCollector)
+ * @see com.rabbitmq.stream.observation.micrometer.MicrometerObservationCollectorBuilder
+ */
+public interface ObservationCollector<T> {
+
+  ObservationCollector<Void> NO_OP =
+      new ObservationCollector<Void>() {
+        @Override
+        public Void prePublish(String stream, Message message) {
+          return null;
+        }
+
+        @Override
+        public void published(Void context, Message message) {}
+
+        @Override
+        public MessageHandler subscribe(MessageHandler handler) {
+          return handler;
+        }
+      };
+
+  /**
+   * Start observation.
+   *
+   * <p>Implementations are expecting to return an observation context that will be passed in to the
+   * {@link #published(Object, Message)} callback.
+   *
+   * @param stream the stream to publish to
+   * @param message the message to publish
+   * @return observation context
+   */
+  T prePublish(String stream, Message message);
+
+  /**
+   * Callback when the message is about to be published.
+   *
+   * @param context the observation context
+   * @param message the message to publish
+   */
+  void published(T context, Message message);
+
+  /**
+   * Decorate consumer registration.
+   *
+   * @param handler the original handler
+   * @return a decorated handler
+   */
+  MessageHandler subscribe(MessageHandler handler);
+
+  /**
+   * Says whether the implementation does nothing or not.
+   *
+   * @return true if the implementation is a no-op
+   */
+  default boolean isNoop() {
+    return this == NO_OP;
+  }
+}

src/main/java/com/rabbitmq/stream/codec/QpidProtonCodec.java

@@ -1,4 +1,4 @@
-// Copyright (c) 2020-2022 VMware, Inc. or its affiliates.  All rights reserved.
+// Copyright (c) 2020-2023 VMware, Inc. or its affiliates.  All rights reserved.
 //
 // This software, the RabbitMQ Stream Java client library, is dual-licensed under the
 // Mozilla Public License 2.0 ("MPL"), and the Apache License version 2 ("ASL").
@@ -32,6 +32,7 @@
 import org.apache.qpid.proton.codec.WritableBuffer;
 
 public class QpidProtonCodec implements Codec {
+
   private static final Function<String, String> MESSAGE_ANNOTATIONS_STRING_KEY_EXTRACTOR = k -> k;
   private static final Function<Symbol, String> MESSAGE_ANNOTATIONS_SYMBOL_KEY_EXTRACTOR =
       Symbol::toString;
@@ -52,7 +53,7 @@ private static Map<String, Object> createMessageAnnotations(
       return createMapFromAmqpMap(
           MESSAGE_ANNOTATIONS_SYMBOL_KEY_EXTRACTOR, message.getMessageAnnotations().getValue());
     } else {
-      return null;
+      return new LinkedHashMap<>();
     }
   }
 
@@ -258,7 +259,7 @@ public Message decode(byte[] data) {
         createMessageAnnotations(message));
   }
 
-  protected Properties createProperties(org.apache.qpid.proton.message.Message message) {
+  protected static Properties createProperties(org.apache.qpid.proton.message.Message message) {
     if (message.getProperties() != null) {
       return new QpidProtonProperties(message.getProperties());
     } else {
@@ -504,6 +505,21 @@ public Map<String, Object> getApplicationProperties() {
     public Map<String, Object> getMessageAnnotations() {
       return messageAnnotations;
     }
+
+    @Override
+    public Message annotate(String key, Object value) {
+      this.messageAnnotations.put(key, value);
+      return this;
+    }
+
+    @Override
+    public Message copy() {
+      return new QpidProtonMessage(
+          message,
+          createProperties(message),
+          createApplicationProperties(message),
+          createMessageAnnotations(message));
+    }
   }
 
   static class QpidProtonAmqpMessageWrapper implements Message {
@@ -579,6 +595,38 @@ public Map<String, Object> getMessageAnnotations() {
         return null;
       }
     }
+
+    @Override
+    public Message annotate(String key, Object value) {
+      MessageAnnotations annotations = this.message.getMessageAnnotations();
+      if (annotations == null) {
+        annotations = new MessageAnnotations(new LinkedHashMap<>());
+        this.message.setMessageAnnotations(annotations);
+      }
+      annotations.getValue().put(Symbol.getSymbol(key), value);
+      return this;
+    }
+
+    @Override
+    public Message copy() {
+      org.apache.qpid.proton.message.Message copy =
+          org.apache.qpid.proton.message.Message.Factory.create();
+      copy.setProperties(this.message.getProperties());
+      copy.setBody(this.message.getBody());
+      copy.setApplicationProperties(this.message.getApplicationProperties());
+      if (this.message.getMessageAnnotations() != null) {
+        Map<Symbol, Object> annotations = message.getMessageAnnotations().getValue();
+        Map<Symbol, Object> annotationCopy;
+        if (annotations == null) {
+          annotationCopy = null;
+        } else {
+          annotationCopy = new LinkedHashMap<>(annotations.size());
+          annotationCopy.putAll(annotations);
+        }
+        copy.setMessageAnnotations(new MessageAnnotations(annotationCopy));
+      }
+      return new QpidProtonAmqpMessageWrapper(this.hasPublishingId, this.publishingId, copy);
+    }
   }
 
   // from

src/main/java/com/rabbitmq/stream/codec/QpidProtonMessageBuilder.java

@@ -1,4 +1,4 @@
-// Copyright (c) 2020-2021 VMware, Inc. or its affiliates.  All rights reserved.
+// Copyright (c) 2020-2023 VMware, Inc. or its affiliates.  All rights reserved.
 //
 // This software, the RabbitMQ Stream Java client library, is dual-licensed under the
 // Mozilla Public License 2.0 ("MPL"), and the Apache License version 2 ("ASL").

src/main/java/com/rabbitmq/stream/codec/SimpleCodec.java

@@ -1,4 +1,4 @@
-// Copyright (c) 2020-2021 VMware, Inc. or its affiliates.  All rights reserved.
+// Copyright (c) 2020-2023 VMware, Inc. or its affiliates.  All rights reserved.
 //
 // This software, the RabbitMQ Stream Java client library, is dual-licensed under the
 // Mozilla Public License 2.0 ("MPL"), and the Apache License version 2 ("ASL").