[#74] Avoid useless IO operations

Author: agilelab-tmnd1991

core/src/main/scala/it/agilelab/darwin/manager/AvroSchemaManager.scala

@@ -170,6 +170,21 @@ abstract class AvroSchemaManager(connector: Connector, endianness: ByteOrder) ex
     }
   }
 
+  /** Extracts the schema from the avro single-object encoded in the input array.
+   *
+   * @param array avro single-object encoded array
+   * @return the schema ID extracted from the input data
+   */
+  def extractSchema(array: Array[Byte]): Either[Exception, Schema] = {
+    try {
+      val id = AvroSingleObjectEncodingUtils.extractId(array, endianness)
+      getSchema(id)
+        .toRight(new RuntimeException(s"Cannot find schema with id $id"))
+    } catch {
+      case ie: IllegalArgumentException => Left(ie)
+    }
+  }
+
   /** Extracts a [[SchemaPayloadPair]] that contains the Schema and the Avro-encoded payload
    *
    * @param avroSingleObjectEncoded a byte array of a Single-Object encoded payload

hbase/src/main/scala/it/agilelab/darwin/connector/hbase/HBaseConnector.scala

@@ -115,17 +115,22 @@ case class HBaseConnector(config: Config) extends Connector with Logging {
   }
 
   override def insert(schemas: Seq[(Long, Schema)]): Unit = {
-    log.debug(s"inserting ${schemas.size} schemas in HBase table $NAMESPACE_STRING:$TABLE_NAME_STRING")
-    val mutator = connection.getBufferedMutator(TABLE_NAME)
-    schemas.map { case (id, schema) =>
-      val put = new Put(Bytes.toBytes(id))
-      put.addColumn(CF, QUALIFIER_SCHEMA, Bytes.toBytes(schema.toString))
-      put.addColumn(CF, QUALIFIER_NAME, Bytes.toBytes(schema.getName))
-      put.addColumn(CF, QUALIFIER_NAMESPACE, Bytes.toBytes(schema.getNamespace))
-      put
-    }.foreach(mutator.mutate)
-    mutator.flush()
-    log.debug(s"insertion of schemas into $NAMESPACE_STRING:$TABLE_NAME_STRING successful")
+    if (schemas.nonEmpty) {
+
+      log.debug(s"inserting ${schemas.size} schemas in HBase table $NAMESPACE_STRING:$TABLE_NAME_STRING")
+      using(connection.getBufferedMutator(TABLE_NAME)) { mutator =>
+        schemas.map { case (id, schema) =>
+          val put = new Put(Bytes.toBytes(id))
+          put.addColumn(CF, QUALIFIER_SCHEMA, Bytes.toBytes(schema.toString))
+          put.addColumn(CF, QUALIFIER_NAME, Bytes.toBytes(schema.getName))
+          put.addColumn(CF, QUALIFIER_NAMESPACE, Bytes.toBytes(schema.getNamespace))
+          put
+        }.foreach(mutator.mutate)
+        mutator.flush()
+        log.debug(s"insertion of schemas into $NAMESPACE_STRING:$TABLE_NAME_STRING successful")
+      }
+    }
+
   }
 
   override def createTable(): Unit = {

postgres/src/main/scala/it/agilelab/darwin/connector/postgres/PostgresConnector.scala

@@ -61,9 +61,11 @@ class PostgresConnector(config: Config) extends Connector with PostgresConnectio
   }
 
   override def insert(schemas: Seq[(Long, Schema)]): Unit = {
-    MODE match {
-      case ExceptionDriven => insertExceptionDriven(schemas)
-      case OneTransaction => insertOneTransaction(schemas)
+    if (schemas.nonEmpty) {
+      MODE match {
+        case ExceptionDriven => insertExceptionDriven(schemas)
+        case OneTransaction => insertOneTransaction(schemas)
+      }
     }
   }
 
@@ -151,21 +153,26 @@ class PostgresConnector(config: Config) extends Connector with PostgresConnectio
 
   private def findSchemas(connection: Connection, ids: Seq[(Long, Schema)]): Map[Long, Schema] = {
 
-    val withIdx = ids.zipWithIndex
-    val statement = connection.prepareStatement(s"select * from $TABLE_NAME where id in " +
-      withIdx.map(_ => "?").mkString("(", ",", ")")
-    )
-    withIdx.foreach { case (f, idx) =>
-      statement.setLong(idx + 1, f._1)
-    }
-    using(statement.executeQuery()) { resultSet =>
-      val schemas = Map.newBuilder[Long, Schema]
-      while (resultSet.next()) {
-        val id = resultSet.getLong("id")
-        val schema = parser.parse(resultSet.getString("schema"))
-        schemas += (id -> schema)
+    if (ids.nonEmpty) {
+      val withIdx = ids.zipWithIndex
+      using(connection.prepareStatement(s"select * from $TABLE_NAME where id in " +
+        withIdx.map(_ => "?").mkString("(", ",", ")")
+      )) { statement =>
+        withIdx.foreach { case (f, idx) =>
+          statement.setLong(idx + 1, f._1)
+        }
+        using(statement.executeQuery()) { resultSet =>
+          val schemas = Map.newBuilder[Long, Schema]
+          while (resultSet.next()) {
+            val id = resultSet.getLong("id")
+            val schema = parser.parse(resultSet.getString("schema"))
+            schemas += (id -> schema)
+          }
+          schemas.result()
+        }
       }
-      schemas.result()
+    } else {
+      Map.empty
     }
   }