Issue with Lambda event source mapping creation using a Kinesis stream consumer ARN about localstack HOT 4 OPEN

Thank you @AbdullahMakhdoom for your investigation!

Upon further examination of the code, I initially believed that your suggestion would solve the issue concerning describing the target stream using a stream consumer ARN, as you re-assign a valid stream_arn post retrieval from the stream consumer ARN.

However, upon closer inspection, I've discovered that the current implementation of the _monitor_stream_event_sources method still retrieves data from the stream shards using the standard interator method, rather than subscribing to shards using the stream consumer, as outlined in here.

It seems I misunderstood the problem initially, thinking it was just about describing the event source mapping target stream bug using an ARN that belongs to a stream consumer not a stream itself. However, it's actually more complex than that, involving differences in consuming strategies and their implementation.

from localstack.

Welcome to LocalStack! Thanks for reporting your first issue and our team will be working towards fixing the issue for you or reach out for more background information. We recommend joining our Slack Community for real-time help and drop a message to LocalStack Pro Support if you are a Pro user! If you are willing to contribute towards fixing this issue, please have a look at our contributing guidelines and our contributing guide.

from localstack.

I can see there is a function from localstack.services.kinesis.provider import find_stream_for_consumer, which returns stream_name. Maybe we can modify the function to return stream_arn as well.

def find_stream_for_consumer(consumer_arn):
    account_id = extract_account_id_from_arn(consumer_arn)
    region_name = extract_region_from_arn(consumer_arn)
    kinesis = connect_to(aws_access_key_id=account_id, region_name=region_name).kinesis
    for stream_name in kinesis.list_streams()["StreamNames"]:
        stream_arn = arns.kinesis_stream_arn(stream_name, account_id, region_name)
        for cons in kinesis.list_stream_consumers(StreamARN=stream_arn)["Consumers"]:
            if cons["ConsumerARN"] == consumer_arn:
[+]                return stream_name, stream_arn
    raise Exception("Unable to find stream for stream consumer %s" % consumer_arn)

Then use this function in to get stream_arn corresponsing to conusmer arn.

def _monitor_stream_event_sources(self, *args):
    # additional code not directly related to the issue at hand...
    try:
[+]        if len(stream_arn.split("/")[-1].split(":")) > 1:
[+]            consumer_arn = stream_arn
[+]             _, stream_arn = find_stream_for_consumer(consumer_arn)
        stream_description = self._get_stream_description(stream_client, stream_arn)
    except Exception as e:
        LOG.error(
            "Cannot describe target stream %s of event source mapping %s: %s",
            stream_arn,
            mapping_uuid,
            e,
        )
        continue
    # additional code not directly related to the issue at hand...

Let me know your thoughts @viren-nadkarni @mohllal
Would be glad to pursue a PR for this solution if it makes sense.

from localstack.

I just noticed that the Kinesis mock server utilized by Localstack doesn't support SubscribeToShard operation.

LocalStack is proxying the functionality, as a workaround, using the GetRecords operation. However, this approach differs from AWS functionality, which relies on the HTTP2 server push mechanism between the shard and the consumer.

Based on the above implementation, there's no behaviour difference between utilizing a dedicated fan-out stream consumer and a consumer as a Lambda event source mapping source on LocalStack.

This holds true even if we were able to modify the implementation of the StreamEventSourceListener class to accommodate the proxied subscribe_to_shard method.

from localstack.