ack.txt

mysql> select * from ack;
+------+---------------------+------+
| n    | t                   | c    |
+------+---------------------+------+
|    1 | 2014-01-31 23:37:00 |    6 |
|    2 | 2014-01-31 23:38:00 |    6 |
|    3 | 2014-01-31 23:39:00 |    6 |
+------+---------------------+------+
3 rows in set (0.00 sec)

jdbc-river-ack-demo.sh

#!/bin/bash

# JDBC river 1.0.0.RC1.4 acknowledge demo

mysql=/usr/local/mysql/bin/mysql

# do not forget "create database test"

echo "drop table if exists orders" | $mysql test
echo "create table orders (customer varchar(32), department varchar(32), product varchar(32), quantity integer, created timestamp default '0000-00-00 00:00:00')"  | $mysql test
echo "INSERT INTO orders VALUES ('Big','American Fruits','Apples',1,'2013-02-25 23:36:46'),('Large','German Fruits','Bananas',1,'2013-02-25 23:36:46'),('Huge','German Fruits','Oranges',2,'2013-02-25 23:36:46'),('Good','German Fruits','Apples',2,'2012-06-01 13:52:25'),('Bad','English Fruits','Oranges',3,'2012-06-01 14:31:24'),('Small','American Fruits','Apples',1,'2013-02-25 23:50:12');" | $mysql test
echo "drop table if exists ack" | $mysql test
echo "create table ack ( n integer, t datetime, c integer)" | $mysql test

# start river, run each minute, select orders, put them by created column as _id into Elasticsearch, send acknowledge info back to DB

curl -XDELETE '0:9200/_river/my_jdbc_river/'

curl -XPUT '0:9200/_river/my_jdbc_river/_meta' -d '{
    "type" : "jdbc",
    "jdbc" : {
        "url" : "jdbc:mysql://localhost:3306/test",
        "user" : "",
        "password" : "",
        "schedule" : "0 0-59 0-23 ? * *",
        "sql" : [
            {
                "statement" : "select *, created as _id, \"myjdbc\" as _index, \"mytype\" as _type from orders"
            },
            {
                "statement" : "insert into ack(n,t,c) values(?,?,?)",
                "parameter" : [ "$job", "$now", "$count" ]
            }
        ]
    }
}'