### Is there an existing issue for this?

- [X] I have searched the existing issues

### Feature description

AWS has a feature which allows you to [filter incoming lambda invocations using rules](https://docs.aws.amazon.com/lambda/latest/dg/invocation-eventfiltering.html). Without this feature I need to filter inside the lambda code itself which will incur more compute and cost in AWS. In the spirit of having as few diffs between localstack and AWS this is highly desireable for my pipeline.

(And fwiw, my pipeline is DynamoDB -> DynamoDB Stream -> Lambda (Python) -> Opensearch)

The current behavior is that create/update-event-source-mapping does parse (and validate) this option, but it isn't honored against incoming events at runtime.

### 🧑‍💻 Implementation

Implement with the same logic as AWS with support for all the operators in the linked article above.

### Anything else?

_No response_
asynchronous=not config.SYNCHRONOUS_KINESIS_EVENTS,
            lock_discriminator=lock_discriminator,
        )
        if isinstance(result, InvocationResult):
            status_code = getattr(result.result, "status_code", 0)
            if status_code >= 400:
                return False, status_code
            return True, status_code
        return False, 500
    def _listen_to_shard_and_invoke_lambda(self, params: Dict):
        """
        Continuously listens to a stream's shard. Divides records read from the shard into batches and use them to
        invoke a Lambda.
        This function is intended to be invoked as a FuncThread. Because FuncThreads can only take a single argument,
        we pack the numerous arguments needed to invoke this method into a single dictionary.
        :param params: Dictionary containing the following elements needed to execute this method:
            * function_arn: ARN of the Lambda function to invoke
            * stream_arn: ARN of the stream associated with the shard to listen on
            * batch_size: number of records to pass to the Lambda function per invocation
            * parallelization_factor: parallelization factor for executing lambda funcs asynchronously
            * lock_discriminator: discriminator for checking semaphore on lambda function execution. Also used for
                                  checking if this listener loops should continue to run.
            * shard_id: ID of the shard to listen on
            * stream_client: AWS service client for communicating with the stream API
            * shard_iterator: shard iterator object for iterating over records in stream
            * max_num_retries: maximum number of times to attempt invoking a batch against the Lambda before giving up
                               and moving on
            * failure_destination: Optional destination config for sending record metadata to if Lambda invocation fails
                                   more than max_num_retries
        """