[FEAT] [Firehose] Implement S3 buffered delivery for PutRecord / PutRecordBatch #1886

Service

Kinesis Data Firehose

API Action / Feature

PutRecord / PutRecordBatch — buffered delivery to an S3 destination

AWS Documentation

• PutRecord
• S3DestinationConfiguration
• BufferingHints

Why is this needed?

We are writing integration tests for a logging backend that puts structured JSON records into a Firehose delivery stream backed by S3. On real AWS, PutRecord buffers the record and flushes it to the S3 destination within BufferingHints.IntervalInSeconds seconds

On Floci 1.5.33, CreateDeliveryStream and PutRecord both succeed and return correct responses but no object ever appears in the destination S3 bucket, regardless of how long we wait. The delivery pipeline is effectively a black hole: records are accepted and then silently dropped.

This blocks any integration test that needs to assert on delivered content (record shape, key format, size guards, end-to-end logging round-trips).

Steps to reproduce

import boto3, time

ENDPOINT = "http://localhost:4566"   # adjust port as needed
REGION   = "us-east-1"
BUCKET   = "test-delivery-bucket"
STREAM   = "test-delivery-stream"
CREDS    = dict(
    aws_access_key_id="test",
    aws_secret_access_key="test",
    region_name=REGION,
    endpoint_url=ENDPOINT,
)

s3 = boto3.client("s3", **CREDS)
fh = boto3.client("firehose", **CREDS)

# 1. Create destination bucket
s3.create_bucket(Bucket=BUCKET)

# 2. Create delivery stream with 1-second buffering
fh.create_delivery_stream(
    DeliveryStreamName=STREAM,
    DeliveryStreamType="DirectPut",
    S3DestinationConfiguration={
        "RoleARN": "arn:aws:iam::000000000000:role/firehose_delivery_role",
        "BucketARN": f"arn:aws:s3:::{BUCKET}",
        "BufferingHints": {"SizeInMBs": 1, "IntervalInSeconds": 1},
    },
)

# 3. Put a record
fh.put_record(
    DeliveryStreamName=STREAM,
    Record={"Data": b'{"hello":"world"}'},
)
print("Record put. Waiting 30 s for delivery...")

# 4. Poll S3 — objects never appear
for i in range(30):
    time.sleep(1)
    objects = s3.list_objects_v2(Bucket=BUCKET).get("Contents", [])
    if objects:
        print(f"Delivered after {i+1}s: {[o['Key'] for o in objects]}")
        break
else:
    print("No delivery after 30 s")  # ← always reached

Expected: at least one S3 object appears within a few seconds of IntervalInSeconds=1.

Actual: No delivery after 30 s — the bucket stays empty indefinitely.

Container logs

After PutRecord the logs are completely silent. There is no flush attempt, no S3 write, and no error:

INFO  [FirehoseService] Created Firehose delivery stream: test-delivery-stream
INFO  [AwsJson11Controller] AwsJson11Controller firehose action: PutRecord

No Flushed {0} records from stream {1} to s3://{2}/{3} line ever appears, even though that log message is compiled into the binary (confirmed via strings).

For comparison, immediately after a Kinesis PutRecord the logs show ListShards, GetShardIterator, and GetRecords activity — the Kinesis pipeline is fully end-to-end. Firehose produces nothing.

Binary analysis

Inspecting the native binary (/app/application) with strings shows:

• FirehoseService$1 is present — an anonymous inner class, which in Quarkus/Java typically represents a scheduled Runnable.
• Failed to flush Firehose stream {0}: {1} is present — error-path logging for the flush.
• Flushed {0} records from stream {1} to s3://{2}/{3} is present — success-path logging for the flush.

However, floci.storage.services.firehose.flush-interval-ms is absent from the binary, while the equivalent key exists for every other stateful service (dynamodb, kinesis, cloudwatchlogs, sns, ssm, secretsmanager, etc.). This suggests the scheduled flush task exists in the code but is either not wired into the Quarkus scheduler at startup, or the per-stream timer started in createDeliveryStream does not start correctly.

A secondary hypothesis is that the internal S3 client used by the flush task resolves to s3.amazonaws.com instead of the local endpoint, causing the delivery to fail silently (no log line because the failure is swallowed before reaching the Failed to flush log path).

Environment

Additional context

PR #1857 fixes the S3 object key format to match real AWS. The PR description references FirehoseExtendedS3IntegrationTest with delivery-plane tests, which implies delivery was expected to work at that point.

Are you willing to contribute a PR?

• Yes
• No