diff --git a/examples/build.gradle.kts b/examples/build.gradle.kts
index ef6d35ca4a6..eeec1a5f9fa 100644
--- a/examples/build.gradle.kts
+++ b/examples/build.gradle.kts
@@ -4,7 +4,7 @@ plugins {
 
 allprojects {
     group = "aws.sdk.kotlin.example"
-    version = "1.0-SNAPSHOT"
+    version = "2.0-SNAPSHOT"
 
     repositories {
         maven {
diff --git a/examples/gradle.properties b/examples/gradle.properties
index a27c31caa6d..856b6c1aa15 100644
--- a/examples/gradle.properties
+++ b/examples/gradle.properties
@@ -1,4 +1,4 @@
 
 # AWS SDK
-awsSdkKotlinVersion=0.1.0
+awsSdkKotlinVersion=0.2.0
 
diff --git a/examples/s3-media-ingestion/build.gradle.kts b/examples/s3-media-ingestion/build.gradle.kts
new file mode 100644
index 00000000000..601108f4aba
--- /dev/null
+++ b/examples/s3-media-ingestion/build.gradle.kts
@@ -0,0 +1,11 @@
+plugins {
+    kotlin("jvm")
+}
+
+val awsSdkKotlinVersion: String by project
+
+dependencies {
+    implementation(kotlin("stdlib"))
+    implementation("org.jetbrains.kotlinx:kotlinx-coroutines-core:1.4.3")
+    implementation("aws.sdk.kotlin:s3:$awsSdkKotlinVersion")
+}
diff --git a/examples/s3-media-ingestion/src/main/kotlin/Main.kt b/examples/s3-media-ingestion/src/main/kotlin/Main.kt
new file mode 100644
index 00000000000..0c405aa2284
--- /dev/null
+++ b/examples/s3-media-ingestion/src/main/kotlin/Main.kt
@@ -0,0 +1,169 @@
+/*
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0.
+ */
+import aws.sdk.kotlin.services.s3.S3Client
+import aws.sdk.kotlin.services.s3.model.*
+import kotlinx.coroutines.flow.*
+import kotlinx.coroutines.runBlocking
+import software.aws.clientrt.content.ByteStream
+import software.aws.clientrt.content.fromFile
+import software.aws.clientrt.content.writeToFile
+import java.io.File
+import java.nio.file.Files
+
+/**
+ * This program reads media files from a specified directory and uploads media files to S3.
+ * After uploading it will then download uploaded files back into a local directory.
+ *
+ * Any file with the extension `.avi` will be processed.  To test create a text file and
+ * name it such that it matches the [filenameMetadataRegex] regex, ex:
+ * `title_2000.avi`.
+ *
+ * When running the sample adjust the following path constants as needed for your local environment.
+ */
+const val bucketName = "s3-media-ingestion-example"
+const val ingestionDirPath = "/tmp/media-in"
+const val completedDirPath = "/tmp/media-processed"
+const val failedDirPath = "/tmp/media-failed"
+const val downloadDirPath = "/tmp/media-down"
+
+// media metadata is extracted from filename: <title>_<year>.avi
+val filenameMetadataRegex = "([\\w\\s]+)_([\\d]+).avi".toRegex()
+
+fun main(): Unit = runBlocking {
+    val client = S3Client { region = "us-east-2" }
+
+    try {
+        // Setup
+        client.ensureBucketExists(bucketName)
+        listOf(completedDirPath, failedDirPath, downloadDirPath).forEach { validateDirectory(it) }
+        val ingestionDir = validateDirectory(ingestionDirPath)
+
+        // Upload files
+        val uploadResults = ingestionDir
+            .walk()
+            .asFlow()
+            .mapNotNull(::mediaMetadataExtractor)
+            .map { mediaMetadata -> client.uploadToS3(mediaMetadata) }
+            .toList()
+
+        moveFiles(uploadResults)
+
+        // Print results of operation
+        val (successes, failures) = uploadResults.partition { it is Success }
+        when (failures.isEmpty()) {
+            true -> println("Media uploaded successfully: $successes")
+            false -> println("Successfully uploaded: $successes \nFailed to upload: $failures")
+        }
+
+        // Download files to verify
+        client.listObjects(ListObjectsRequest { bucket = bucketName }).contents?.forEach { obj ->
+            client.getObject(GetObjectRequest { key = obj.key; bucket = bucketName }) { response ->
+                val outputFile = File(downloadDirPath, obj.key!!)
+                response.body?.writeToFile(outputFile).also { size ->
+                    println("Downloaded $outputFile ($size bytes) from S3")
+                }
+            }
+        }
+    } finally {
+        client.close()
+    }
+}
+
+/** Check for valid S3 configuration based on account */
+suspend fun S3Client.ensureBucketExists(bucketName: String) {
+    if (!bucketExists(bucketName)) {
+        createBucket(
+            CreateBucketRequest {
+                bucket = bucketName
+                createBucketConfiguration {
+                    locationConstraint = BucketLocationConstraint.UsEast2
+                }
+            }
+        )
+    }
+}
+
+/** Upload to S3 if file not already uploaded */
+suspend fun S3Client.uploadToS3(mediaMetadata: MediaMetadata): UploadResult {
+    if (keyExists(bucketName, mediaMetadata.s3KeyName))
+        return FileExistsError("${mediaMetadata.s3KeyName} already uploaded.", mediaMetadata)
+
+    return try {
+        putObject(
+            PutObjectRequest {
+                bucket = bucketName
+                key = mediaMetadata.s3KeyName
+                body = ByteStream.fromFile(mediaMetadata.file)
+                metadata = mediaMetadata.toMap()
+            }
+        )
+        Success("$bucketName/${mediaMetadata.s3KeyName}", mediaMetadata)
+    } catch (e: Exception) { // Checking Service Exception coming in future release
+        UploadError(e, mediaMetadata)
+    }
+}
+
+/** Determine if a object exists in a bucket */
+suspend fun S3Client.keyExists(s3bucket: String, s3key: String) =
+    try {
+        headObject(
+            HeadObjectRequest {
+                bucket = s3bucket
+                key = s3key
+            }
+        )
+        true
+    } catch (e: Exception) { // Checking Service Exception coming in future release
+        false
+    }
+
+/** Determine if a object exists in a bucket */
+suspend fun S3Client.bucketExists(s3bucket: String) =
+    try {
+        headBucket(HeadBucketRequest { bucket = s3bucket })
+        true
+    } catch (e: Exception) { // Checking Service Exception coming in future release
+        false
+    }
+
+/** Move files to directories based on upload results */
+fun moveFiles(uploadResults: List<UploadResult>) =
+    uploadResults
+        .map { uploadResult -> uploadResult.mediaMetadata.file.toPath() to (uploadResult is Success) }
+        .forEach { (file, uploadSuccess) ->
+            val targetFilePath = if (uploadSuccess) completedDirPath else failedDirPath
+            val targetPath = File(targetFilePath)
+            Files.move(file, File(targetPath, file.fileName.toString()).toPath())
+        }
+
+// Classes for S3 upload results
+sealed class UploadResult { abstract val mediaMetadata: MediaMetadata }
+data class Success(val location: String, override val mediaMetadata: MediaMetadata) : UploadResult()
+data class UploadError(val error: Throwable, override val mediaMetadata: MediaMetadata) : UploadResult()
+data class FileExistsError(val reason: String, override val mediaMetadata: MediaMetadata) : UploadResult()
+
+// Classes, properties, and functions for media metadata
+data class MediaMetadata(val title: String, val year: Int, val file: File)
+val MediaMetadata.s3KeyName get() = "$title-$year"
+fun MediaMetadata.toMap() = mapOf("title" to title, "year" to year.toString())
+fun mediaMetadataExtractor(file: File): MediaMetadata? {
+    if (!file.isFile || file.length() == 0L) return null
+
+    val matchResult = filenameMetadataRegex.find(file.name) ?: return null
+
+    val (title, year) = matchResult.destructured
+    return MediaMetadata(title, year.toInt(), file)
+}
+
+/** Validate file path and optionally create directory */
+fun validateDirectory(dirPath: String): File {
+    val dir = File(dirPath)
+
+    require(dir.isDirectory || !dir.exists()) { "Unable to use $dir" }
+
+    if (!dir.exists()) require(dir.mkdirs()) { "Unable to create $dir" }
+
+    return dir
+}
diff --git a/examples/settings.gradle.kts b/examples/settings.gradle.kts
index 0cfef73a710..90b1085dd58 100644
--- a/examples/settings.gradle.kts
+++ b/examples/settings.gradle.kts
@@ -1,3 +1,4 @@
 rootProject.name = "aws-sdk-kotlin-examples"
 
 include(":dynamodb-movies")
+include(":s3-media-ingestion")
\ No newline at end of file