### 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_
"""
        to be implemented by subclasses
        :returns: An AWS service client instance for communicating with the appropriate API
        """
        raise NotImplementedError
    def _get_stream_description(self, stream_client, stream_arn):
        """
        to be implemented by subclasses
        :returns: The stream description object returned by the client's describe_stream method
        """
        raise NotImplementedError
    def _get_shard_iterator(self, stream_client, stream_arn, shard_id, iterator_type):
        """
        to be implemented by subclasses
        :returns: The shard iterator object returned by the client's get_shard_iterator method
        """
        raise NotImplementedError
    def _create_lambda_event_payload(
        self, stream_arn: str, records: List[Dict], shard_id: Optional[str] = None
    ) -> Dict:
        """
        to be implemented by subclasses
        Get an event payload for invoking a Lambda function using the given records and stream metadata
        :param stream_arn: ARN of the event source stream
        :param records: Batch of records to include in the payload, obtained from the source stream
        :param shard_id: ID of the shard the records came from. This is only needed for Kinesis event payloads.
        :returns: An event payload suitable for invoking a Lambda function
        """
        raise NotImplementedError
    def _get_starting_and_ending_sequence_numbers(