diff --git a/src/main/kotlin/com/github/imflog/schema/registry/tasks/download/DownloadSubjectExtension.kt b/src/main/kotlin/com/github/imflog/schema/registry/tasks/download/DownloadSubjectExtension.kt
index c0a763e..3ed876b 100644
--- a/src/main/kotlin/com/github/imflog/schema/registry/tasks/download/DownloadSubjectExtension.kt
+++ b/src/main/kotlin/com/github/imflog/schema/registry/tasks/download/DownloadSubjectExtension.kt
@@ -44,7 +44,8 @@ data class DownloadSubject(
     val outputPath: String,
     val version: Int? = null,
     val regex: Boolean = false,
-    val outputFileName: String? = null
+    val outputFileName: String? = null,
+    val downloadReferences: Boolean = false
 )
 
 data class MetadataExtension(
@@ -52,4 +53,4 @@ data class MetadataExtension(
     val outputPath: String? = null
 ) {
     constructor(enabled: Boolean) : this(enabled, null)
-}
\ No newline at end of file
+}
diff --git a/src/main/kotlin/com/github/imflog/schema/registry/tasks/download/DownloadTaskAction.kt b/src/main/kotlin/com/github/imflog/schema/registry/tasks/download/DownloadTaskAction.kt
index f9b89ea..d94270d 100644
--- a/src/main/kotlin/com/github/imflog/schema/registry/tasks/download/DownloadTaskAction.kt
+++ b/src/main/kotlin/com/github/imflog/schema/registry/tasks/download/DownloadTaskAction.kt
@@ -36,14 +36,37 @@ class DownloadTaskAction(
                 val metadata = getSchemaMetadata(downloadSubject)
                 val outputDir = File(rootDir.toURI()).resolve(downloadSubject.outputPath)
                 outputDir.mkdirs()
-                if (metadataConfiguration.enabled) {
+                val metadataDirectory = if (metadataConfiguration.enabled) {
                     val metadataDirectory = metadataConfiguration.outputPath?.run {
                         File(rootDir.toURI()).resolve(this)
                     } ?: outputDir
                     metadataDirectory.mkdirs()
+                    metadataDirectory
+                } else {
+                    outputDir
+                }
+
+                if (metadataConfiguration.enabled) {
                     writeSchemaMetadata(downloadSubject, metadata, metadataDirectory)
                 }
                 writeSchemaFile(downloadSubject, metadata, outputDir)
+
+                if (downloadSubject.downloadReferences) {
+                    metadata.references.forEach {
+                        logger.infoIfNotQuiet("Start downloading referenced schema ${it.subject}/${it.version} for ${downloadSubject.subject}")
+                        val referenceSubject = DownloadSubject(
+                            subject = it.subject,
+                            outputPath = downloadSubject.outputPath,
+                            version = it.version
+                        )
+                        val referenceMetadata = getSchemaMetadata(referenceSubject)
+
+                        if (metadataConfiguration.enabled) {
+                            writeSchemaMetadata(referenceSubject, referenceMetadata, metadataDirectory)
+                        }
+                        writeSchemaFile(referenceSubject, referenceMetadata, outputDir)
+                    }
+                }
             } catch (e: Exception) {
                 logger.error("Error during schema retrieval for ${downloadSubject.subject}", e)
                 errorCount++
diff --git a/src/test/kotlin/com/github/imflog/schema/registry/tasks/download/DownloadTaskActionTest.kt b/src/test/kotlin/com/github/imflog/schema/registry/tasks/download/DownloadTaskActionTest.kt
index dd5b10f..7fbd913 100644
--- a/src/test/kotlin/com/github/imflog/schema/registry/tasks/download/DownloadTaskActionTest.kt
+++ b/src/test/kotlin/com/github/imflog/schema/registry/tasks/download/DownloadTaskActionTest.kt
@@ -3,6 +3,7 @@ package com.github.imflog.schema.registry.tasks.download
 import io.confluent.kafka.schemaregistry.avro.AvroSchema
 import io.confluent.kafka.schemaregistry.avro.AvroSchemaProvider
 import io.confluent.kafka.schemaregistry.client.MockSchemaRegistryClient
+import io.confluent.kafka.schemaregistry.client.rest.entities.SchemaReference
 import io.confluent.kafka.schemaregistry.json.JsonSchemaProvider
 import io.confluent.kafka.schemaregistry.protobuf.ProtobufSchemaProvider
 import org.assertj.core.api.Assertions
@@ -63,7 +64,7 @@ class DownloadTaskActionTest {
                 DownloadSubject(testSubject, outputDir),
                 DownloadSubject(fooSubject, outputDir)
             ),
-            MetadataExtension(),
+            MetadataExtension()
         ).run()
 
         // then
@@ -136,7 +137,7 @@ class DownloadTaskActionTest {
             arrayListOf(
                 DownloadSubject("te.*", outputDir, null, true)
             ),
-            MetadataExtension(),
+            MetadataExtension()
         ).run()
 
         // then
@@ -184,7 +185,7 @@ class DownloadTaskActionTest {
             registryClient,
             folderRoot,
             arrayListOf(DownloadSubject(subject, outputDir)),
-            MetadataExtension(),
+            MetadataExtension()
         ).run()
 
         // then
@@ -220,7 +221,7 @@ class DownloadTaskActionTest {
                 DownloadSubject(invalidSubjectPattern, outputDir, null, true),
                 DownloadSubject("test", outputDir)
             ),
-            MetadataExtension(),
+            MetadataExtension()
         ).run()
 
         // then
@@ -274,7 +275,7 @@ class DownloadTaskActionTest {
             arrayListOf(
                 DownloadSubject("test", outputDir, v1Id)
             ),
-            MetadataExtension(),
+            MetadataExtension()
         ).run()
 
         // Then
@@ -317,7 +318,7 @@ class DownloadTaskActionTest {
             registryClient,
             folderRoot,
             arrayListOf(DownloadSubject(testSubject, outputDir)),
-            MetadataExtension(true),
+            MetadataExtension(true)
         ).run()
 
         // then
@@ -365,7 +366,7 @@ class DownloadTaskActionTest {
             registryClient,
             folderRoot,
             arrayListOf(DownloadSubject(testSubject, outputDir)),
-            MetadataExtension(true, metadataDir),
+            MetadataExtension(true, metadataDir)
         ).run()
 
         // then
@@ -380,4 +381,149 @@ class DownloadTaskActionTest {
             .containsIgnoringCase("\"schema\" :")
             .containsIgnoringCase("\"references\" :")
     }
+
+    @Test
+    fun `Should download referenced schemas with metadata in the specified output dir`() {
+        // given
+        val testSubject = "test"
+        val testLibSubject = "test_lib"
+        val outputDir = "src/main/avro/external"
+        val metadataDir = "src/main/avro/metadata"
+
+        val registryClient = MockSchemaRegistryClient(listOf(AvroSchemaProvider()))
+
+        registryClient.register(
+            testLibSubject,
+            registryClient.parseSchema(
+                AvroSchema.TYPE,
+                """{
+                    "type": "record",
+                    "name": "test_lib",
+                    "fields": [
+                        { "name": "name", "type": "string" }
+                    ]
+                }""",
+                listOf()
+            ).get()
+        )
+
+        registryClient.register(
+            testSubject,
+            registryClient.parseSchema(
+                AvroSchema.TYPE,
+                """{
+                    "type": "record",
+                    "name": "test",
+                    "fields": [
+                        { "name": "name", "type": "test_lib" }
+                    ]
+                }""",
+                listOf(
+                    SchemaReference(
+                        "name",
+                        testLibSubject,
+                        1
+                    )
+                )
+            ).get()
+        )
+
+        folderRule.resolve("src/main/avro/external").toFile().mkdir()
+        val folderRoot = folderRule.toFile()
+
+        // when
+        val errorCount = DownloadTaskAction(
+            registryClient,
+            folderRoot,
+            arrayListOf(DownloadSubject(testSubject, outputDir, downloadReferences = true)),
+            MetadataExtension(true, metadataDir)
+        ).run()
+
+        // then
+        Assertions.assertThat(errorCount).isEqualTo(0)
+        Assertions.assertThat(File(folderRoot, "src/main/avro/external/test.avsc")).isNotNull
+        Assertions.assertThat(File(folderRoot, "src/main/avro/metadata/test-metadata.json")).isNotNull
+        // Would be cleaner to use a JSON assertion library but I am not sure this is really required for now
+        Assertions.assertThat(File(folderRoot, "src/main/avro/metadata/test-metadata.json").readText())
+            .containsIgnoringCase("\"id\" :")
+            .containsIgnoringCase("\"version\" :")
+            .containsIgnoringCase("\"schema_type\" :")
+            .containsIgnoringCase("\"schema\" :")
+            .containsIgnoringCase("\"references\" :")
+
+        Assertions.assertThat(File(folderRoot, "src/main/avro/external/test_lib.avsc")).isNotNull
+        Assertions.assertThat(File(folderRoot, "src/main/avro/metadata/test_lib-metadata.json")).isNotNull
+        // Would be cleaner to use a JSON assertion library but I am not sure this is really required for now
+        Assertions.assertThat(File(folderRoot, "src/main/avro/metadata/test_lib-metadata.json").readText())
+            .containsIgnoringCase("\"id\" :")
+            .containsIgnoringCase("\"version\" :")
+            .containsIgnoringCase("\"schema_type\" :")
+            .containsIgnoringCase("\"schema\" :")
+            .containsIgnoringCase("\"references\" :")
+    }
+
+    @Test
+    fun `Should download referenced schemas in the specified output dir`() {
+        // given
+        val testSubject = "test"
+        val testLibSubject = "test_lib"
+        val outputDir = "src/main/avro/external"
+        val metadataDir = "src/main/avro/metadata"
+
+        val registryClient = MockSchemaRegistryClient(listOf(AvroSchemaProvider()))
+
+        registryClient.register(
+            testLibSubject,
+            registryClient.parseSchema(
+                AvroSchema.TYPE,
+                """{
+                    "type": "record",
+                    "name": "test_lib",
+                    "fields": [
+                        { "name": "name", "type": "string" }
+                    ]
+                }""",
+                listOf()
+            ).get()
+        )
+
+        registryClient.register(
+            testSubject,
+            registryClient.parseSchema(
+                AvroSchema.TYPE,
+                """{
+                    "type": "record",
+                    "name": "test",
+                    "fields": [
+                        { "name": "name", "type": "test_lib" }
+                    ]
+                }""",
+                listOf(
+                    SchemaReference(
+                        "name",
+                        testLibSubject,
+                        1
+                    )
+                )
+            ).get()
+        )
+
+        folderRule.resolve("src/main/avro/external").toFile().mkdir()
+        val folderRoot = folderRule.toFile()
+
+        // when
+        val errorCount = DownloadTaskAction(
+            registryClient,
+            folderRoot,
+            arrayListOf(DownloadSubject(testSubject, outputDir, downloadReferences = true)),
+            MetadataExtension()
+        ).run()
+
+        // then
+        Assertions.assertThat(errorCount).isEqualTo(0)
+        Assertions.assertThat(File(folderRoot, "src/main/avro/external/test.avsc")).isNotNull
+
+        Assertions.assertThat(File(folderRoot, "src/main/avro/external/test_lib.avsc")).isNotNull
+        // Would be cleaner to use a JSON assertion library but I am not sure this is really required for now
+    }
 }