On publish failures, log error and count them as sys exceptions (apac…

…he#3704) * On publish failures, log error and count them as sys exceptions * Took feedback
bbejeck · Mar 5, 2019 · 07cebb1 · 07cebb1
1 parent 9dc3df6
commit 07cebb1
Show file tree

Hide file tree

Showing 3 changed files with 13 additions and 4 deletions.
diff --git a/...ar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/ContextImpl.java b/...ar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/ContextImpl.java
@@ -89,6 +89,8 @@ class ContextImpl implements Context, SinkContext, SourceContext {
     private StateContextImpl stateContext;
     private Map<String, Object> userConfigs;
 
+    private ComponentStatsManager statsManager;
+
     Map<String, String[]> userMetricsLabels = new HashMap<>();
     private final String[] metricsLabels;
     private final Summary userMetricsSummary;
@@ -103,12 +105,13 @@ class ContextImpl implements Context, SinkContext, SourceContext {
 
     public ContextImpl(InstanceConfig config, Logger logger, PulsarClient client, List<String> inputTopics,
                        SecretsProvider secretsProvider, CollectorRegistry collectorRegistry, String[] metricsLabels,
-                       Utils.ComponentType componentType) {
+                       Utils.ComponentType componentType, ComponentStatsManager statsManager) {
         this.config = config;
         this.logger = logger;
         this.publishProducers = new HashMap<>();
         this.inputTopics = inputTopics;
         this.topicSchema = new TopicSchema(client);
+        this.statsManager = statsManager;
 
         this.producerBuilder = (ProducerBuilderImpl<?>) client.newProducer().blockIfQueueFull(true).enableBatching(true)
                 .batchingMaxPublishDelay(1, TimeUnit.MILLISECONDS);
@@ -359,7 +362,13 @@ public <O> CompletableFuture<Void> publish(String topicName, O object, Schema<O>
             }
         }
 
-        return producer.sendAsync(object).thenApply(msgId -> null);
+        CompletableFuture<Void> future = producer.sendAsync(object).thenApply(msgId -> null);
+        future.exceptionally(e -> {
+            this.statsManager.incrSysExceptions(e);
+            logger.error("Failed to publish to topic {} with error {}", topicName, e);
+            return null;
+        });
+        return future;
     }
 
     @Override

diff --git a/...ons/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstanceRunnable.java b/...ons/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstanceRunnable.java
@@ -215,7 +215,7 @@ ContextImpl setupContext() {
         Logger instanceLog = LoggerFactory.getLogger(
                 "function-" + instanceConfig.getFunctionDetails().getName());
         return new ContextImpl(instanceConfig, instanceLog, client, inputTopics, secretsProvider,
-                collectorRegistry, metricsLabels, this.componentType);
+                collectorRegistry, metricsLabels, this.componentType, this.stats);
     }
 
     /**

diff --git a/...unctions/instance/src/test/java/org/apache/pulsar/functions/instance/ContextImplTest.java b/...unctions/instance/src/test/java/org/apache/pulsar/functions/instance/ContextImplTest.java
@@ -79,7 +79,7 @@ public void setup() {
             client,
             new ArrayList<>(),
             new EnvironmentBasedSecretsProvider(), new CollectorRegistry(), new String[0],
-                Utils.ComponentType.FUNCTION);
+                Utils.ComponentType.FUNCTION, null);
     }
 
     @Test(expectedExceptions = IllegalStateException.class)