Merge pull request #4 from mmolimar/develop

Release 0.2 version

Author: mmolimar

.travis.yml

@@ -1,8 +1,30 @@
 before_install:
+  - git clone https://github.com/confluentinc/common.git
+  - cd common
+  - git checkout v4.1.0-rc3
+  - mv pom.xml pom.xml.bak && sed  '58s/.*/        <kafka.version>1.1.0<\/kafka.version>/' pom.xml.bak > pom.xml
+  - mvn -B clean install -Dmaven.test.skip=true
+  - cd ..
+  - git clone https://github.com/confluentinc/support-metrics-common.git
+  - cd support-metrics-common
+  - git checkout v4.1.0-rc3
+  - mvn -B clean install -Dmaven.test.skip=true
+  - cd ..
+  - git clone https://github.com/confluentinc/rest-utils.git
+  - cd rest-utils
+  - git checkout v4.1.0-rc3
+  - mvn -B clean install -Dmaven.test.skip=true
+  - cd ..
+  - git clone https://github.com/confluentinc/schema-registry.git
+  - cd schema-registry
+  - git checkout v4.1.0-rc3
+  - mvn -B clean install -Dmaven.test.skip=true
+  - cd ..
   - git clone https://github.com/confluentinc/ksql.git
   - cd ksql
-  - git checkout 0.1.x
-  - mvn -pl '!ksql-cli' clean install -Dmaven.test.skip=true
+  - git checkout v4.1.0-rc3
+  - ex -sc '55d7|x' ksql-cli/pom.xml && ex -sc '37d7|x' ksql-engine/pom.xml && ex -sc '25d7|x' ksql-parser/pom.xml && ex -sc '84d7|x' ksql-rest-app/pom.xml && ex -sc '37d7|x' ksql-tools/pom.xml
+  - mvn -B clean install -Dmaven.test.skip=true
   - cd ..
 script:
   - sbt clean coverage test it:test coverageReport && sbt coverageAggregate

README.md

@@ -10,16 +10,22 @@ perform queries to Kafka and then, the engine translates those requests to Kafka
 
 ### Building from source ###
 
-First of all, the KSQL lib has to be installed into your local repo (till now, there isn't a release available).
+First of all, the KSQL lib has to be installed into your local repo.
 
 So, cloning the KSQL repo:
 
-``git clone https://github.com/confluentinc/ksql.git && cd ksql && git checkout 0.1.x``
+``git clone https://github.com/confluentinc/ksql.git && cd ksql && git v4.1.0-rc3``
 
 and installing it:
 
 ``mvn clean install -Dmaven.skip.test=true``
 
+Probably, you'll have to do the same things for these Confluent projects (previous to the KSQL project installation):
+* [Confluent common](https://github.com/confluentinc/common.git)
+* [Confluent support-metrics-common](https://github.com/confluentinc/support-metrics-common.git)
+* [Confluent rest-utils](https://github.com/confluentinc/rest-utils.git)
+* [Confluent schema-registry](https://github.com/confluentinc/schema-registry.git)
+
 Once you did that, just have to clone the ksql-jdbc-driver repo and package it:
 
 ``git clone https://github.com/mmolimar/ksql-jdbc-driver.git && cd ksql-jdbc-driver``
@@ -57,15 +63,16 @@ where:
 
 * **\<ksql-engine>**: represents the KSQL engine host.
 * **\<port>**: is the KSQL engine port.
-* **\<propertyN>**: are the custom client properties (optionals). Fow now, there is only one property and it's to
-set if the KSQL connection is secured or not. The property name is ``secured`` and its value is a boolean
-(``true``|``false``). Its default value is ``false``.
+* **\<propertyN>**: are the custom client properties (optionals). Available properties:
+  * ``secured``: sets if the KSQL connection is secured or not. It's a boolean (``true``|``false``) and its default
+  value is ``false``.
+  * ``timeout``: sets the max wait time between each message when receiving them. It's a long and its default
+  value is ``0`` which means that is infinite.
 
 ## TODO's
 
 - [ ] Standalone mode: connecting directly to Kafka brokers.
 - [ ] Make the driver more compliant with the JDBC spec.
-- [ ] Enable a timeout when waiting for messages in a query.
 
 ## Contribute
 

build.sbt

@@ -1,6 +1,6 @@
 name := "ksql-jdbc-driver"
 
-version := "0.2-SNAPSHOT"
+version := "0.2"
 
 initialize := {
   assert(Integer.parseInt(sys.props("java.specification.version").split("\\.")(1)) >= 7, "Java 7 or above required")
@@ -9,10 +9,11 @@ initialize := {
 scalaVersion := "2.11.11"
 
 resolvers += "Confluent Maven Repo" at "http://packages.confluent.io/maven/"
+resolvers += "Confluent Snapshots Maven Repo" at "https://s3-us-west-2.amazonaws.com/confluent-snapshots/"
 resolvers += Resolver.mavenLocal
 
-libraryDependencies += "io.confluent.ksql" % "ksql-rest-app" % "0.1-SNAPSHOT"
-libraryDependencies += "org.apache.kafka" %% "kafka" % "0.11.0.0-cp1" % "test"
-libraryDependencies += "org.scalatest" %% "scalatest" % "3.0.4" % "test"
+libraryDependencies += "io.confluent.ksql" % "ksql-rest-app" % "4.1.0"
+libraryDependencies += "org.apache.kafka" %% "kafka" % "1.1.0" % "test"
+libraryDependencies += "org.scalatest" %% "scalatest" % "3.0.5" % "test"
 libraryDependencies += "org.scalamock" %% "scalamock-scalatest-support" % "3.6.0" % "test"
 

src/it/scala/com/github/mmolimar/ksql/jdbc/KsqlDriverIntegrationTest.scala

@@ -15,9 +15,9 @@ class KsqlDriverIntegrationTest extends WordSpec with Matchers with BeforeAndAft
   val kafkaCluster = new EmbeddedKafkaCluster(zkServer.getConnection)
   val ksqlEngine = new EmbeddedKsqlEngine(kafkaCluster.getBrokerList)
 
-  val kafkaProducer = TestUtils.buildProducer(kafkaCluster.getBrokerList)
+  lazy val kafkaProducer = TestUtils.buildProducer(kafkaCluster.getBrokerList)
 
-  val ksqlUrl = s"jdbc:ksql://localhost:${ksqlEngine.getPort}"
+  val ksqlUrl = s"jdbc:ksql://localhost:${ksqlEngine.getPort}?timeout=20000"
   var ksqlConnection: Connection = _
   val topic = TestUtils.randomString()
 
@@ -224,15 +224,15 @@ class KsqlDriverIntegrationTest extends WordSpec with Matchers with BeforeAndAft
         |}
       """.stripMargin.getBytes
     val record = new ProducerRecord[Array[Byte], Array[Byte]](topic, key, value)
-    kafkaProducer.send(record).get(1000, TimeUnit.MILLISECONDS)
+    kafkaProducer.send(record).get(10000, TimeUnit.MILLISECONDS)
     Thread.sleep(100)
 
   }
 
   private def createTestTableOrStream(str: String, isStream: Boolean = false) = {
     ksqlConnection.createStatement.execute(s"CREATE ${if (isStream) "STREAM" else "TABLE"} $str " +
       s"(FIELD1 INT, FIELD2 DOUBLE, FIELD3 VARCHAR) " +
-      s"WITH (KAFKA_TOPIC='$topic', VALUE_FORMAT='JSON');") should be(true)
+      s"WITH (KAFKA_TOPIC='$topic', VALUE_FORMAT='JSON', KEY='FIELD1');") should be(true)
   }
 
   override def beforeAll = {

src/it/scala/com/github/mmolimar/ksql/jdbc/embedded/EmbeddedKafkaCluster.scala

@@ -4,9 +4,9 @@ import java.io.File
 import java.util.Properties
 
 import com.github.mmolimar.ksql.jdbc.utils.TestUtils
-import kafka.admin.AdminUtils
 import kafka.server.{KafkaConfig, KafkaServer}
-import kafka.utils.{Logging, ZkUtils}
+import kafka.utils.Logging
+import kafka.zk.AdminZkClient
 
 class EmbeddedKafkaCluster(zkConnection: String,
                            ports: Seq[Int] = Seq(TestUtils.getAvailablePort),
@@ -17,6 +17,9 @@ class EmbeddedKafkaCluster(zkConnection: String,
   private var brokers: Seq[KafkaServer] = Seq.empty
   private var logDirs: Seq[File] = Seq.empty
 
+  private lazy val zkClient = TestUtils.buildZkClient(zkConnection)
+  private lazy val adminZkClient = new AdminZkClient(zkClient)
+
   def startup = {
     info("Starting up embedded Kafka brokers")
 
@@ -58,21 +61,19 @@ class EmbeddedKafkaCluster(zkConnection: String,
 
   def createTopic(topic: String, numPartitions: Int = 1, replicationFactor: Int = 1) = {
     info(s"Creating topic $topic")
-    AdminUtils.createTopic(getZkUtils, topic, numPartitions, replicationFactor)
+    adminZkClient.createTopic(topic, numPartitions, replicationFactor)
   }
 
   def deleteTopic(topic: String) {
     info(s"Deleting topic $topic")
-    AdminUtils.deleteTopic(getZkUtils, topic)
+    adminZkClient.deleteTopic(topic)
   }
 
   def deleteTopics(topics: Seq[String]) = topics.foreach(deleteTopic(_))
 
-  def existTopic(topic: String): Boolean = AdminUtils.topicExists(getZkUtils, topic)
-
-  def listTopics = getZkUtils.getAllTopics
+  def existTopic(topic: String): Boolean = zkClient.topicExists(topic)
 
-  private def getZkUtils: ZkUtils = if (brokers.isEmpty) null else brokers.head.zkUtils
+  def listTopics = zkClient.getAllTopicsInCluster
 
   private def resolvePort(port: Int) = if (port <= 0) TestUtils.getAvailablePort else port
 

src/it/scala/com/github/mmolimar/ksql/jdbc/embedded/EmbeddedKsqlEngine.scala

@@ -4,6 +4,7 @@ import java.io.IOException
 
 import com.github.mmolimar.ksql.jdbc.utils.TestUtils
 import io.confluent.ksql.rest.server.{KsqlRestApplication, KsqlRestConfig}
+import io.confluent.ksql.version.metrics.KsqlVersionCheckerAgent
 import kafka.utils.Logging
 import org.apache.kafka.clients.producer.ProducerConfig
 
@@ -22,7 +23,7 @@ class EmbeddedKsqlEngine(brokerList: String, port: Int = TestUtils.getAvailableP
     "ksql.command.topic.suffix" -> "commands"
   ))
 
-  lazy val ksqlEngine = KsqlRestApplication.buildApplication(config, true)
+  lazy val ksqlEngine = KsqlRestApplication.buildApplication(config, true, new KsqlVersionCheckerAgent)
 
   @throws[IOException]
   def startup = {

src/main/scala/com/github/mmolimar/ksql/jdbc/Exceptions.scala

@@ -2,47 +2,44 @@ package com.github.mmolimar.ksql.jdbc
 
 import java.sql.{SQLException, SQLFeatureNotSupportedException}
 
-sealed trait KsqlException
+sealed trait KsqlException {
+  def message: String = ""
+}
 
-case class InvalidUrl(url: String) extends KsqlException
+case class InvalidUrl(url: String) extends KsqlException {
+  override def message = s"URL with value ${url} is not valid. It must match de regex ${KsqlDriver.urlRegex}"
+}
 
-case class CannotConnect(url: String, msg: String) extends KsqlException
+case class CannotConnect(url: String, msg: String) extends KsqlException {
+  override def message = s"Cannot connect to this URL ${url}. Error message: ${msg}"
+}
 
-case class InvalidProperty(name: String) extends KsqlException
+case class InvalidProperty(name: String) extends KsqlException {
+  override def message = s"Invalid property ${name}."
+}
 
-case class NotSupported(msg: String = "Feature not supported") extends KsqlException
+case class NotSupported(override val message: String = "Feature not supported") extends KsqlException
 
-case class KsqlQueryError(msg: String = "Error executing query") extends KsqlException
+case class KsqlQueryError(override val message: String = "Error executing query") extends KsqlException
 
-case class KsqlCommandError(msg: String = "Error executing command") extends KsqlException
+case class KsqlCommandError(override val message: String = "Error executing command") extends KsqlException
 
-case class InvalidColumn(msg: String = "Invalid column") extends KsqlException
+case class InvalidColumn(override val message: String = "Invalid column") extends KsqlException
 
-case class EmptyRow(msg: String = "Current row is empty") extends KsqlException
+case class EmptyRow(override val message: String = "Current row is empty") extends KsqlException
 
-case class UnknownTableType(msg: String = "Table type does not exist") extends KsqlException
+case class UnknownTableType(override val message: String = "Table type does not exist") extends KsqlException
 
-case class UnknownCatalog(msg: String = "Catalog does not exist") extends KsqlException
+case class UnknownCatalog(override val message: String = "Catalog does not exist") extends KsqlException
 
-case class UnknownSchema(msg: String = "Schema does not exist") extends KsqlException
+case class UnknownSchema(override val message: String = "Schema does not exist") extends KsqlException
 
 object Exceptions {
 
   implicit def wrapException(error: KsqlException): SQLException = {
     error match {
-      case e: InvalidUrl => new SQLException(s"URL with value ${e.url} is not valid." +
-        s"It must match de regex ${KsqlDriver.urlRegex}")
-      case e: CannotConnect => new SQLException(s"Cannot connect to this URL ${e.url}. Error message: ${e.msg}")
-      case e: InvalidProperty => new SQLException(e.name)
-      case e: NotSupported => new SQLFeatureNotSupportedException(e.msg)
-      case e: KsqlQueryError => new SQLException(e.msg)
-      case e: KsqlCommandError => new SQLException(e.msg)
-      case e: InvalidColumn => new SQLException(e.msg)
-      case e: EmptyRow => new SQLException(e.msg)
-      case e: UnknownTableType => new SQLException(e.msg)
-      case e: UnknownCatalog => new SQLException(e.msg)
-      case e: UnknownSchema => new SQLException(e.msg)
-      case _ => new SQLException("Unknown KSQL Exception")
+      case ns: NotSupported => new SQLFeatureNotSupportedException(ns.message)
+      case e => new SQLException(e.message)
     }
   }
 

src/main/scala/com/github/mmolimar/ksql/jdbc/KsqlConnection.scala

@@ -12,12 +12,16 @@ import scala.collection.JavaConverters._
 import scala.util.{Failure, Success, Try}
 
 case class KsqlConnectionValues(ksqlServer: String, port: Int, config: Map[String, String]) {
+
   def getKsqlUrl: String = {
     val protocol = if (isSecured) "https://" else "http://"
     protocol + ksqlServer + ":" + port
   }
 
-  def isSecured: Boolean = config.get("secured").getOrElse("false").toBoolean
+  def isSecured: Boolean = config.getOrElse("secured", "false").toBoolean
+
+  def timeout: Long = config.getOrElse("timeout", "0").toLong
+
 }
 
 class KsqlConnection(values: KsqlConnectionValues, properties: Properties) extends Connection with WrapperNotSupported {
@@ -88,7 +92,7 @@ class KsqlConnection(values: KsqlConnectionValues, properties: Properties) exten
       throw NotSupported("ResultSetType, ResultSetConcurrency and ResultSetHoldability must be" +
         " TYPE_FORWARD_ONLY, CONCUR_READ_ONLY, HOLD_CURSORS_OVER_COMMIT respectively ")
     }
-    new KsqlStatement(ksqlClient)
+    new KsqlStatement(ksqlClient, values.timeout)
   }
 
   override def getHoldability: Int = throw NotSupported()

src/main/scala/com/github/mmolimar/ksql/jdbc/KsqlStatement.scala

@@ -6,7 +6,8 @@ import com.github.mmolimar.ksql.jdbc.Exceptions._
 import com.github.mmolimar.ksql.jdbc.resultset.KsqlResultSet
 import io.confluent.ksql.rest.client.KsqlRestClient
 
-class KsqlStatement(private val ksqlClient: KsqlRestClient) extends Statement with WrapperNotSupported {
+class KsqlStatement(private val ksqlClient: KsqlRestClient, val timeout: Long = 0)
+  extends Statement with WrapperNotSupported {
 
   override def setMaxFieldSize(max: Int): Unit = throw NotSupported()
 
@@ -104,6 +105,6 @@ class KsqlStatement(private val ksqlClient: KsqlRestClient) extends Statement wi
 
   private def fixSql(sql: String) = if (sql.trim.endsWith(";")) sql else sql + ";"
 
-  private implicit def toResultSet(stream: KsqlRestClient.QueryStream): ResultSet = new KsqlResultSet(stream)
+  private implicit def toResultSet(stream: KsqlRestClient.QueryStream): ResultSet = new KsqlResultSet(stream, timeout)
 
 }

src/main/scala/com/github/mmolimar/ksql/jdbc/resultset/KsqlResultSet.scala

@@ -9,20 +9,35 @@ import io.confluent.ksql.rest.client.KsqlRestClient
 import io.confluent.ksql.rest.entity.StreamedRow
 
 import scala.collection.JavaConversions._
+import scala.concurrent.ExecutionContext.Implicits.global
+import scala.concurrent.duration._
+import scala.concurrent.{Await, Future, TimeoutException}
+import scala.util.{Failure, Success, Try}
 
-class KsqlResultSet(private[jdbc] val stream: KsqlRestClient.QueryStream) extends AbstractResultSet[StreamedRow](stream) {
+class KsqlResultSet(private[jdbc] val stream: KsqlRestClient.QueryStream, val timeout: Long = 0)
+  extends AbstractResultSet[StreamedRow](stream) {
 
   private val emptyRow: StreamedRow = new StreamedRow(new GenericRow, null)
 
-  override def next: Boolean = stream.hasNext match {
-    case true =>
-      stream.next match {
-        case record if Option(record.getRow) == None => false
-        case record =>
-          currentRow = Some(record)
-          true
-      }
-    case false => false
+  private val waitDuration = if (timeout > 0) timeout millis else Duration.Inf
+
+  override def next: Boolean = {
+    def hasNext = stream.hasNext match {
+      case true =>
+        stream.next match {
+          case record if Option(record.getRow) == None => false
+          case record =>
+            currentRow = Some(record)
+            true
+        }
+      case false => false
+    }
+
+    Try(Await.result(Future(hasNext), waitDuration)) match {
+      case Success(r) => r
+      case Failure(_: TimeoutException) => false
+      case Failure(e) => throw e
+    }
   }
 
   override def isBeforeFirst: Boolean = false