feat: pending with hardcoded service URL (#26)

Signed-off-by: jacque1ine <js3fung@uwaterloo.ca>
Signed-off-by: Jacqueline Fung <fungjsl@gmail.com>

Author: jacque1ine

pipeline.yaml

@@ -4,11 +4,11 @@ metadata:
   name: simple-pipeline
 spec:
   limits: 
-    readBatchSize: 6
+    readBatchSize: 1
   vertices:
     - name: in
       scale:
-        min: 2
+        min: 1
       volumes: # Shared between containers that are part of the same pod, useful for sharing configurations
         - name: pulsar-config-volume
           configMap:

pulsar-config-map.yaml

@@ -15,8 +15,7 @@ data:
         consumer: 
           enabled: true
           consumerConfig: 
-            topicNames: "marchten"
-            subscriptionName: "sub"
+            topicNames: "marchtwelve"
 
 
 

src/main/java/io/numaproj/pulsar/config/PulsarConfig.java

@@ -3,6 +3,9 @@
 import org.apache.pulsar.client.api.PulsarClient;
 import org.apache.pulsar.client.api.PulsarClientException;
 import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.admin.PulsarAdmin;
+import org.apache.pulsar.client.admin.PulsarAdminException;
+import java.net.MalformedURLException;
 
 import java.util.Map;
 import java.util.UUID;
@@ -57,4 +60,12 @@ public Producer<byte[]> pulsarProducer(PulsarClient pulsarClient, PulsarProducer
                 .loadConf(producerConfig)
                 .create();
     }
-}
+
+    @Bean
+    public PulsarAdmin pulsarAdmin(PulsarClientProperties pulsarClientProperties) throws PulsarClientException {
+        return PulsarAdmin.builder()
+                .serviceHttpUrl("http://host.docker.internal:8080/")
+                // .authentication(...) if needed
+                .build();
+    }
+}

src/main/java/io/numaproj/pulsar/config/PulsarConsumerProperties.java

@@ -20,17 +20,31 @@
 @ConfigurationProperties(prefix = "spring.pulsar.consumer")
 @Slf4j
 public class PulsarConsumerProperties {
+
+    private static final String DEFAULT_SUBSCRIPTION_NAME = "sub";
+    
     private Map<String, Object> consumerConfig = new HashMap<>(); // Default to an empty map
 
     @PostConstruct
     public void init() {
         // Pulsar expects topicNames to be type set, but the configMap accepts a string
-        if (consumerConfig.containsKey("topicNames")) {
-            String topicName = (String) consumerConfig.remove("topicNames");
+        String topicNameKey = "topicNames";
+        if (consumerConfig.containsKey(topicNameKey)) {
+            String topicName = (String) consumerConfig.remove(topicNameKey);
             Set<String> topicNames = new HashSet<>();
             topicNames.add(topicName);
-            consumerConfig.put("topicNames", topicNames);
+            consumerConfig.put(topicNameKey, topicNames);
+        }
+
+        // If 'subscriptionName' not present, provide a default
+        String subscriptionNameKey = "subscriptionName";
+        if (!consumerConfig.containsKey(subscriptionNameKey)) {
+            consumerConfig.put(subscriptionNameKey, DEFAULT_SUBSCRIPTION_NAME);
+            log.info("No subscriptionName provided. Setting default: '{}'", DEFAULT_SUBSCRIPTION_NAME);
+        } else {
+            log.info("subscriptionName was already set, leaving as-is.");
         }
-        log.info("Consumer Config: " + consumerConfig);
+
+        log.info("Consumer Config: {}", consumerConfig);
     }
 }

src/main/java/io/numaproj/pulsar/consumer/PulsarSource.java

@@ -7,11 +7,16 @@
 import io.numaproj.numaflow.sourcer.ReadRequest;
 import io.numaproj.numaflow.sourcer.Server;
 import io.numaproj.numaflow.sourcer.Sourcer;
+import io.numaproj.pulsar.config.PulsarConsumerProperties;
 import lombok.extern.slf4j.Slf4j;
 
 import org.apache.pulsar.client.api.Consumer;
 import org.apache.pulsar.client.api.Messages;
 import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.admin.PulsarAdmin;
+import org.apache.pulsar.client.admin.PulsarAdminException;
+import org.apache.pulsar.common.policies.data.TopicStats;
+import org.apache.pulsar.common.policies.data.SubscriptionStats;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.stereotype.Component;
@@ -22,6 +27,7 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 
 @Slf4j
 @Component
@@ -36,6 +42,12 @@ public class PulsarSource extends Sourcer {
     @Autowired
     private PulsarConsumerManager pulsarConsumerManager;
 
+    @Autowired
+    private PulsarAdmin pulsarAdmin;
+
+    @Autowired
+    PulsarConsumerProperties pulsarConsumerProperties;
+
     @PostConstruct
     public void startServer() throws Exception {
         server = new Server(this);
@@ -124,9 +136,22 @@ public void ack(AckRequest request) {
 
     @Override
     public long getPending() {
-        // TO DO: Currently this is received but not acked. Should be num messages in
-        // backlog
-        return messagesToAck.size();
+        try {
+            // If changing to support multiple topics, need to update this
+            Set<String> topicNames = (Set<String>) pulsarConsumerProperties.getConsumerConfig().get("topicNames");
+            String topicName = (String) topicNames.iterator().next(); // Assumes there is only one topic name in the set
+            String subscriptionName = (String) pulsarConsumerProperties.getConsumerConfig().get("subscriptionName");
+
+            TopicStats topicStats = pulsarAdmin.topics().getStats(topicName);
+            SubscriptionStats subscriptionStats = topicStats.getSubscriptions().get(subscriptionName);
+            // will remove later - used for testing
+            log.info("Number of messages in the backlog: {}", subscriptionStats.getMsgBacklog()); 
+            return subscriptionStats.getMsgBacklog();
+        } catch (PulsarAdminException e) {
+            log.error("Error while fetching admin stats for pending messages", e);
+            // Return a negative value to indicate no pending information
+            return -1;
+        }
     }
 
     @Override