Camel Components

AWS S3 - Streaming Upload

Streaming Upload mode

With the stream mode enabled, users will be able to upload data to S3 without knowing ahead of time the dimension of the data, by leveraging multipart upload. The upload will be completed when the batchSize has been completed or the batchMessageNumber has been reached. There are three possible naming strategies: progressive, random and timestamp. With the progressive strategy, each file will have the name composed by keyName option and a progressive counter, and eventually the file extension (if any), while with the random strategy a UUID will be added after keyName and eventually the file extension will be appended. With the timestamp strategy, the time the upload started (in milliseconds) is added after keyName instead of the UUID. With the random and timestamp strategies every file gets this part, the first file of the producer too, so a producer that is started again does not replace the files written before.

Additionally, streaming upload mode supports timestamp-based file grouping, which allows messages to be automatically grouped into time windows based on their timestamps.

As an example:

Java-only: Endpoint DSL builder style
from(kafka("topic1").brokers("localhost:9092"))
        .log("Kafka Message is: ${body}")
        .to(aws2S3("camel-bucket").streamingUploadMode(true).batchMessageNumber(25)
                .namingStrategy(AWS2S3EndpointBuilderFactory.AWSS3NamingStrategyEnum.progressive)
                .keyName("{{kafkaTopic1}}/{{kafkaTopic1}}.txt"));

from(kafka("topic2").brokers("localhost:9092"))
         .log("Kafka Message is: ${body}")
         .to(aws2S3("camel-bucket").streamingUploadMode(true).batchMessageNumber(25)
                 .namingStrategy(AWS2S3EndpointBuilderFactory.AWSS3NamingStrategyEnum.random)
                 .keyName("{{kafkaTopic2}}/{{kafkaTopic2}}.txt"));

The default size for a batch is 1 Mb, but you can adjust it according to your requirements.

When you stop your producer route, the producer will take care of flushing the remaining buffered message and complete the upload.

In Streaming upload, you’ll be able to restart the producer from the point where it left. It’s important to note that this feature is critical only when using the progressive naming strategy.

By setting the restartingPolicy to lastPart, you will restart uploading files and contents from the last part number the producer left.

As example: - Start the route with progressive naming strategy and keyname equals to camel.txt, with batchMessageNumber equals to 20, and restartingPolicy equals to lastPart - Send 70 messages. - Stop the route - On your S3 bucket you should now see four files: camel.txt, camel-1.txt, camel-2.txt and camel-3.txt, the first three will have 20 messages, while the last one is only 10. - Restart the route - Send 25 messages - Stop the route - You’ll now have two other files in your bucket: camel-5.txt and camel-6.txt, the first with 20 messages and the second with 5 messages. - Go ahead

This won’t be needed when using the random naming strategy.

On the opposite, you can specify the override restartingPolicy. In that case, you’ll be able to override whatever you written before (for that particular keyName) in your bucket.

In Streaming upload mode, the only keyName option that will be taken into account is the endpoint option. Using the header will throw an NPE and this is done by design. Setting the header means potentially change the file name on each exchange, and this is against the aim of the streaming upload producer. The keyName needs to be fixed and static. The selected naming strategy will do the rest of the work.

Another possibility is specifying a streamingUploadTimeout with batchMessageNumber and batchSize options. With this option, the user will be able to complete the upload of a file after a certain time passed. In this way, the upload completion will be passed on three tiers: the timeout, the number of messages and the batch size.

As an example:

Java-only: Endpoint DSL builder style
from(kafka("topic1").brokers("localhost:9092"))
        .log("Kafka Message is: ${body}")
        .to(aws2S3("camel-bucket").streamingUploadMode(true).batchMessageNumber(25)
                .streamingUploadTimeout(10000)
                .namingStrategy(AWS2S3EndpointBuilderFactory.AWSS3NamingStrategyEnum.progressive)
                .keyName("{{kafkaTopic1}}/{{kafkaTopic1}}.txt"));

In this case, the upload will be completed after 10 seconds.

Timestamp Grouping

The streaming upload mode also supports timestamp-based file grouping, which allows messages to be automatically grouped into time windows and written to the same S3 file based on their timestamps. This feature enables append-like behavior where messages with timestamps falling within the same time window are combined into a single file.

To enable timestamp grouping, use the following configuration options:

  • timestampGroupingEnabled: Set to true to enable timestamp-based grouping (default: false)

  • timestampWindowSizeMillis: The size of the time window in milliseconds (default: 300000 = 5 minutes)

  • timestampHeaderName: The name of the message header containing the timestamp (default: Exchange.MESSAGE_TIMESTAMP)

Messages are grouped based on their timestamps extracted from the specified header. The timestamp can be provided as:

  • Long: Unix timestamp in milliseconds

  • Date: Java Date object

  • String: String representation of Unix timestamp in milliseconds

Files are automatically named using a timestamp-based pattern that includes the time window information, for example grouped-messages_20240101_0800_0800-0805.txt for the window from 08:00 to 08:05 (UTC). The upload of a window is completed when it reaches batchMessageNumber messages or batchSize bytes, or when streamingUploadTimeout runs. When more messages of that window come after it, they are uploaded to a new object of the window, named with the number of the uploads of the window before it: grouped-messages_20240101_0800_0800-0805-1.txt, then -2, and so on. The first time a producer uploads a window, it lists the objects of the window in the bucket (one ListObjectsV2 request with the window’s key as prefix), so after a restart it continues after the highest number already stored instead of replacing that object. This needs the s3:ListBucket permission: without it the producer logs a WARN and numbers the uploads of the window from the first one, so an upload after a restart can replace an object of the window again.

These keys do not sort in upload order: …​0805-1.txt sorts before …​0805.txt (- sorts before .), and -10 sorts before -2. To read the uploads of a window in order, sort them by the number after the time range (none for the first upload).

Producers that run at the same time (for example two routes or two applications) with the same keyName and timestamp grouping still choose the same names for the same window, so one can replace the object of the other. Give each of them its own keyName.

Example Configuration

Basic timestamp grouping with 5-minute windows:

  • Java

  • XML

  • YAML

from("timer:messages?period=10000")
    .setHeader(Exchange.MESSAGE_TIMESTAMP, simple("${date:now:timestamp}"))
    .setBody(constant("Message with timestamp"))
    .to("aws2-s3://my-bucket?streamingUploadMode=true&timestampGroupingEnabled=true&timestampWindowSizeMillis=300000&keyName=grouped-messages.txt");
<route>
  <from uri="timer:messages?period=10000"/>
  <setHeader name="CamelMessageTimestamp">
    <simple>${date:now:timestamp}</simple>
  </setHeader>
  <setBody>
    <constant>Message with timestamp</constant>
  </setBody>
  <to uri="aws2-s3://my-bucket?streamingUploadMode=true&amp;timestampGroupingEnabled=true&amp;timestampWindowSizeMillis=300000&amp;keyName=grouped-messages.txt"/>
</route>
- route:
    from:
      uri: timer:messages
      parameters:
        period: 10000
      steps:
        - setHeader:
            name: CamelMessageTimestamp
            simple: "${date:now:timestamp}"
        - setBody:
            constant: Message with timestamp
        - to:
            uri: aws2-s3://my-bucket
            parameters:
              streamingUploadMode: true
              timestampGroupingEnabled: true
              timestampWindowSizeMillis: 300000
              keyName: grouped-messages.txt

Custom window size (1 minute) with custom header name:

  • Java

  • XML

  • YAML

from("direct:timestamped")
    .setHeader("MyTimestamp", simple("${date:now:timestamp}"))
    .setBody(constant("Custom timestamped message"))
    .to("aws2-s3://my-bucket?streamingUploadMode=true&timestampGroupingEnabled=true&timestampWindowSizeMillis=60000&timestampHeaderName=MyTimestamp&keyName=custom-grouped.txt");
<route>
  <from uri="direct:timestamped"/>
  <setHeader name="MyTimestamp">
    <simple>${date:now:timestamp}</simple>
  </setHeader>
  <setBody>
    <constant>Custom timestamped message</constant>
  </setBody>
  <to uri="aws2-s3://my-bucket?streamingUploadMode=true&amp;timestampGroupingEnabled=true&amp;timestampWindowSizeMillis=60000&amp;timestampHeaderName=MyTimestamp&amp;keyName=custom-grouped.txt"/>
</route>
- route:
    from:
      uri: direct:timestamped
      steps:
        - setHeader:
            name: MyTimestamp
            simple: "${date:now:timestamp}"
        - setBody:
            constant: Custom timestamped message
        - to:
            uri: aws2-s3://my-bucket
            parameters:
              streamingUploadMode: true
              timestampGroupingEnabled: true
              timestampWindowSizeMillis: 60000
              timestampHeaderName: MyTimestamp
              keyName: custom-grouped.txt

Large files with multipart and timestamp grouping:

  • Java

  • XML

  • YAML

from("direct:large-timestamped")
    .setHeader(Exchange.MESSAGE_TIMESTAMP, simple("${date:now:timestamp}"))
    .setBody(constant("Large message content..."))
    .to("aws2-s3://my-bucket?streamingUploadMode=true&timestampGroupingEnabled=true&timestampWindowSizeMillis=1800000&multiPartUpload=true&partSize=5242880&keyName=large-grouped.txt");
<route>
  <from uri="direct:large-timestamped"/>
  <setHeader name="CamelMessageTimestamp">
    <simple>${date:now:timestamp}</simple>
  </setHeader>
  <setBody>
    <constant>Large message content...</constant>
  </setBody>
  <to uri="aws2-s3://my-bucket?streamingUploadMode=true&amp;timestampGroupingEnabled=true&amp;timestampWindowSizeMillis=1800000&amp;multiPartUpload=true&amp;partSize=5242880&amp;keyName=large-grouped.txt"/>
</route>
- route:
    from:
      uri: direct:large-timestamped
      steps:
        - setHeader:
            name: CamelMessageTimestamp
            simple: "${date:now:timestamp}"
        - setBody:
            constant: "Large message content..."
        - to:
            uri: aws2-s3://my-bucket
            parameters:
              streamingUploadMode: true
              timestampGroupingEnabled: true
              timestampWindowSizeMillis: 1800000
              multiPartUpload: true
              partSize: 5242880
              keyName: large-grouped.txt

File Naming

Files are automatically named using the following pattern:

{baseFileName}_{YYYYMMDD}_{HHMM}_{HHMM-HHMM}.{extension}

For time windows smaller than 1 minute, seconds precision is used:

{baseFileName}_{YYYYMMDD}_{HHMMSS}_{HHMMSS-HHMMSS}.{extension}

Example file names: - 5-minute window: messages_20240101_0800_0800-0805.txt - 1-minute window: data_20240315_1430_1430-1431.log - 5-second window: events_20241225_000000_000000-000005.json

Time Window Examples

With a 5-minute window (300000ms): - Window 1: 08:00:00 - 08:04:59 → file_20240101_0800_0800-0805.txt - Window 2: 08:05:00 - 08:09:59 → file_20240101_0805_0805-0810.txt - Window 3: 08:10:00 - 08:14:59 → file_20240101_0810_0810-0815.txt

Fallback Behavior

  • If no timestamp header is found, the current system time is used

  • If the timestamp header contains invalid data, the current system time is used

  • Warning messages are logged when fallback behavior occurs

Performance Considerations

  • Multiple concurrent time windows are supported

  • Each window maintains its own upload state

  • Memory usage scales with the number of active time windows

  • Completed windows are automatically cleaned up

  • Works with all existing streaming upload features (multipart uploads, timeouts, etc.)