I needed a sink to Postgres DB, so I started to build a custom Flink SinkFunction. As FlinkKafkaProducer implements TwoPhaseCommitSinkFunction, then I decided to do the same. As stated in O'Reilley's book *Stream Processing with Apache Flink*, you just need to implement the abstract methods, enable checkpointing and you're up to go. But what really happens when I run my code is that `commit` method is called only once, and it is called before `invoke`, what is totally unexpected since you shouldn't be ready to commit if your set of ready-to-commit transactions is empty. And the worst is that, after committing, `invoke` is called for all of the transaction lines present in my file, and then `abort` is called, which is even more unexpected.

When the Sink is initialized, It is of my understanding that the following should occur:

1. beginTransaction is called and sends an identifier to invoke
2. invoke adds the lines to the transaction, according to the identifier received
3. pre-commit makes all final modification on current transaction data
4. commit handles the finalized transaction of pre-commited data

So, I can't see why my program doesn't show this behaviour.

Here goes my sink code:

```
package PostgresConnector

import java.sql.{BatchUpdateException, DriverManager, PreparedStatement, SQLException, Timestamp}
import java.text.ParseException
import java.util.{Date, Properties, UUID}
import org.apache.flink.api.common.ExecutionConfig
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.sink.{SinkFunction, TwoPhaseCommitSinkFunction}
import org.apache.flink.streaming.api.scala._
import org.slf4j.{Logger, LoggerFactory}

class PostgreSink(props : Properties, config : ExecutionConfig) extends TwoPhaseCommitSinkFunction[(String,String,String,String),String,String](createTypeInformation[String].createSerializer(config),createTypeInformation[String].createSerializer(config)){

    private var transactionMap : Map[String,Array[(String,String,String,String)]] = Map()

    private var parsedQuery : PreparedStatement = _

    private val insertionString : String = "INSERT INTO mydb (field1,field2,point) values (?,?,point(?,?))"

    override def invoke(transaction: String, value: (String,String,String,String), context: SinkFunction.Context[_]): Unit = {

        val LOG = LoggerFactory.getLogger(classOf[FlinkCEPClasses.FlinkCEPPipeline])

        val res = this.transactionMap.get(transaction)

        if(res.isDefined){

            var array = res.get

            array = array ++ Array(value)

            this.transactionMap += (transaction -> array)

        }else{

            val array = Array(value)

            this.transactionMap += (transaction -> array)

        }

        LOG.info("\n\nPassing through invoke\n\n")

        ()

    }

    override def beginTransaction(): String = {

        val LOG: Logger = LoggerFactory.getLogger(classOf[FlinkCEPClasses.FlinkCEPPipeline])

        val identifier = UUID.randomUUID.toString

        LOG.info("\n\nPassing through beginTransaction\n\n")

        identifier

    }

    override def preCommit(transaction: String): Unit = {

        val LOG = LoggerFactory.getLogger(classOf[FlinkCEPClasses.FlinkCEPPipeline])

        try{

            val tuple : Option[Array[(String,String,String,String)]]= this.transactionMap.get(transaction)

            if(tuple.isDefined){

                tuple.get.foreach( (value : (String,String,String,String)) => {

                    LOG.info("\n\n"+value.toString()+"\n\n")

                    this.parsedQuery.setString(1,value._1)
                    this.parsedQuery.setString(2,value._2)
                    this.parsedQuery.setString(3,value._3)
                    this.parsedQuery.setString(4,value._4)
                    this.parsedQuery.addBatch()

                })

            }

        }catch{

            case e : SQLException =>
                LOG.info("\n\nError when adding transaction to batch: SQLException\n\n")

            case f : ParseException =>
                LOG.info("\n\nError when adding transaction to batch: ParseException\n\n")

            case g : NoSuchElementException =>
                LOG.info("\n\nError when adding transaction to batch: NoSuchElementException\n\n")

            case h : Exception =>
                LOG.info("\n\nError when adding transaction to batch: Exception\n\n")

        }

        this.transactionMap = this.transactionMap.empty

        LOG.info("\n\nPassing through preCommit...\n\n")
    }

    override def commit(transaction: String): Unit = {

        val LOG : Logger = LoggerFactory.getLogger(classOf[FlinkCEPClasses.FlinkCEPPipeline])

        if(this.parsedQuery != null) {
            LOG.info("\n\n" + this.parsedQuery.toString+ "\n\n")
        }

        try{

            this.parsedQuery.executeBatch
            val LOG : Logger = LoggerFactory.getLogger(classOf[FlinkCEPClasses.FlinkCEPPipeline])
            LOG.info("\n\nExecuting batch\n\n")

        }catch{

            case e : SQLException =>
                val LOG : Logger = LoggerFactory.getLogger(classOf[FlinkCEPClasses.FlinkCEPPipeline])
                LOG.info("\n\n"+"Error : SQLException"+"\n\n")

        }

        this.transactionMap = this.transactionMap.empty

        LOG.info("\n\nPassing through commit...\n\n")

    }

    override def abort(transaction: String): Unit = {

        val LOG : Logger = LoggerFactory.getLogger(classOf[FlinkCEPClasses.FlinkCEPPipeline])

        this.transactionMap = this.transactionMap.empty

        LOG.info("\n\nPassing through abort...\n\n")

    }

    override def open(parameters: Configuration): Unit = {

        val LOG: Logger = LoggerFactory.getLogger(classOf[FlinkCEPClasses.FlinkCEPPipeline])

        val driver = props.getProperty("driver")
        val url = props.getProperty("url")
        val user = props.getProperty("user")
        val password = props.getProperty("password")
        Class.forName(driver)
        val connection = DriverManager.getConnection(url + "?user=" + user + "&password=" + password)
        this.parsedQuery = connection.prepareStatement(insertionString)

        LOG.info("\n\nConfiguring BD conection parameters\n\n")
    }
}

```

And this is my main program:

```
package FlinkCEPClasses

import PostgresConnector.PostgreSink
import org.apache.flink.api.java.io.TextInputFormat
import org.apache.flink.api.java.utils.ParameterTool
import org.apache.flink.cep.PatternSelectFunction
import org.apache.flink.cep.pattern.conditions.SimpleCondition
import org.apache.flink.cep.scala.pattern.Pattern
import org.apache.flink.core.fs.{FileSystem, Path}
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.cep.scala.{CEP, PatternStream}
import org.apache.flink.streaming.api.functions.source.FileProcessingMode
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import java.util.Properties

import org.apache.flink.api.common.ExecutionConfig
import org.slf4j.{Logger, LoggerFactory}

class FlinkCEPPipeline {

  val LOG: Logger = LoggerFactory.getLogger(classOf[FlinkCEPPipeline])
  LOG.info("\n\nStarting the pipeline...\n\n")

  var env : StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

  env.enableCheckpointing(10)
  env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime)
  env.setParallelism(1)

  //var input : DataStream[String] = env.readFile(new TextInputFormat(new Path("/home/luca/Desktop/lines")),"/home/luca/Desktop/lines",FileProcessingMode.PROCESS_CONTINUOUSLY,1)

  var input : DataStream[String] = env.readTextFile("/home/luca/Desktop/lines").name("Raw stream")

  var tupleStream : DataStream[(String,String,String,String)] = input.map(new S2PMapFunction()).name("Tuple Stream")

  var properties : Properties = new Properties()

  properties.setProperty("driver","org.postgresql.Driver")
  properties.setProperty("url","jdbc:postgresql://localhost:5432/mydb")
  properties.setProperty("user","luca")
  properties.setProperty("password","root")

  tupleStream.addSink(new PostgreSink(properties,env.getConfig)).name("Postgres Sink").setParallelism(1)
  tupleStream.writeAsText("/home/luca/Desktop/output",FileSystem.WriteMode.OVERWRITE).name("File Sink").setParallelism(1)

  env.execute()

}

```

My S2PMapFunction code:

```
package FlinkCEPClasses

import org.apache.flink.api.common.functions.MapFunction

case class S2PMapFunction() extends MapFunction[String,(String,String,String,String)] {

    override def map(value: String): (String, String, String,String) = {

            var tuple = value.replaceAllLiterally("(","").replaceAllLiterally(")","").split(',')

            (tuple(0),tuple(1),tuple(2),tuple(3))

    }
}

```

My pipeline works like this: I read lines from a file, map them to a tuple of strings, and use the data inside the tuples to save them in a Postgres DB

If you want to simulate the data, just create a file with lines in a format like this:
`(field1,field2,pointx,pointy)`

### Edit

The execution order of the TwoPhaseCommitSinkFUnction's methods is the following:

```
Starting pipeline...
beginTransaction
preCommit
beginTransaction
commit
invoke
invoke
invoke
invoke
invoke
invoke
invoke
invoke
invoke
invoke
invoke
invoke
invoke
invoke
invoke
invoke
invoke
invoke
abort

```