### 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_
Abstract class for listening to streams associated with event source mappings, batching data from those streams,
    and invoking the appropriate Lambda functions with those data batches.
    Because DynamoDB Streams and Kinesis Streams have similar but different APIs, this abstract class is useful for
    reducing repeated code. The various methods that must be implemented by inheriting subclasses essentially wrap
    client API methods or middleware-style operations on data payloads to compensate for the minor differences between
    these two services.
    """
    _COORDINATOR_THREAD: Optional[
        FuncThread
    ] = None  
    _STREAM_LISTENER_THREADS: Dict[
        str, FuncThread
    ] = {}  
    _POLL_INTERVAL_SEC: float = 1
    _FAILURE_PAYLOAD_DETAILS_FIELD_NAME = ""  
    @staticmethod
    def source_type() -> Optional[str]:
        """
        to be implemented by subclasses
        :returns: The type of event source this listener is associated with
        """
        
        return None
    def _get_matching_event_sources(self) -> List[Dict]:
        """
        to be implemented by subclasses
        :returns: A list of active Event Source Mapping objects (as dicts) that match the listener type
        """
        raise NotImplementedError
    def _get_stream_client(self, region_name: str):