package Ġio . l enses . jdbc 4 . queries Ċ Ċ import Ġio . kot lint est . matchers . collections . shouldBe Empty Ċ import Ġio . kot lint est . matchers . collections . should Have Size Ċ import Ġio . kot lint est . shouldBe Ċ import Ġio . kot lint est . specs . Word Spec Ċ import Ġio . l enses . jdbc 4 . L enses Driver Ċ import Ġio . l enses . jdbc 4 . Producer Setup Ċ import Ġio . l enses . jdbc 4 . result set . toList Ċ import Ġjava . sql . Driver Manager Ċ Ċ class ĠPoll Topic Via Jdbc Test Ġ: ĠWord Spec (), ĠProducer Setup Ġ{ ĊĊĠ Ġinit Ġ{ ĊĊĠĠĠ ĠL enses Driver () ĊĊĠĠĠ Ġval Ġconn Ġ= Ġconn () ĊĊĊĠĠĠ Ġ" JDBC ĠDriver " Ġshould Ġ{ ĊĠĠĠĠĠ Ġ" read Ġa Ġtopic Ġvia Ġpolling " Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġtopic Ġ= Ġnew Topic Name () ĊĠĠĠĠĠĠĠ Ġconn . createStatement (). executeQuery (""" ĊĠĠĠĠĠĠĠĠĠ ĠCREATE ĠTABLE Ġ$ topic (_ key Ġstring , Ġname Ġstring , Ġdifficulty Ġint ) ĠFORMAT Ġ( Av ro , ĠAv ro ); ĊĠĠĠĠĠĠĠ Ġ""" . trim Indent ()) ĊĊĠĠĠĠĠĠĠ Ġvar Ġlast Offset Ġ= Ġ 0 ĊĠĠĠĠĠĠĠ Ġval Ġsql 1 Ġ= Ġ" SELECT Ġ* ĠFROM Ġ$ topic ĠWHERE Ġ_ meta . partition = 0 Ġand Ġ_ meta . offset Ġ>= Ġ$ last Offset " ĊĠĠĠĠĠĠĠ Ġconn . createStatement (). executeQuery ( sql 1 ). toList (). should Have Size ( 0 ) ĊĊĊĠĠĠĠĠĠĠ Ġconn . createStatement (). executeQuery (""" ĊĠĠĠĠĠĠĠĠĠ ĠINSERT ĠINTO Ġ$ topic (_ key , Ġname , Ġdifficulty ) ĠĊĠĠĠĠĠĠĠĠĠ ĠVALUES (" 1 ", Ġ" Learn ĠL enses ĠSQL ", Ġ 3 ), ĊĠĠĠĠĠĠĠĠĠ Ġ(" 2 ", Ġ" Learn ĠQuant um ĠPhysics ", Ġ 1 0 ); ĊĠĠĠĠĠĠĠ Ġ""" . trim Indent ()) ĊĊĠĠĠĠĠĠĠ Ġval Ġsql 2 Ġ= Ġ" SELECT Ġ*, Ġ_ meta . offset Ġas Ġ_ offset , Ġ_ key Ġas Ġkey ĠFROM Ġ$ topic ĠWHERE Ġ_ meta . partition = 0 Ġand Ġ_ meta . offset Ġ>= Ġ$ last Offset " ĊĠĠĠĠĠĠĠ Ġval Ġrs Ġ= Ġconn . createStatement (). executeQuery ( sql 2 ) ĊĠĠĠĠĠĠĠ Ġval Ġlist Ġ= Ġmutable ListOf < Any >() ĊĠĠĠĠĠĠĠ Ġwhile Ġ( rs . next ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠ Ġval Ġkey Ġ= Ġrs . getString (" key ") ĊĠĠĠĠĠĠĠĠĠ Ġval Ġname Ġ= Ġrs . getString (" name ") ĊĠĠĠĠĠĠĠĠĠ Ġval Ġdifficulty Ġ= Ġrs . getInt (" difficulty ") ĊĠĠĠĠĠĠĠĠĠ Ġval Ġoffset Ġ= Ġrs . getInt ("_ offset ") ĊĠĠĠĠĠĠĠĠĠ Ġlist . add ( listOf ( key , Ġname , Ġdifficulty , Ġoffset )) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġlist . should Have Size ( 2 ) ĊĠĠĠĠĠĠĠ Ġlist ĠshouldBe ĠlistOf ( ĊĠĠĠĠĠĠĠĠĠĠĠ ĠlistOf (" 1 ", Ġ" Learn ĠL enses ĠSQL ", Ġ 3 , Ġ 0 ), ĊĠĠĠĠĠĠĠĠĠĠĠ ĠlistOf (" 2 ", Ġ" Learn ĠQuant um ĠPhysics ", Ġ 1 0 , Ġ 1 ) ĊĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠ Ġlast Offset Ġ= Ġ 1 ĊĊĠĠĠĠĠĠĠ Ġval Ġsql 3 Ġ= Ġ" SELECT Ġ*, Ġ_ meta . offset Ġas Ġ_ offset , Ġ_ key Ġas Ġkey ĠFROM Ġ$ topic ĠWHERE Ġ_ meta . partition = 0 Ġand Ġ_ meta . offset Ġ> Ġ$ last Offset " ĊĠĠĠĠĠĠĠ Ġconn . createStatement (). executeQuery ( sql 3 ). toList (). shouldBe Empty () ĊĠĠĠĠĠĠĠ Ġconn . createStatement (). executeQuery (""" ĊĠĠĠĠĠĠĠĠĠ ĠINSERT ĠINTO Ġ$ topic (_ key , Ġname , Ġdifficulty ) ĠĊĠĠĠĠĠĠĠĠĠ ĠVALUES (" 3 ", Ġ" Learn ĠFrench ", Ġ 5 ); ĊĠĠĠĠĠĠĠ Ġ""" . trim Indent ()) ĊĊĠĠĠĠĠĠĠ Ġval Ġsql 4 Ġ= Ġ" SELECT Ġ*, Ġ_ meta . offset Ġas Ġ_ offset , Ġ_ key Ġas Ġkey ĠFROM Ġ$ topic ĠWHERE Ġ_ meta . partition = 0 Ġand Ġ_ meta . offset Ġ> Ġ$ last Offset " ĊĠĠĠĠĠĠĠ Ġval Ġrs 2 Ġ= Ġconn . createStatement (). executeQuery ( sql 4 ) ĊĠĠĠĠĠĠĠ Ġval Ġone Record List Ġ= Ġmutable ListOf < Any >() ĊĠĠĠĠĠĠĠ Ġwhile Ġ( rs 2 . next ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠ Ġval Ġkey Ġ= Ġrs 2 . getString (" key ") ĊĠĠĠĠĠĠĠĠĠ Ġval Ġname Ġ= Ġrs 2 . getString (" name ") ĊĠĠĠĠĠĠĠĠĠ Ġval Ġdifficulty Ġ= Ġrs 2 . getInt (" difficulty ") ĊĠĠĠĠĠĠĠĠĠ Ġval Ġoffset Ġ= Ġrs 2 . getInt ("_ offset ") ĊĠĠĠĠĠĠĠĠĠ Ġone Record List . add ( listOf ( key , Ġname , Ġdifficulty , Ġoffset )) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġone Record List . should Have Size ( 1 ) ĊĠĠĠĠĠĠĠ Ġone Record List ĠshouldBe ĠlistOf ( ĊĠĠĠĠĠĠĠĠĠĠĠ ĠlistOf ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" 3 ", Ġ" Learn ĠFrench ", Ġ 5 , Ġ 2 ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠ Ġlast Offset Ġ= Ġ 2 ĊĠĠĠĠĠĠĠ Ġval Ġsql 5 Ġ= Ġ" SELECT Ġ*, Ġ_ meta . offset Ġas Ġ_ offset , Ġ_ key Ġas Ġkey ĠFROM Ġ$ topic ĠWHERE Ġ_ meta . partition = 0 Ġand Ġ_ meta . offset Ġ> Ġ$ last Offset " ĊĠĠĠĠĠĠĠ Ġconn . createStatement (). executeQuery ( sql 5 ). toList (). shouldBe Empty () ĊĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĠ Ġ} Ċ }