<commit_before>#### VoltDB Sink Connector

The connector is for moving data from Kafka to VoltDB. The connector consumes kafka messages and produces procedure parameters for invoking procedures inside VoltDB. The procedure can be user defined procedure or standard TABLE CRUD procedure typically named as TABLENAME.insert (for insert). When the kafka messages have schema the schema is looked at to convert the values. If node schema is present message will converted as configured. 

#### Setup

* Install Gradle

	On a Mac if you have [Homebrew](http://brew.sh/) setup then simply install the gradle bottle

	```bash
	brew install gradle
	```
	On Linux setup [SDK](http://sdkman.io/), and install gradle as follows
	
	```bash
	sdk install gradle
	```
* Create `gradle.properties` file

	```bash
	echo voltdbhome=/voltdb/home/dirname > gradle.properties
    set the `voltdbhome` property to the base directory where your VoltDB is installed
    set the `dbversion` property to the VoltDB build version
	```

* Compile artifacts

	```bash
    gradle prepare zip shadowJar
	```

* To setup an eclipse project run gradle as follows

	```bash
    gradle cleanEclipse eclipse
	```
	then import it into your eclipse workspace by using File->Import projects menu option, and add connect-api-0.10.0.0.jar, connect-json-0.10.0.0.jar and kafka-clients-1.10.0.0.jar to classpath.

#### Installing Connector 

* All the scripts for running Kafka Connect will use the CLASSPATH environment variable if it is set when they are invoked.
* Copy the built jar from `build/libs` to a directory  where Kafka server is installed or accessible. 
  example: path-to-kafka-root/voltdb/
* copy /config/voltdb-sink-connector.properties, voltdb-sink-connector.json to a folder and configure connector properties
* configure connect-standalone.properties or connect-distributed.properties under path-to-kafka-root/config/
  For more details please see http://docs.confluent.io/3.0.1/connect/index.html

#### Connect Properties (voltdb-sink-connector.properties)
- **name** (default:KafkaSinkConnector) Unique name for the connector
- **connector.class** (default:org.voltdb.connect.kafka.KafkaSinkConnector) The Java class for the connector
- **voltdb.connection.user** The user name to connect VoltDB
- **voltdb.connection.password** The password to connect VoltDB
- **voltdb.servers** (mandatory) A list of Voltdb server nodes with ',' as delimiter. example: server1:21212,server2:21212
- **voltdb.procedure** (mandatory) The procedure name to be used to insert data to VoltDB.
- **formatter.factory.class** The data formatter factory used to convert non-schema Kafka data into the format required by VoltDB procedure.
   *org.voltdb.connect.formatter.CSVFormatterFactory* is used as default.
- **formatter.type** The type of formatter, such as csv, tsv.
- **data.converter.class** The Java class for data conversion from SinkRecord. *org.voltdb.connect.converter.JsonDataConverter* as default.
   * Formatter and converter properties are used if the kafka record does not have valid value schema. 
   * Also when kafka records does not have schema you must modify following connector properties (standalone or distributed)
       * **key.converter** Update it to org.apache.kafka.connect.storage.StringConverter
       * **value.converter** Update it to org.apache.kafka.connect.storage.StringConverter
       * **key.converter.schemas.enable** Disable schema by setting it to false
       * **value.converter.schemas.enable** Disable schema by setting it to false
- **kerberos.authentication** The authentication module if enabled.

#### Connect JSON Properties (voltdb-sink-connector.json)

The json file contains the same properties as voltdb-sink-connector.properties. Example:
```
	{
   		"name":"KafkaSinkConnector",
   		"config":{
      		"connector.class":"org.voltdb.connect.kafka.KafkaSinkConnector",
      		"tasks.max":"1",
      		"voltdb.servers":"localhost:21212",
      		"voltdb.procedure":"TestTable.insert",
      		"formatter.factory.class":"org.voltdb.connect.formatter.CSVFormatterFactory",
      		"voltdb.connection.user":"",
      		"voltdb.connection.password":"",
      		"formatter.type":"csv",
      		"data.converter.class":"org.voltdb.connect.converter.JsonDataConverter",
      		"topics":"testTopic",
      		"kerberos.authentication":"false"
   		}
	}
```

#### Instructions for running

* Download and install [Kafka](http://kafka.apache.org/downloads.html) 0.10.0.0 or the latest
* Start VoltDB, create correct table and store procedure for the connector:

```sql
  	A sample table to VoltDB for testing:
    CREATE TABLE STOCK (
		ID integer not null,
		Symbol varchar(50),
		Quantity integer,
		TradeType varchar(50),
		price float,
		constraint pk_stock PRIMARY KEY(ID)
   );
```
* Start zookeeper

	```<commit_msg>Fix readme (#6)<commit_after>#### VoltDB Kafka Sink Connector

The VoltDB Kafka Sink Connector provides a simple, continuous link from a Kafka topic or set of topics to a VoltDB table or set of VoltDB tables.

The connector consumes Kafka log messages and invokes VoltDB stored procedures, either custom or default, to parse and insert the messages into VoltDB tables.

When the Kafka messages have schema metadata, the schema is used to convert the values. If no schema metadata is present, the message will be converted as configured. 

The VoltDB Kafka Sink Connector is available as source at [Github](https://github.com/VoltDB/voltdb-kafka-connector) with build instructions below, or as a ready to use Java jar file that can be added to your VoltDB installation, available for [download](https://www.voltdb.com/download-confluentconnector)

See "Installing the Connector" for next steps after building or downloading the Connector.

Following sections discuss configuring the Connector and offer an example application that shows the Connector in action with VoltDB and a Kafka installation.

#### Installing the Connector 

All the scripts for running Kafka Connect use the CLASSPATH environment variable if it is set.
* copy the built jar from `build/libs` to a directory where the Kafka server is installed or accessible, for example: path-to-kafka-root/voltdb/
* copy config/voltdb-sink-connector.properties, voltdb-sink-connector.json to a folder and configure the connector properties as described in the next section
* configure connect-standalone.properties or connect-distributed.properties under path-to-kafka-root/config. Note that Apache or Confluent Kafka distributions may have configuration files in different locations.
For more details please see http://docs.confluent.io/3.0.1/connect/index.html

#### Building the Connector from source

* Install Gradle

	On Mac OS X use [Homebrew](http://brew.sh/) to install gradle

	```bash
	brew install gradle
	```
	On Linux setup [SDK](http://sdkman.io/), and install gradle as follows
	
	```bash
	sdk install gradle
	```
* Create `gradle.properties` file

	```bash
	echo voltdbhome=/voltdb/home/dirname > gradle.properties
    set the `voltdbhome` property to the base directory where your VoltDB is installed
    set the `dbversion` property to the VoltDB build version
	```

* Compile artifacts

	```bash
    gradle prepare zip shadowJar
	```

* To setup an Eclipse project, run gradle as follows

	```bash
    gradle cleanEclipse eclipse
	```
	then import it into your Eclipse workspace by using File->Import projects menu option, and add connect-api-0.10.0.0.jar, connect-json-0.10.0.0.jar and kafka-clients-1.10.0.0.jar to the classpath.

#### Connect Properties (voltdb-sink-connector.properties)
- **name** (default:KafkaSinkConnector) Unique name for the connector
- **connector.class** (default:org.voltdb.connect.kafka.KafkaSinkConnector) The Java class for the connector
- **voltdb.connection.user** The user name to connect VoltDB
- **voltdb.connection.password** The password to connect VoltDB
- **voltdb.servers** (mandatory) A list of Voltdb server nodes with ',' as delimiter. example: server1:21212,server2:21212
- **voltdb.procedure** (mandatory) The procedure name to be used to insert data to VoltDB.
- **formatter.factory.class** The data formatter factory used to convert non-schema Kafka data into the format required by VoltDB procedure.
   *org.voltdb.connect.formatter.CSVFormatterFactory* is used as default.
- **formatter.type** The type of formatter, such as csv, tsv.
- **data.converter.class** The Java class for data conversion from SinkRecord. *org.voltdb.connect.converter.JsonDataConverter* as default.
   * Formatter and converter properties are used if the kafka record does not have valid value schema. 
   * Also when kafka records does not have schema you must modify following connector properties (standalone or distributed)
       * **key.converter** Update it to org.apache.kafka.connect.storage.StringConverter
       * **value.converter** Update it to org.apache.kafka.connect.storage.StringConverter
       * **key.converter.schemas.enable** Disable schema by setting it to false
       * **value.converter.schemas.enable** Disable schema by setting it to false
- **kerberos.authentication** The authentication module if enabled.

#### Connect JSON Properties (voltdb-sink-connector.json)

The json file contains the same properties as voltdb-sink-connector.properties. Example:
```
	{
   		"name":"KafkaSinkConnector",
   		"config":{
      		"connector.class":"org.voltdb.connect.kafka.KafkaSinkConnector",
      		"tasks.max":"1",
      		"voltdb.servers":"localhost:21212",
      		"voltdb.procedure":"TestTable.insert",
      		"formatter.factory.class":"org.voltdb.connect.formatter.CSVFormatterFactory",
      		"voltdb.connection.user":"",
      		"voltdb.connection.password":"",
      		"formatter.type":"csv",
      		"data.converter.class":"org.voltdb.connect.converter.JsonDataConverter",
      		"topics":"testTopic",
      		"kerberos.authentication":"false"
   		}
	}
```

#### Instructions for running

* Download and install [Kafka](http://kafka.apache.org/downloads.html) 0.10.0.0 or later
* Start VoltDB, create correct table and store procedure for the connector:

```sql
  	A sample table to VoltDB for testing:
    CREATE TABLE STOCK (
		ID integer not null,
		Symbol varchar(50),
		Quantity integer,
		TradeType varchar(50),
		price float,
		constraint pk_stock PRIMARY KEY(ID)
   );
```
* Start zookeeper

	```
{'fraction_non_alphanumeric': 0.09720616570327553, 'fraction_numerical': 0.008766859344894027, 'mean_word_length': 4.208730556949322, 'pattern_counts': {'":': 26, '<': 3, '<?xml version=': 0, '>': 7, 'https://': 2, 'lorem ipsum': 0, 'www.': 1, 'xml': 0}, 'pii_count': 8, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 7, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2784409', 'n_tokens_mistral': 3153, 'n_tokens_neox': 2979, 'n_words': 1068}