Add delimiter and/or generic implementation of IDataProcessor for Kinesis Firehose

[LadislavMalecek]

Describe the feature

L2 DeliveryStream construct is missing a generic processor, only LambdaProcessor is available.
Add:

• GenericProcessor that can be filled in with custom ProcessorProperties that match CfnDeliveryStream.ProcessorProperty
• DelimiterProcessor.

DelimiterProcessor should implement:

type='AppendDelimiterToRecord',
parameters=[
    CfnDeliveryStream.ProcessorParameterProperty(
        parameter_name='Delimiter',
        parameter_value='\\n'
    )
]

Use Case

While using L2 DeliveryStream construct, you are only able to use the LambdaProcessor. We need to use the generic IDataProcessor to specify a delimiter to be inserted between the individual Kinesis records that are being processed and passed to S3.

Proposed Solution

No response

Other Information

No response

Acknowledgements

• I may be able to implement this feature request
• This feature might incur a breaking change

CDK version used

1.155

Environment details (OS name and version, etc.)

Python

[ChristophrK]

Hi,
I know this is a rather old issue but since it's still hard to find info on this I thought I update with my experience. You can add the DelimiterProcessor to the CfnDeliveryStream.ProcessorProperty Array of the CfnDeliveryStream.ProcessingConfigurationProperty inside CfnDeliveryStream.ExtendedS3DestinationConfigurationProperty. In the docs it says that Lambda would be the only valid value, but it also states that AppendDelimiterToRecord | Lambda | MetadataExtraction | RecordDeAggregation are allowed. The docs for CfnDeliveryStream.ExtendedS3DestinationConfigurationProperty can be found ExtendedS3DestinationConfiguration.

So atleast from my perspective this Issue could be closed.

[gioruffa]

Hi @ChristophrK ,
It is a very old issue, but I will give it my 2 cents since I stumbled upon the same problem.

I think that @LadislavMalecek is referring to the L2 constructs available in aws_kinesisfirehose_alpha and aws_kinesisfirehose_destinations_alpha while you are referring to the automatically generated L1 constructs available in aws_kinesisfirehose, which, indeed, supports the DelimiterProcessor because cloudformation does.

The L2 constructs come with many goodies, for example the automatic handling of grants between firehose and the source data stream. It would be nice to have support for the DelimiterProcessor in the alpha library since it is a very common use case, especially when consuming objects written by Firehose using Athena.

[Warsaalk]

I also spent a long time looking for a solution to this. The solution I found is far from ideal and should be used with caution. If you already have a processor defined within the firehose.S3Bucket class, the code below will overwrite that configuration.

I would recommend using it only for very specific cases.

class S3ProcessorAppendDelimiterToRecord implements IAspect {
    constructor(private readonly deliveryStreamName: string) {}

    public visit(node: IConstruct): void {
        if (node instanceof CfnDeliveryStream && node.deliveryStreamName === this.deliveryStreamName) {
    
            if (node.extendedS3DestinationConfiguration !== void 0) {
                // @ts-ignore -> This property is defined read-only, of course, but we want to override it.
                (node.extendedS3DestinationConfiguration as CfnDeliveryStream.ExtendedS3DestinationConfigurationProperty).processingConfiguration = {
                    enabled: true,
                    processors: [{
                        type: "AppendDelimiterToRecord",
                        parameters: [{
                            parameterName: "Delimiter",
                            parameterValue: "\\n"
                        }]
                    }]
                }
            }
        }
    }
}

declare const s3Destination: firehose.S3Bucket; // Your firehose S3 Bucket

const deliveryStreamName = "Stream1"
const deliveryStream = new firehose.DeliveryStream(scope, deliveryStreamName, {
	deliveryStreamName,
	destination: s3Destination
});

Aspects.of(scope).add(new S3ProcessorAppendDelimiterToRecord(deliveryStreamName));