Repository: redpanda-data/connect (language: go)

Issue:
cached processor with deleted() 
I noticed that when using a `cached` processor if one of the `cached` child processors errors and deletes (drops) the message, benthos will error with a message saying: `Child processors failed: child processors resulted in zero messages`. For example, use this config and input:
```json
{ "question": "what is the best stream processing pipeline?", "answer": "benthos" }
```

```yaml
cache_resources:
  - label: dedupe
    memory:
      init_values: {}
input:
  stdin:
    codec: lines
pipeline:
  threads: 1
  processors:
    - branch:
        processors:
          - cached:
              key: 'dedupe::${! this.question }'
              cache: dedupe
              processors:
                # mocking root = if errored() { deleted() }
                - bloblang: root = deleted()
output:
  stdout:
    codec: lines
```

Will result in this output:
```
INFO Running main config from specified file       @service=benthos path=config.yaml
INFO Launching a benthos instance, use CTRL+C to close  @service=benthos
ERRO Child processors failed: child processors resulted in zero messages  @service=benthos label="" path=root.pipeline.processors.0
INFO Listening for HTTP requests at: http://0.0.0.0:4195  @service=benthos
{ "question": "what is the best stream processing pipeline?", "answer": "benthos" }
INFO Pipeline has terminated. Shutting down the service  @service=benthos
```

What's the right way to drop a message under the cached processor (like on an error)? Is the `child processors resulted in zero messages` error expected in this case? Thanks! 




Discussion / hints:
hey @praetorian-thendrickson, what's the branch processor there for? As I'm fairly sure that's where the problem is, a branch processor needs the number of messages to align at the end of the child processors. Also you don't need a `cached` processor for deduplication as there's a specific processor for that: https://www.benthos.dev/docs/components/processors/dedupe
Thanks for the response, the branch processor is for a `result_map` like the example here https://www.benthos.dev/docs/components/processors/cached#examples. I think I over simplified my sample config. But with the same config as in the example for cached (I'm using the cached to pull an event from the cache if it's already been processed, and then the result map to add some information to update the event from the cache) if the processor errors and the message gets dropped, then the `child processors resulted in zero messages` happens. Is there a better way to do the `cached` + `result_map` where the processor might drop a message? 
```
pipeline:
  processors:
    - branch:
        processors:
          - cached:
              key: '${! meta("kafka_topic") }-${! meta("kafka_partition") }'
              cache: foo_cache
              processors:
                - mapping: 'root = ""'
                - http:
                    url: http://example.com/enrichment/${! meta("kafka_topic") }/${! meta("kafka_partition") }
                    verb: GET
        result_map: 'root.enrichment = this'
```

Fix patch:
diff --git a/internal/impl/pure/processor_cached.go b/internal/impl/pure/processor_cached.go
index 31ae9e2315..15f4eb56ed 100644
--- a/internal/impl/pure/processor_cached.go
+++ b/internal/impl/pure/processor_cached.go
@@ -10,6 +10,7 @@ import (
 
 	"golang.org/x/sync/errgroup"
 
+	"github.com/benthosdev/benthos/v4/public/bloblang"
 	"github.com/benthosdev/benthos/v4/public/service"
 )
 
@@ -20,6 +21,10 @@ func newCachedProcessorConfigSpec() *service.ConfigSpec {
 		Summary("Cache the result of applying one or more processors to messages identified by a key. If the key already exists within the cache the contents of the message will be replaced with the cached result instead of applying the processors. This component is therefore useful in situations where an expensive set of processors need only be executed periodically.").
 		Description("The format of the data when stored within the cache is a custom and versioned schema chosen to balance performance and storage space. It is therefore not possible to point this processor to a cache that is pre-populated with data that this processor has not created itself.").
 		Field(service.NewStringField("cache").Description("The cache resource to read and write processor results from.")).
+		Field(service.NewBloblangField("skip_on").
+			Description("A condition that can be used to skip caching the results from the processors.").
+			Example("errored()").
+			Optional()).
 		Field(service.NewInterpolatedStringField("key").
 			Description("A key to be resolved for each message, if the key already exists in the cache then the cached result is used, otherwise the processors are applied and the result is cached under this key. The key could be static and therefore apply generally to all messages or it could be an interpolated expression that is potentially unique for each message.").
 			Example("my_foo_result").
@@ -97,6 +102,7 @@ type cachedProcessor struct {
 	key        *service.InterpolatedString
 	ttl        *time.Duration
 	processors []*service.OwnedProcessor
+	skipOn     *bloblang.Executor
 }
 
 func newCachedProcessorFromParsedConf(manager *service.Resources, conf *service.ParsedConfig) (proc *cachedProcessor, err error) {
@@ -125,6 +131,15 @@ func newCachedProcessorFromParsedConf(manager *service.Resources, conf *service.
 	}
 
 	proc.processors, err = conf.FieldProcessorList("processors")
+
+	if conf.Contains("skip_on") {
+		var skipOn *bloblang.Executor
+		if skipOn, err = conf.FieldBloblang("skip_on"); err != nil {
+			return nil, err
+		}
+		proc.skipOn = skipOn
+	}
+
 	return
 }
 
@@ -159,29 +174,39 @@ func (proc *cachedProcessor) Process(ctx context.Context, msg *service.Message)
 		return nil, err
 	}
 
+	shouldCache := true
 	var collapsedBatch service.MessageBatch
 	for _, b := range resultBatch {
+		skip, err := shouldSkip(b, proc.skipOn)
+		if err != nil {
+			return nil, err
+		}
+
+		shouldCache = shouldCache && !skip
+
 		collapsedBatch = append(collapsedBatch, b...)
 	}
 
-	// Any errors in creating a serialised batch or caching are non-fatal and
-	// should be logged but otherwise regarded as insignificant to the flowing
-	// messages.
-	result, err := cachedProcSerialiseBatch(collapsedBatch)
-	if err != nil {
-		proc.manager.Logger().Errorf("failed to serialise resulting batch for caching: %w", err)
-		return collapsedBatch, nil
-	}
+	if shouldCache {
+		// Any errors in creating a serialised batch or caching are non-fatal and
+		// should be logged but otherwise regarded as insignificant to the flowing
+		// messages.
+		result, err := cachedProcSerialiseBatch(collapsedBatch)
+		if err != nil {
+			proc.manager.Logger().Errorf("failed to serialise resulting batch for caching: %w", err)
+			return collapsedBatch, nil
+		}
 
-	var setErr error
-	cerr := proc.manager.AccessCache(ctx, proc.cacheName, func(cache service.Cache) {
-		setErr = cache.Set(ctx, cacheKey, result, proc.ttl)
-	})
-	if cerr != nil {
-		proc.manager.Logger().Errorf("failed to access cache for result: %w", err)
-	}
-	if setErr != nil {
-		proc.manager.Logger().Errorf("failed to write result to cache: %w", err)
+		var setErr error
+		cerr := proc.manager.AccessCache(ctx, proc.cacheName, func(cache service.Cache) {
+			setErr = cache.Set(ctx, cacheKey, result, proc.ttl)
+		})
+		if cerr != nil {
+			proc.manager.Logger().Errorf("failed to access cache for result: %w", err)
+		}
+		if setErr != nil {
+			proc.manager.Logger().Errorf("failed to write result to cache: %w", err)
+		}
 	}
 
 	return collapsedBatch, nil
@@ -199,6 +224,29 @@ func (proc *cachedProcessor) Close(ctx context.Context) error {
 	return group.Wait()
 }
 
+func shouldSkip(batch service.MessageBatch, predicate *bloblang.Executor) (bool, error) {
+	if predicate == nil {
+		return false, nil
+	}
+
+	predResult, err := batch.BloblangQuery(0, predicate)
+	if err != nil {
+		return false, fmt.Errorf("failed to execute skip_on mapping: %w", err)
+	}
+
+	raw, err := predResult.AsStructured()
+	if err != nil {
+		return false, fmt.Errorf("skip_on mapping did not return structured result: %w", err)
+	}
+
+	skip, ok := raw.(bool)
+	if !ok {
+		return false, fmt.Errorf("skip_on did return boolean result: %v", raw)
+	}
+
+	return skip, nil
+}
+
 //------------------------------------------------------------------------------
 
 // We use versioning for the bytes we write to the cache, this allows us to
diff --git a/website/docs/components/processors/cached.md b/website/docs/components/processors/cached.md
index 702979c0c8..6068b666ca 100644
--- a/website/docs/components/processors/cached.md
+++ b/website/docs/components/processors/cached.md
@@ -26,6 +26,7 @@ Introduced in version 4.3.0.
 label: ""
 cached:
   cache: ""
+  skip_on: ""
   key: ""
   ttl: ""
   processors: []
@@ -33,49 +34,6 @@ cached:
 
 The format of the data when stored within the cache is a custom and versioned schema chosen to balance performance and storage space. It is therefore not possible to point this processor to a cache that is pre-populated with data that this processor has not created itself.
 
-## Fields
-
-### `cache`
-
-The cache resource to read and write processor results from.
-
-
-Type: `string`  
-
-### `key`
-
-A key to be resolved for each message, if the key already exists in the cache then the cached result is used, otherwise the processors are applied and the result is cached under this key. The key could be static and therefore apply generally to all messages or it could be an interpolated expression that is potentially unique for each message.
-This field supports [interpolation functions](/docs/configuration/interpolation#bloblang-queries).
-
-
-Type: `string`  
-
-```yml
-# Examples
-
-key: my_foo_result
-
-key: ${! this.document.id }
-
-key: ${! meta("kafka_key") }
-
-key: ${! meta("kafka_topic") }
-```
-
-### `ttl`
-
-An optional expiry period to set for each cache entry. Some caches only have a general TTL and will therefore ignore this setting.
-
-
-Type: `string`  
-
-### `processors`
-
-The list of processors whose result will be cached.
-
-
-Type: `array`  
-
 ## Examples
 
 <Tabs defaultValue="Cached Enrichment" values={[
@@ -140,4 +98,60 @@ cache_resources:
 </TabItem>
 </Tabs>
 
+## Fields
+
+### `cache`
+
+The cache resource to read and write processor results from.
+
+
+Type: `string`  
+
+### `skip_on`
+
+A condition that can be used to skip caching the results from the processors.
+
+
+Type: `string`  
+
+```yml
+# Examples
+
+skip_on: errored()
+```
+
+### `key`
+
+A key to be resolved for each message, if the key already exists in the cache then the cached result is used, otherwise the processors are applied and the result is cached under this key. The key could be static and therefore apply generally to all messages or it could be an interpolated expression that is potentially unique for each message.
+This field supports [interpolation functions](/docs/configuration/interpolation#bloblang-queries).
+
+
+Type: `string`  
+
+```yml
+# Examples
+
+key: my_foo_result
+
+key: ${! this.document.id }
+
+key: ${! meta("kafka_key") }
+
+key: ${! meta("kafka_topic") }
+```
+
+### `ttl`
+
+An optional expiry period to set for each cache entry. Some caches only have a general TTL and will therefore ignore this setting.
+
+
+Type: `string`  
+
+### `processors`
+
+The list of processors whose result will be cached.
+
+
+Type: `array`  
+
 


Test patch:
diff --git a/internal/impl/pure/processor_cached_test.go b/internal/impl/pure/processor_cached_test.go
index ed4bd67c74..763f23df83 100644
--- a/internal/impl/pure/processor_cached_test.go
+++ b/internal/impl/pure/processor_cached_test.go
@@ -14,6 +14,7 @@ func TestCachedHappy(t *testing.T) {
 	conf, err := newCachedProcessorConfigSpec().ParseYAML(`
 key: ${! content() }
 cache: foo
+skip_on: errored()
 processors:
   - bloblang: 'root = content().string() + " FOO " + uuid_v4()'
   - bloblang: 'root = content().string() + " BAR " + uuid_v4()'
@@ -146,3 +147,52 @@ processors:
 
 	assert.NoError(t, proc.Close(tCtx))
 }
+
+func TestCachedSkip(t *testing.T) {
+	conf, err := newCachedProcessorConfigSpec().ParseYAML(`
+key: ${! content() }
+cache: foo
+skip_on: "errored()"
+processors:
+  - bloblang: |
+      let body = content().string()
+      root = "%s FOO %d".format($body, count($body))
+  - bloblang: root = throw("simulated error")
+`, nil)
+	require.NoError(t, err)
+
+	mRes := service.MockResources(
+		service.MockResourcesOptAddCache("foo"),
+		service.MockResourcesOptAddCache("bar"),
+	)
+
+	proc, err := newCachedProcessorFromParsedConf(mRes, conf)
+	require.NoError(t, err)
+
+	tCtx := context.Background()
+
+	resBatchA1, err := proc.Process(tCtx, service.NewMessage([]byte("keya")))
+	require.NoError(t, err)
+	require.Len(t, resBatchA1, 1)
+
+	resBatchA2, err := proc.Process(tCtx, service.NewMessage([]byte("keya")))
+	require.NoError(t, err)
+	require.Len(t, resBatchA2, 1)
+
+	resBytesA1, err := resBatchA1[0].AsBytes()
+	require.NoError(t, err)
+	assert.Equal(t, string(resBytesA1), "keya FOO 1")
+
+	resBytesA2, err := resBatchA2[0].AsBytes()
+	require.NoError(t, err)
+	// Since caching is skipped, the second message runs through processors
+	assert.Equal(t, string(resBytesA2), "keya FOO 2")
+
+	require.NoError(t, mRes.AccessCache(tCtx, "foo", func(c service.Cache) {
+		val, err := c.Get(tCtx, "keya")
+		assert.ErrorIs(t, err, service.ErrKeyNotFound)
+		assert.Empty(t, val)
+	}))
+
+	assert.NoError(t, proc.Close(tCtx))
+}