Cherry-pick #24167 to 7.12: [Filebeat][AWS] Fix vpcflow pipeline exception: Cannot invoke "Object.getClass()" because "receiver" is null

CHANGELOG.next.asciidoc

@@ -245,6 +245,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - Fix Logstash module handling of logstash.log.log_event.action field. {issue}20709[20709]
 - aws/s3access dataset was populating event.duration using the wrong unit. {pull}23920[23920]
 - Zoom module pipeline failed to ingest some chat_channel events. {pull}23904[23904]
+- Fix aws/vpcflow generating errors for empty logs or unidentified formats. {pull}24167[24167]
 
 *Heartbeat*
 

x-pack/filebeat/module/aws/vpcflow/ingest/pipeline.yml

@@ -119,7 +119,7 @@ processors:
       ignore_empty_value: true
 
   - set:
-      if: "ctx.aws.vpcflow.instance_id != '-'"
+      if: "ctx.aws?.vpcflow?.instance_id != null && ctx.aws.vpcflow.instance_id != '-'"
       field: cloud.instance.id
       value: "{{aws.vpcflow.instance_id}}"
       ignore_empty_value: true
@@ -131,11 +131,9 @@ processors:
   - script:
       lang: painless
       ignore_failure: true
+      if: "ctx.aws?.vpcflow?.tcp_flags != null"
       source: |
-        if (ctx?.aws?.vpcflow?.tcp_flags == null)
-          return;
-
-        if (ctx?.aws?.vpcflow?.tcp_flags_array == null) {
+        if (ctx.aws.vpcflow.tcp_flags_array == null) {
           ArrayList al = new ArrayList();
           ctx.aws.vpcflow.put("tcp_flags_array", al);
         }

x-pack/filebeat/module/aws/vpcflow/test/bad.log

@@ -0,0 +1 @@
+Phony unsupported log format.

x-pack/filebeat/module/aws/vpcflow/test/bad.log-expected.json

@@ -0,0 +1,18 @@
+[
+    {
+        "cloud.provider": "aws",
+        "event.category": "network_traffic",
+        "event.dataset": "aws.vpcflow",
+        "event.kind": "event",
+        "event.module": "aws",
+        "event.original": "Phony unsupported log format.",
+        "event.type": "flow",
+        "fileset.name": "vpcflow",
+        "input.type": "log",
+        "log.offset": 0,
+        "service.type": "aws",
+        "tags": [
+            "forwarded"
+        ]
+    }
+]