Merge pull request #13 from factorhouse/feature/factor-platform

Add TargetProduct enum

Author: wavejumper

CHANGELOG.md

@@ -1,6 +1,14 @@
 # Change Log
 All notable changes to this project will be documented in this file. This change log follows the conventions of [keepachangelog.com](http://keepachangelog.com/).
 
+## [1.1.0] - 2026-04-22
+
+## Changed
+
+- Added new enum `io.factorhouse.kpow.StreamsRegistry.TargetProduct`: supported values are `KPOW` and `FACTOR_PLATFORM`
+- `io.factorhouse.kpow.StreamsRegistry` now accepts an optional third argument where you can specify the target product.
+- For backwards compatibility, the existing 2-artity `io.factorhouse.kpow.StreamsRegistry` constructor will initialize the `StreamsRegistry` instance with a default target product of `KPOW`.
+
 ## [1.0.0] - 2025-02-27
 
 A major release of the StreamsAgent with some breaking changes.

project.clj

@@ -1,4 +1,4 @@
-(defproject io.factorhouse/kpow-streams-agent "1.0.0"
+(defproject io.factorhouse/kpow-streams-agent "1.1.0"
   :description "Kpow's Kafka Streams monitoring agent"
   :url "https://github.com/factorhouse/kpow-streams-agent"
   :license {:name         "Apache-2.0 License"
@@ -23,10 +23,10 @@
                    [:roles
                     [:role "developer"]
                     [:role "maintainer"]]]])
-  :dependencies [[org.clojure/clojure "1.12.0"]
-                 [com.cognitect/transit-clj "1.0.333"]
-                 [org.clojure/tools.logging "1.3.0"]
-                 [org.apache.kafka/kafka-streams "3.7.1" :scope "provided"]]
+  :dependencies [[org.clojure/clojure "1.12.4"]
+                 [com.cognitect/transit-clj "1.1.357"]
+                 [org.clojure/tools.logging "1.3.1"]
+                 [org.apache.kafka/kafka-streams "4.0.2"]]
   :pom-plugins [[org.sonatype.central/central-publishing-maven-plugin "0.6.0"
                  {:extensions    "true"
                   :configuration [:publishingServerId "central"]}]
@@ -50,11 +50,11 @@
             :aot        :all}
   :profiles {:kaocha     {:dependencies [[lambdaisland/kaocha "1.91.1392"]]}
              :dev        {:resource-paths ["dev-resources"]
-                          :plugins        [[lein-cljfmt "0.9.2"]]
-                          :dependencies   [[org.slf4j/slf4j-api "2.0.16"]
-                                           [ch.qos.logback/logback-classic "1.3.14"]
-                                           [cheshire "5.13.0" :exclusions [com.fasterxml.jackson.core/jackson-databind]]
-                                           [clj-kondo "2024.09.27"]]}
+                          :plugins        [[dev.weavejester/lein-cljfmt "0.16.4"]]
+                          :dependencies   [[org.slf4j/slf4j-api "2.0.17"]
+                                           [ch.qos.logback/logback-classic "1.5.32"]
+                                           [cheshire "6.2.0"]
+                                           [clj-kondo "2026.04.15"]]}
              :smoke      {:pedantic? :abort}}
   :aliases {"kaocha" ["with-profile" "+kaocha" "run" "-m" "kaocha.runner"]
             "kondo"  ["with-profile" "+smoke" "run" "-m" "clj-kondo.main" "--lint" "src"]

src/clojure/io/factorhouse/kpow/agent.clj

@@ -13,9 +13,6 @@
                                      TopologyDescription$Node TopologyDescription$Processor TopologyDescription$Sink
                                      TopologyDescription$Source TopologyDescription$Subtopology)))
 
-(def kpow-snapshot-topic
-  {:topic "__oprtr_snapshot_state"})
-
 (extend-protocol p/Datafiable
   Taxon
   (datafy [v]
@@ -249,8 +246,8 @@
   {})
 
 (defn init-registry
-  [producer metrics-filter]
-  (start-registry {:snapshot-topic kpow-snapshot-topic
+  [producer metrics-filter snapshot-topic]
+  (start-registry {:snapshot-topic {:topic snapshot-topic}
                    :producer       producer
                    :metrics-filter metrics-filter}))
 

src/java/io/factorhouse/kpow/StreamsRegistry.java

@@ -29,6 +29,25 @@
  */
 public class StreamsRegistry implements AutoCloseable {
 
+    /**
+     * Specifies the target Factor House product this StreamsRegistry instance should produce to.
+     */
+    public enum TargetProduct {
+        KPOW,
+        FACTOR_PLATFORM;
+
+        public String getInternalTopic() {
+            switch (this) {
+                case KPOW:
+                    return "__oprtr_snapshot_state";
+                case FACTOR_PLATFORM:
+                    return "fh_snapshots";
+                default:
+                    throw new IllegalStateException("Unexpected value: " + this);
+            }
+        }
+    }
+
     /**
      * Represents an agent for a registered Kafka Streams application.
      * Provides the unique identifier of the registered application.
@@ -119,6 +138,8 @@ public static Properties filterProperties(Properties props) {
      * used to configure the underlying Kafka producer, and the {@link MetricFilter} determines which metrics
      * are collected and sent to Kpow's internal Kafka topic.</p>
      *
+     * <p>This method will construct an instance of the {@link StreamsRegistry} with a default target product of Kpow.</p>
+     *
      * <p><b>Important:</b> The Kafka producer properties provided in {@code props} must match the connection
      * details of Kpow's primary cluster, where Kpow's internal topic resides.</p>
      *
@@ -129,6 +150,28 @@ public static Properties filterProperties(Properties props) {
      * @throws IllegalArgumentException if the provided {@code props} are invalid or incomplete.
      */
     public StreamsRegistry(Properties props, MetricFilter metricsFilter) {
+        this(props, metricsFilter, TargetProduct.KPOW);
+    }
+
+    /**
+     * Constructs a {@code StreamsRegistry} instance using the specified Kafka properties and metrics filter.
+     *
+     * <p>This constructor initializes the registry, allowing Kafka Streams applications to be registered
+     * for monitoring through Kpow's user interface and API. The provided {@link Properties} object is
+     * used to configure the underlying Kafka producer, and the {@link MetricFilter} determines which metrics
+     * are collected and sent to Kpow's internal Kafka topic.</p>
+     *
+     * <p><b>Important:</b> The Kafka producer properties provided in {@code props} must match the connection
+     * details of Kpow's primary cluster, where Kpow's internal topic resides.</p>
+     *
+     * @param props         the {@link Properties} object containing Kafka configuration.
+     *                      Must include essential properties like {@code bootstrap.servers}.
+     * @param metricsFilter the {@link MetricFilter} to customize which metrics are reported.
+     *                      Use {@link MetricFilter#defaultMetricFilter()} for default behavior.
+     * @param targetProduct the {@link TargetProduct} specifies which Factor House product should receive the agent's metrics.
+     * @throws IllegalArgumentException if the provided {@code props} are invalid or incomplete.
+     */
+    public StreamsRegistry(Properties props, MetricFilter metricsFilter, TargetProduct targetProduct) {
         IFn require = Clojure.var("clojure.core", "require");
         require.invoke(Clojure.read("io.factorhouse.kpow.agent"));
         IFn agentFn = Clojure.var("io.factorhouse.kpow.agent", "init-registry");
@@ -138,7 +181,7 @@ public StreamsRegistry(Properties props, MetricFilter metricsFilter) {
         Serializer valSerializer = (Serializer) serdesFn.invoke();
         Properties producerProps = filterProperties(props);
         KafkaProducer producer = new KafkaProducer<>(producerProps, keySerializer, valSerializer);
-        agent = agentFn.invoke(producer, metricsFilter);
+        agent = agentFn.invoke(producer, metricsFilter, targetProduct.getInternalTopic());
     }
 
     /**

test/io/factorhouse/agent_test.clj

@@ -82,7 +82,7 @@
 (deftest agent-test
   (let [records        (atom [])
         metrics-filter (-> (MetricFilter.) (.accept))
-        registry       (agent/init-registry (mock-producer records) metrics-filter)
+        registry       (agent/init-registry (mock-producer records) metrics-filter "__oprtr_snapshot_state")
         agent-id       (agent/register registry
                                        (mock-streams [(mock-metric "first.metric" "first" "mock metric" {} 1.0)
                                                       (mock-metric "application-id" "first" "mock metric" {"client-id" "abc123"} "xxx")
@@ -146,7 +146,7 @@
                            (.acceptNameStartsWith "first")
                            (.acceptNameStartsWith "rocksdb")
                            (.deny))
-        registry       (agent/init-registry (mock-producer records) metrics-filter)
+        registry       (agent/init-registry (mock-producer records) metrics-filter "__oprtr_snapshot_state")
         agent-id       (agent/register registry
                                        (mock-streams [(mock-metric "first.metric" "first" "mock metric" {} 1.0)
                                                       (mock-metric "rocksdb.foo" "first" "mock metric" {"client-id" "abc123"} 3.0)
@@ -202,4 +202,4 @@
 
     (is (agent/unregister registry agent))
 
-    (is (empty? (agent/close-registry registry)))))
+    (is (empty? (agent/close-registry registry)))))