Add a better worded exception of unsupported SpecificRecord coder case (

#4815)

* Add a better worded exception of unsupported SpecificRecord coder case

* Fix formatting

* Apply review comments

scio-core/src/main/scala/com/spotify/scio/coders/instances/AvroCoders.scala

@@ -128,11 +128,21 @@ trait AvroCoders {
 
   implicit def avroSpecificRecordCoder[T <: SpecificRecord: ClassTag]: Coder[T] = {
     val clazz = ScioUtil.classOf[T]
+
     // Try to get the schema with SpecificData.getSchema
     // This relies on private SCHEMA$ field that may not be defined on custom SpecificRecord instance
-    // Otherwise create a default instance and call getSchema
     val schema = Try(SpecificData.get().getSchema(clazz))
-      .getOrElse(clazz.getDeclaredConstructor().newInstance().getSchema)
+      // Otherwise create a default instance and call getSchema
+      .orElse(Try(clazz.getDeclaredConstructor().newInstance().getSchema))
+      .getOrElse {
+        val msg =
+          "Failed to create a coder for SpecificRecord because it is impossible to retrieve an " +
+            s"Avro schema by instantiating $clazz. Use only a concrete type implementing " +
+            s"SpecificRecord or use GenericRecord type in your transformations if a concrete " +
+            s"type is not known in compile time."
+        throw new RuntimeException(msg)
+      }
+
     val useReflectApi = true // keep this for backward compatibility
     Coder.beam(AvroCoder.of(clazz, schema, useReflectApi))
   }

scio-test/src/test/scala/com/spotify/scio/coders/AvroCoderTest.scala

@@ -20,6 +20,7 @@ package com.spotify.scio.coders
 import org.scalatest.flatspec.AnyFlatSpec
 import com.spotify.scio.testing.CoderAssertions._
 import org.apache.avro.generic.GenericRecord
+import org.apache.avro.specific.SpecificRecord
 import org.scalactic.Equality
 import org.scalatest.matchers.should.Matchers
 
@@ -29,6 +30,16 @@ final class AvroCoderTest extends AnyFlatSpec with Matchers {
     Avro.user coderShould notFallback()
   }
 
+  it should "support not Avro's SpecificRecord if a concrete type is not provided" in {
+    val caught = intercept[RuntimeException] {
+      Avro.user.asInstanceOf[SpecificRecord] coderShould notFallback()
+    }
+
+    caught.getMessage should startWith(
+      "Failed to create a coder for SpecificRecord because it is impossible to retrieve an Avro"
+    )
+  }
+
   it should "support avrohugger generated SpecificRecord" in {
     Avro.scalaSpecificAvro coderShould notFallback()
   }