Add safeguards for missing SMB impl from classpath

scio-smb/src/main/java/org/apache/beam/sdk/extensions/smb/AvroSortedBucketIO.java

@@ -41,6 +41,16 @@
 /** API for reading and writing Avro sorted-bucket files. */
 public class AvroSortedBucketIO {
   private static final String DEFAULT_SUFFIX = ".avro";
+
+  // make sure avro is part of the classpath
+  static {
+    try {
+      Class.forName("org.apache.avro.Schema");
+    } catch (ClassNotFoundException e) {
+      throw new MissingImplementationException("avro", e);
+    }
+  }
+
   /** Returns a new {@link Read} for Avro generic records. */
   public static Read<GenericRecord> read(TupleTag<GenericRecord> tupleTag, Schema schema) {
     return new AutoValue_AvroSortedBucketIO_Read.Builder<GenericRecord>()

scio-smb/src/main/java/org/apache/beam/sdk/extensions/smb/MissingImplementationException.java

@@ -0,0 +1,34 @@
+/*
+ * Copyright 2023 Spotify AB
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.beam.sdk.extensions.smb;
+
+import org.apache.commons.lang.StringUtils;
+
+public class MissingImplementationException extends RuntimeException {
+
+    public MissingImplementationException(String format, Throwable cause) {
+        super(buildMessage(format), cause);
+    }
+
+    private static String buildMessage(String format) {
+        return String.format(
+                "%s is not part of the classpath. Add the 'com.spotify:scio-%s' dependency",
+                StringUtils.capitalize(format),
+                format);
+    }
+
+}

scio-smb/src/main/java/org/apache/beam/sdk/extensions/smb/ParquetAvroSortedBucketIO.java

@@ -44,6 +44,17 @@
 /** API for reading and writing Parquet sorted-bucket files as Avro. */
 public class ParquetAvroSortedBucketIO {
   private static final String DEFAULT_SUFFIX = ".parquet";
+
+  // make sure parquet is part of the classpath
+  static {
+    try {
+      Class.forName("org.apache.parquet.schema.Types");
+    } catch (ClassNotFoundException e) {
+      throw new MissingImplementationException("parquet", e);
+    }
+  }
+
+
   /** Returns a new {@link Read} for Avro generic records. */
   public static Read<GenericRecord> read(TupleTag<GenericRecord> tupleTag, Schema schema) {
     return new AutoValue_ParquetAvroSortedBucketIO_Read.Builder<GenericRecord>()

scio-smb/src/main/java/org/apache/beam/sdk/extensions/smb/TensorFlowBucketIO.java

@@ -40,6 +40,15 @@
 public class TensorFlowBucketIO {
   private static final String DEFAULT_SUFFIX = ".tfrecord";
 
+  // make sure tensorflow is part of the classpath
+  static {
+    try {
+      Class.forName("org.tensorflow.proto.example.Example");
+    } catch (ClassNotFoundException e) {
+      throw new MissingImplementationException("tensorflow", e);
+    }
+  }
+
   /**
    * Returns a new {@link Read} for TensorFlow TFRecord files with TensorFlow {@link Example}
    * records.

scio-smb/src/main/scala/org/apache/beam/sdk/extensions/smb/ParquetTypeFileOperations.scala

@@ -33,9 +33,18 @@ import org.apache.parquet.hadoop.metadata.CompressionCodecName
 import java.nio.channels.{ReadableByteChannel, WritableByteChannel}
 
 object ParquetTypeFileOperations {
+
   val DefaultCompression = CompressionCodecName.ZSTD
   val DefaultConfiguration: Configuration = null
 
+  // make sure parquet is part of the classpath
+  try {
+    Class.forName("org.apache.parquet.schema.Types");
+  } catch {
+    case e: ClassNotFoundException =>
+      throw new MissingImplementationException("parquet", e);
+  }
+
   def apply[T: Coder: ParquetType](): ParquetTypeFileOperations[T] = apply(DefaultCompression)
 
   def apply[T: Coder: ParquetType](