Added …Inlet.create Java API for errorHandler.

core/cloudflow-streamlets/src/main/scala/cloudflow/streamlets/avro/AvroInlet.scala

@@ -16,10 +16,11 @@
 
 package cloudflow.streamlets.avro
 
-import cloudflow.streamlets._
+import java.util.Optional
+import scala.reflect.ClassTag
 import org.apache.avro.specific.SpecificRecordBase
 
-import scala.reflect.ClassTag
+import cloudflow.streamlets._
 import AvroUtil._
 
 case class AvroInlet[T <: SpecificRecordBase: ClassTag](
@@ -40,6 +41,18 @@ object AvroInlet {
   def create[T <: SpecificRecordBase](name: String, clazz: Class[T]): AvroInlet[T] =
     AvroInlet[T](name)(ClassTag.apply(clazz))
 
+  // Java API
   def create[T <: SpecificRecordBase](name: String, clazz: Class[T], hasUniqueGroupId: Boolean): AvroInlet[T] =
     AvroInlet[T](name, hasUniqueGroupId)(ClassTag.apply(clazz))
+
+  // Java API
+  def create[T <: SpecificRecordBase](
+      name: String,
+      clazz: Class[T],
+      hasUniqueGroupId: Boolean,
+      errorHandler: (Array[Byte], Throwable) => Optional[T]): AvroInlet[T] =
+    AvroInlet[T](name, hasUniqueGroupId, (a, t) => {
+      val opt = errorHandler(a, t)
+      if (opt.isPresent) Some(opt.get()) else None
+    })(ClassTag.apply(clazz))
 }

core/cloudflow-streamlets/src/main/scala/cloudflow/streamlets/proto/javadsl/ProtoInlet.scala

@@ -15,7 +15,7 @@
  */
 
 package cloudflow.streamlets.proto.javadsl
-
+import java.util.Optional
 import cloudflow.streamlets._
 import com.google.protobuf.Descriptors.Descriptor
 import com.google.protobuf.{ GeneratedMessageV3, TextFormat }
@@ -39,10 +39,19 @@ final case class ProtoInlet[T <: GeneratedMessageV3](
 }
 
 object ProtoInlet {
-  // Java API
   def create[T <: GeneratedMessageV3](name: String, clazz: Class[T]): ProtoInlet[T] =
     ProtoInlet[T](name, clazz)
 
   def create[T <: GeneratedMessageV3](name: String, clazz: Class[T], hasUniqueGroupId: Boolean): ProtoInlet[T] =
     ProtoInlet[T](name, clazz, hasUniqueGroupId)
+
+  def create[T <: GeneratedMessageV3](
+      name: String,
+      clazz: Class[T],
+      hasUniqueGroupId: Boolean,
+      errorHandler: (Array[Byte], Throwable) => Optional[T]): ProtoInlet[T] =
+    ProtoInlet[T](name, clazz, hasUniqueGroupId, (a, t) => {
+      val opt = errorHandler(a, t)
+      if (opt.isPresent) Some(opt.get()) else None
+    })
 }