diff --git a/.gitlab/.gitlab-ci.yml b/.gitlab/.gitlab-ci.yml
index 3cf8c1676..59d94c4a6 100644
--- a/.gitlab/.gitlab-ci.yml
+++ b/.gitlab/.gitlab-ci.yml
@@ -365,6 +365,9 @@ run go tests dax/test/dax:
 run go tests cli kafka integration:
   stage: integration
   image: golang:$GOVERSION
+  tags:
+  - shell
+  - aws
   variables:
     KAFKA_RUNNER_TEST_FEATUREBASE_HOST: pilosa:10101
     KAFKA_RUNNER_TEST_FEATUREBASEGRPC_HOST: pilosa:20101
@@ -375,19 +378,14 @@ run go tests cli kafka integration:
   script:
     - echo "running fbsql integration tests"
     - cd ./idk/
-    - BRANCH_NAME=${CI_COMMIT_REF_SLUG} make start-pilosa
-    - BRANCH_NAME=${CI_COMMIT_REF_SLUG} make start-zookeeper
-    - BRANCH_NAME=${CI_COMMIT_REF_SLUG} make start-kafka
-    - BRANCH_NAME=${CI_COMMIT_REF_SLUG} make start-schema-registry
-    - cd ..
-    - go test -coverprofile=coverage-cli-kafka-integration.out -run -timeout=10m TestKafkaRunner ./cli
+    - make test-cli
   after_script:
     - cd ./idk/
+    - make save-pilosa-logs
     - make shutdown
-    - rm -rf /mnt/ramdisk/test-$CI_JOB_ID
   artifacts:
     paths:
-      - coverage-cli-kafka-integration.out
+      - ./idk/testdata/*_coverage.out
   needs:
     - job: build amd container fb
     - job: build fbsql amd64
@@ -523,16 +521,23 @@ upload to sonarcloud:
 package for linux amd64:
   stage: build
   image: golang:$GOVERSION
-  extends: .go-cache
   rules:
     - if: '$CI_PIPELINE_SOURCE == "push" || $CI_PIPELINE_SOURCE == "schedule" || $CI_PIPELINE_SOURCE == "web"'
   variables:
     GOOS: "linux"
     GOARCH: "amd64"
   script:
+    - export VERSION=$(git describe --tags 2>/dev/null || echo unknown)
     - echo 'deb [trusted=yes] https://repo.goreleaser.com/apt/ /' | tee /etc/apt/sources.list.d/goreleaser.list
-    - apt update && apt install nfpm=2.11.3
-    - make package
+    - apt-get update -y && apt-get install -y -qq nfpm=2.11.3
+    - nfpm version # smoke
+    - mv "featurebase_${GOOS}_${GOARCH}" featurebase
+    - mv "./build/fbsql_${GOOS}_${GOARCH}" fbsql
+    - nfpm package --packager deb --target "featurebase.${VERSION}.${GOARCH}.deb"
+    - nfpm package --packager rpm --target "featurebase.${VERSION}.${GOARCH}.rpm"
+  needs:
+    - build featurebase
+    - build fbsql amd64
   artifacts:
     paths:
       - "*.deb"
@@ -561,16 +566,23 @@ trigger_m-cloud-images:
 package for linux arm64:
   stage: build
   image: golang:$GOVERSION
-  extends: .go-cache
   rules:
     - if: '$CI_PIPELINE_SOURCE == "push" || $CI_PIPELINE_SOURCE == "schedule" || $CI_PIPELINE_SOURCE == "web"'
   variables:
     GOOS: "linux"
     GOARCH: "arm64"
   script:
+    - export VERSION=$(git describe --tags 2>/dev/null || echo unknown)
     - echo 'deb [trusted=yes] https://repo.goreleaser.com/apt/ /' | tee /etc/apt/sources.list.d/goreleaser.list
-    - apt update && apt install nfpm=2.11.3
-    - make package
+    - apt-get update -y && apt-get install -y -qq nfpm=2.11.3
+    - nfpm version # smoke
+    - mv "featurebase_${GOOS}_${GOARCH}" featurebase
+    - mv "./build/fbsql_${GOOS}_${GOARCH}" fbsql
+    - nfpm package --packager deb --target "featurebase.${VERSION}.${GOARCH}.deb"
+    - nfpm package --packager rpm --target "featurebase.${VERSION}.${GOARCH}.rpm"
+  needs:
+    - build featurebase
+    - build fbsql arm64 linux
   artifacts:
     paths:
       - "*.deb"
diff --git a/Makefile b/Makefile
index b32ba3be1..777e65bd2 100644
--- a/Makefile
+++ b/Makefile
@@ -123,7 +123,7 @@ build:
 
 package:
 	GOOS=$(GOOS) GOARCH=$(GOARCH) $(MAKE) build
-	GOOS=$(GOOS) GOARCH=$(GOARCH) $(MAKE) build-fbsql
+	GOOS=$(GOOS) GOARCH=$(GOARCH) $(MAKE) docker-build-fbsql
 	GOARCH=$(GOARCH) VERSION=$(VERSION) nfpm package --packager deb --target featurebase.$(VERSION).$(GOARCH).deb
 	GOARCH=$(GOARCH) VERSION=$(VERSION) nfpm package --packager rpm --target featurebase.$(VERSION).$(GOARCH).rpm
 
diff --git a/cli/cli_kafka_integration_test.go b/cli/cli_kafka_integration_test.go
index c8ca6ec8c..bac2e7615 100644
--- a/cli/cli_kafka_integration_test.go
+++ b/cli/cli_kafka_integration_test.go
@@ -8,6 +8,7 @@ import (
 	"fmt"
 	"io/ioutil"
 	"os"
+	"sort"
 	"strconv"
 	"strings"
 	"testing"
@@ -340,9 +341,11 @@ func verifyQueryReponse(t *testing.T, wqr *featurebase.WireQueryResponse, expect
 		for j, element := range line {
 			switch newElement := element.(type) {
 			case featurebase.StringSet:
-				newline[j] = newElement.SortedStringSlice()
+				sort.Strings(newElement)
+				newline[j] = newElement
 			case featurebase.IDSet:
-				newline[j] = newElement.SortedInt64Slice()
+				sort.Slice(newElement, func(i, j int) bool { return newElement[i] < newElement[j] })
+				newline[j] = newElement
 			default:
 				newline[j] = element
 			}
diff --git a/cli/kafka.go b/cli/kafka.go
index 1392dbc9f..effbffb4e 100644
--- a/cli/kafka.go
+++ b/cli/kafka.go
@@ -10,7 +10,7 @@ func (cmd *Command) newKafkaRunner(cfgFile string) (*kafka.Runner, error) {
 
 	cfg, err := kafka.ConfigFromFile(cfgFile)
 	if err != nil {
-		return nil, err
+		return nil, errors.Wrap(err, "getting config from file")
 	}
 
 	if err := kafka.ValidateConfig(cfg); err != nil {
@@ -49,14 +49,14 @@ func (cmd *Command) newKafkaRunner(cfgFile string) (*kafka.Runner, error) {
 		return nil, errors.Wrap(err, "getting fields from config")
 	}
 
-	// for avro, let the SchemaManager and IDK handle fields
-	if cfg.Encode == "avro" {
-		flds = nil
-	}
-
-	return kafka.NewRunner(
+	kr := kafka.NewRunner(
 		idkCfg,
 		batch.NewSQLBatcher(cmd, flds),
 		cmd.stderr,
-	), nil
+	)
+
+	// set pilosa host which is the only IDK config to come through
+	// configuration flags rather than the kafka config file
+	kr.Main.PilosaHosts = []string{cmd.host + ":" + cmd.port}
+	return kr, nil
 }
diff --git a/cli/kafka/config.go b/cli/kafka/config.go
index 4e09b9dc2..3ffdee64f 100644
--- a/cli/kafka/config.go
+++ b/cli/kafka/config.go
@@ -32,10 +32,14 @@ type Config struct {
 	Table  string  `mapstructure:"table" help:"Destination table name."`
 	Fields []Field `mapstructure:"fields"`
 
+	SchemaRegistryURL      string `mapstructure:"schema-registry-url" help:"host and port of schema registry. Defaults to localhost:8081"`
+	SchemaRegistryUsername string `mapstructure:"schema-registry-username" help:"authenticaion key provided by confluent for schema registry."`
+	SchemaRegistryPassword string `mapstructure:"schema-registry-password" help:"authenticaion secret provided by confluent for schema registry."`
+
 	Encode             string `mapstructure:"encode" help:"Encoding format (currently supported formats: avro, json)"`
 	AllowMissingFields bool   `mapstructure:"allow-missing-fields" help:"allow missing fields in messages from kafka"`
 	MaxMessages        int    `mapstructure:"max-messages" help:"max messages read from kakfka"`
-	ConfluentConfig    string `mapstructure:"confluent-config" help:"max messages read from kakfka"`
+	ConfluentConfig    string `mapstructure:"confluent-config" help:"path to JSON file mapping librdkafka consumer configurations to configuration values"`
 }
 
 // Field is a user-facing configuration field.
@@ -62,6 +66,10 @@ type ConfigForIDK struct {
 	PrimaryKeys []string
 	Fields      []idk.RawField
 
+	SchemaRegistryURL      string
+	SchemaRegistryUsername string
+	SchemaRegistryPassword string
+
 	Encode             string
 	AllowMissingFields bool
 	MaxMessages        int
@@ -117,40 +125,40 @@ func ValidateConfig(c Config) error {
 		return validateConfigJSON(c)
 	case encodingTypeAvro:
 		return validateConfigAvro(c)
+	default:
+		return errors.Errorf("encode configuration value must be %s or %s: got %s", encodingTypeJSON, encodingTypeAvro, c.Encode)
 	}
-
-	return nil
-
 }
 
 func validateConfigJSON(c Config) error {
 
-	if len(c.Fields) > 0 {
+	switch len(c.Fields) {
+	case 0:
 		// We only need to do these checks if any fields are specified at all.
 		// If no fields are specified, that's ok because then we default to
 		// using fields based off the existing table.
-		if len(c.Fields) < 2 {
-			return errors.Errorf("at least two fields are required (one should be a primary key)")
-		} else {
-			var found int
-			for i := range c.Fields {
-				if c.Fields[i].PrimaryKey {
-					found++
-				}
-				if c.Fields[i].Name == "" {
-					return errors.Errorf("a name attribute (which isn't equal to \"\") should exist for all fields")
-				}
-				if c.Fields[i].SourceType == "" {
-					return errors.Errorf("a source-type attribute (which isn't equal to \"\") should exist for all fields")
-				}
+		return nil
+	case 1:
+		return errors.Errorf("at least two fields are required (one should be a primary key)")
+	default:
+		var found int
+		for i := range c.Fields {
+			if c.Fields[i].PrimaryKey {
+				found++
 			}
-			if found < 1 {
-				return errors.Errorf("at least one primary key field is required")
+			if c.Fields[i].Name == "" {
+				return errors.Errorf("a name attribute (which isn't equal to \"\") should exist for all fields")
+			}
+			if c.Fields[i].SourceType == "" {
+				return errors.Errorf("a source-type attribute (which isn't equal to \"\") should exist for all fields")
 			}
 		}
+		if found < 1 {
+			return errors.Errorf("at least one primary key field is required")
+		}
+		return nil
 	}
 
-	return nil
 }
 
 // Only primary key fields required
@@ -181,17 +189,20 @@ func ConvertConfig(c Config) (ConfigForIDK, error) {
 
 	// Copy all the shared members from Config to ConfigForIDK.
 	out := ConfigForIDK{
-		Hosts:              c.Hosts,
-		Group:              c.Group,
-		Topics:             c.Topics,
-		BatchSize:          c.BatchSize,
-		BatchMaxStaleness:  c.BatchMaxStaleness,
-		Timeout:            c.Timeout,
-		Table:              c.Table,
-		Encode:             c.Encode,
-		AllowMissingFields: c.AllowMissingFields,
-		MaxMessages:        c.MaxMessages,
-		ConfluentConfig:    c.ConfluentConfig,
+		Hosts:                  c.Hosts,
+		Group:                  c.Group,
+		Topics:                 c.Topics,
+		BatchSize:              c.BatchSize,
+		BatchMaxStaleness:      c.BatchMaxStaleness,
+		Timeout:                c.Timeout,
+		Table:                  c.Table,
+		Encode:                 c.Encode,
+		AllowMissingFields:     c.AllowMissingFields,
+		MaxMessages:            c.MaxMessages,
+		ConfluentConfig:        c.ConfluentConfig,
+		SchemaRegistryURL:      c.SchemaRegistryURL,
+		SchemaRegistryUsername: c.SchemaRegistryUsername,
+		SchemaRegistryPassword: c.SchemaRegistryPassword,
 	}
 
 	if len(c.Fields) == 0 {
@@ -282,6 +293,11 @@ func ConfigToFields(c Config, primaryKeys []string) ([]*dax.Field, error) {
 	// capacity to `len(c.Fields)-1`.
 	out := make([]*dax.Field, 0, len(c.Fields))
 
+	// for avro, let the SchemaManager and IDK handle fields
+	if c.Encode == encodingTypeAvro {
+		return nil, nil
+	}
+
 	for _, fld := range c.Fields {
 		// When we have a single primary key, don't also store that value as a
 		// field in FeatureBase. However, when we have more than one primary
diff --git a/cli/kafka/config_test.go b/cli/kafka/config_test.go
index 5088950ed..35830b829 100644
--- a/cli/kafka/config_test.go
+++ b/cli/kafka/config_test.go
@@ -16,19 +16,19 @@ func TestConfigFromFile(t *testing.T) {
 	tests := []configFromFileTest{
 		{ // confirm json config struct fields are being set
 			configFilePath: "./testdata/config/config01.toml",
-			expectedStruct: `{"Hosts":["kafka:9090"],"Group":"testGroup","Topics":["testTopic"],"BatchSize":35,"BatchMaxStaleness":25000000000,"Timeout":16000000000,"Table":"testTable","Fields":[{"Name":"id","SourceType":"id","SourcePath":null,"PrimaryKey":true},{"Name":"name","SourceType":"string","SourcePath":["test","path"],"PrimaryKey":false},{"Name":"age","SourceType":"int","SourcePath":null,"PrimaryKey":false},{"Name":"hobbies","SourceType":"stringset","SourcePath":null,"PrimaryKey":false}],"Encode":"json","AllowMissingFields":false,"MaxMessages":0,"ConfluentConfig":""}`,
+			expectedStruct: `{"Hosts":["kafka:9090"],"Group":"testGroup","Topics":["testTopic"],"BatchSize":35,"BatchMaxStaleness":25000000000,"Timeout":16000000000,"Table":"testTable","Fields":[{"Name":"id","SourceType":"id","SourcePath":null,"PrimaryKey":true},{"Name":"name","SourceType":"string","SourcePath":["test","path"],"PrimaryKey":false},{"Name":"age","SourceType":"int","SourcePath":null,"PrimaryKey":false},{"Name":"hobbies","SourceType":"stringset","SourcePath":null,"PrimaryKey":false}],"SchemaRegistryURL":"","SchemaRegistryUsername":"","SchemaRegistryPassword":"","Encode":"json","AllowMissingFields":false,"MaxMessages":0,"ConfluentConfig":""}`,
 		},
 		{ // confirm defaults are being set
 			configFilePath: "./testdata/config/config00.toml",
-			expectedStruct: `{"Hosts":["localhost:9092"],"Group":"default-featurebase-group","Topics":["topic00"],"BatchSize":1,"BatchMaxStaleness":5000000000,"Timeout":5000000000,"Table":"table00","Fields":[{"Name":"id","SourceType":"string","SourcePath":null,"PrimaryKey":true},{"Name":"name","SourceType":"string","SourcePath":null,"PrimaryKey":false},{"Name":"age","SourceType":"int","SourcePath":null,"PrimaryKey":false},{"Name":"hobbies","SourceType":"stringset","SourcePath":null,"PrimaryKey":false}],"Encode":"json","AllowMissingFields":false,"MaxMessages":0,"ConfluentConfig":""}`,
+			expectedStruct: `{"Hosts":["localhost:9092"],"Group":"default-featurebase-group","Topics":["topic00"],"BatchSize":1,"BatchMaxStaleness":5000000000,"Timeout":5000000000,"Table":"table00","Fields":[{"Name":"id","SourceType":"string","SourcePath":null,"PrimaryKey":true},{"Name":"name","SourceType":"string","SourcePath":null,"PrimaryKey":false},{"Name":"age","SourceType":"int","SourcePath":null,"PrimaryKey":false},{"Name":"hobbies","SourceType":"stringset","SourcePath":null,"PrimaryKey":false}],"SchemaRegistryURL":"","SchemaRegistryUsername":"","SchemaRegistryPassword":"","Encode":"json","AllowMissingFields":false,"MaxMessages":0,"ConfluentConfig":""}`,
 		},
 		{ // confirm avro config struct field are being set
 			configFilePath: "./testdata/config/config02.toml",
-			expectedStruct: `{"Hosts":["kafka:9090"],"Group":"testGroup","Topics":["testTopic"],"BatchSize":35,"BatchMaxStaleness":25000000000,"Timeout":16000000000,"Table":"testTable","Fields":[{"Name":"id","SourceType":"id","SourcePath":null,"PrimaryKey":true},{"Name":"name","SourceType":"string","SourcePath":["test","path"],"PrimaryKey":false},{"Name":"age","SourceType":"int","SourcePath":null,"PrimaryKey":false},{"Name":"hobbies","SourceType":"stringset","SourcePath":null,"PrimaryKey":false}],"Encode":"avro","AllowMissingFields":false,"MaxMessages":0,"ConfluentConfig":""}`,
+			expectedStruct: `{"Hosts":["kafka:9090"],"Group":"testGroup","Topics":["testTopic"],"BatchSize":35,"BatchMaxStaleness":25000000000,"Timeout":16000000000,"Table":"testTable","Fields":[{"Name":"id","SourceType":"id","SourcePath":null,"PrimaryKey":true},{"Name":"name","SourceType":"string","SourcePath":["test","path"],"PrimaryKey":false},{"Name":"age","SourceType":"int","SourcePath":null,"PrimaryKey":false},{"Name":"hobbies","SourceType":"stringset","SourcePath":null,"PrimaryKey":false}],"SchemaRegistryURL":"","SchemaRegistryUsername":"","SchemaRegistryPassword":"","Encode":"avro","AllowMissingFields":false,"MaxMessages":0,"ConfluentConfig":""}`,
 		},
 		{ // confirm avro config struct field are being set
 			configFilePath: "./testdata/config/config03.toml",
-			expectedStruct: `{"Hosts":["kafka:9090"],"Group":"testGroup","Topics":["testTopic"],"BatchSize":35,"BatchMaxStaleness":25000000000,"Timeout":16000000000,"Table":"testTable","Fields":[{"Name":"id","SourceType":"id","SourcePath":null,"PrimaryKey":true},{"Name":"name","SourceType":"string","SourcePath":["test","path"],"PrimaryKey":false},{"Name":"age","SourceType":"int","SourcePath":null,"PrimaryKey":false},{"Name":"hobbies","SourceType":"stringset","SourcePath":null,"PrimaryKey":false}],"Encode":"avro","AllowMissingFields":true,"MaxMessages":100,"ConfluentConfig":"./test/confluent/config.json"}`,
+			expectedStruct: `{"Hosts":["kafka:9090"],"Group":"testGroup","Topics":["testTopic"],"BatchSize":35,"BatchMaxStaleness":25000000000,"Timeout":16000000000,"Table":"testTable","Fields":[{"Name":"id","SourceType":"id","SourcePath":null,"PrimaryKey":true},{"Name":"name","SourceType":"string","SourcePath":["test","path"],"PrimaryKey":false},{"Name":"age","SourceType":"int","SourcePath":null,"PrimaryKey":false},{"Name":"hobbies","SourceType":"stringset","SourcePath":null,"PrimaryKey":false}],"SchemaRegistryURL":"","SchemaRegistryUsername":"","SchemaRegistryPassword":"","Encode":"avro","AllowMissingFields":true,"MaxMessages":100,"ConfluentConfig":"./test/confluent/config.json"}`,
 		},
 	}
 
diff --git a/cli/kafka/runner.go b/cli/kafka/runner.go
index 5c92a4806..e743fe68f 100644
--- a/cli/kafka/runner.go
+++ b/cli/kafka/runner.go
@@ -41,6 +41,7 @@ func NewRunner(cfg ConfigForIDK, batcher fbbatch.Batcher, logWriter io.Writer) *
 	idkMain.OffsetMode = true
 	idkMain.Namespace = "sql_kafka_runner"
 	idkMain.Pprof = "" // don't initialize pprof until we actually use it in tests
+
 	kr := &Runner{
 		Main:    *idkMain,
 		Hosts:   cfg.Hosts,
@@ -50,10 +51,16 @@ func NewRunner(cfg ConfigForIDK, batcher fbbatch.Batcher, logWriter io.Writer) *
 		Timeout: cfg.Timeout,
 	}
 
+	kr.KafkaBootstrapServers = cfg.Hosts
+	kr.SchemaRegistryURL = cfg.SchemaRegistryURL
+	kr.SchemaRegistryUsername = cfg.SchemaRegistryUsername
+	kr.SchemaRegistryPassword = cfg.SchemaRegistryPassword
+
 	// NewSource should be set based on the encoding of the source (e.g. JSON, Avro)
-	if cfg.Encode == encodingTypeAvro {
+	switch cfg.Encode {
+	case encodingTypeAvro:
 		kr.GetAvroNewSource(cfg)
-	} else if cfg.Encode == encodingTypeJSON {
+	default:
 		kr.GetJSONNewSource(cfg)
 	}
 
@@ -61,9 +68,10 @@ func NewRunner(cfg ConfigForIDK, batcher fbbatch.Batcher, logWriter io.Writer) *
 }
 
 func (r *Runner) GetJSONNewSource(cfg ConfigForIDK) {
+
 	r.NewSource = func() (idk.Source, error) {
 		source := kafka_sasl.NewSource()
-		source.KafkaBootstrapServers = r.Hosts
+		//source.KafkaBootstrapServers = r.Hosts
 		source.Group = r.Group
 		source.Topics = r.Topics
 		source.Log = r.Main.Log()
@@ -86,19 +94,23 @@ func (r *Runner) GetJSONNewSource(cfg ConfigForIDK) {
 }
 
 func (r *Runner) GetAvroNewSource(cfg ConfigForIDK) {
+
 	r.NewSource = func() (idk.Source, error) {
 		source := kafka.NewSource()
 		source.KafkaBootstrapServers = r.Hosts
+		source.SchemaRegistryURL = cfg.SchemaRegistryURL
+		source.SchemaRegistryUsername = cfg.SchemaRegistryUsername
+		source.SchemaRegistryPassword = cfg.SchemaRegistryPassword
 		source.Group = r.Group
 		source.Topics = r.Topics
 		source.Log = r.Main.Log()
 		source.Timeout = r.Timeout
 		source.KafkaConfiguration = cfg.ConfluentConfig
-		confluentcfg, err := common.SetupConfluent(&r.ConfluentCommand)
+		confluentCfg, err := common.SetupConfluent(&r.ConfluentCommand)
 		if err != nil {
 			return nil, err
 		}
-		source.ConfigMap = confluentcfg
+		source.ConfigMap = confluentCfg
 
 		err = source.Open()
 		if err != nil {
diff --git a/cli/kafka/testdata/runner/config/config04.toml b/cli/kafka/testdata/runner/config/config04.toml
index 6df6d5ee8..e6a50b10c 100644
--- a/cli/kafka/testdata/runner/config/config04.toml
+++ b/cli/kafka/testdata/runner/config/config04.toml
@@ -6,7 +6,7 @@ batch-size = 1
 batch-max-staleness = "5s"
 timeout = "5s"
 encode = "avro"
-schemaRegistryHost = "localhost:8081"
+schema-registry-url = "SCHEMA_REGISTRY_SERVICE"
 max-messages = MAX_MESSAGES
 
 [[fields]]
diff --git a/cli/kafka/testdata/runner/config/config06.toml b/cli/kafka/testdata/runner/config/config06.toml
index 1a9384287..134bf2025 100644
--- a/cli/kafka/testdata/runner/config/config06.toml
+++ b/cli/kafka/testdata/runner/config/config06.toml
@@ -6,7 +6,7 @@ batch-size = 1
 batch-max-staleness = "5s"
 timeout = "5s"
 encode = "avro"
-schemaRegistryHost = "localhost:8081"
+schema-registry-url = "SCHEMA_REGISTRY_SERVICE"
 max-messages = MAX_MESSAGES
 
 [[fields]]
diff --git a/cli/kafka/testdata/runner/config/config07.toml b/cli/kafka/testdata/runner/config/config07.toml
index b2bd4e131..ed6aec8ba 100644
--- a/cli/kafka/testdata/runner/config/config07.toml
+++ b/cli/kafka/testdata/runner/config/config07.toml
@@ -6,7 +6,7 @@ batch-size = 1
 batch-max-staleness = "5s"
 timeout = "5s"
 encode = "avro"
-schemaRegistryHost = "localhost:8081"
+schema-registry-url = "SCHEMA_REGISTRY_SERVICE"
 max-messages = MAX_MESSAGES
 
 [[fields]]
diff --git a/idk/Dockerfile-cli-test b/idk/Dockerfile-cli-test
new file mode 100644
index 000000000..50790020e
--- /dev/null
+++ b/idk/Dockerfile-cli-test
@@ -0,0 +1,28 @@
+ARG GO_VERSION=1.19
+
+FROM golang:${GO_VERSION} as build_base
+
+WORKDIR /
+RUN ["apt-get","update","-y"]
+RUN ["apt-get","install","-y","git","unixodbc","unixodbc-dev","netcat", "build-essential","musl-tools"]
+RUN ["git", "clone", "https://github.com/edenhill/librdkafka.git"]
+WORKDIR /librdkafka
+RUN ["./configure", "--install-deps"] 
+RUN ["./configure", "--prefix", "/usr"]
+RUN ["make"]
+RUN ["make", "install"]
+
+FROM build_base
+
+ARG KAFKA_RUNNER_TEST_FEATUREBASE_HOST=pilosa:10101
+ARG KAFKA_RUNNER_TEST_FEATUREBASEGRPC_HOST=pilosa:20101
+ARG KAFKA_RUNNER_TEST_KAFKA_HOST=kafka:9092
+ARG KAFKA_RUNNER_TEST_REGISTRY_HOST=schema-registry:8081
+
+WORKDIR /go/src/github.com/featurebasedb/featurebase/
+
+COPY . .
+
+WORKDIR /go/src/github.com/featurebasedb/featurebase/cli/
+
+CMD ["go","test","-v","-mod=vendor","-tags=odbc,dynamic" "-run TestKafkaRunner","./..."]
diff --git a/idk/Makefile b/idk/Makefile
index 237b2aae8..f7d76ff42 100644
--- a/idk/Makefile
+++ b/idk/Makefile
@@ -338,3 +338,19 @@ update-mocks-%: install-mock-generator
 	$(eval AWS_SDK_VERSION := $(shell grep 'github.com/aws/aws-sdk-go' ../go.mod | cut -d ' ' -f 2))
 	echo Generating mock for AWS service $(AWS_SERVICE) and SDK version $(AWS_SDK_VERSION) && \
 	$(GOPATH)/bin/mockery --name $*API --output idktest/mocks --filename $(AWS_SERVICE).go --dir $(GOPATH)/pkg/mod/github.com/aws/aws-sdk-go@$(AWS_SDK_VERSION)/service/$(AWS_SERVICE)/$(AWS_SERVICE)iface
+
+
+# Integration Testing for CLI / fbsql tooling that integrates with IDK tooling.
+test-cli: 
+	$(MAKE) startup
+	$(MAKE) test-cli-integration
+
+TPKG ?= ../cli_test
+test-cli-integration: testenv vendor
+	$(DOCKER_COMPOSE) build cli-test
+	$(DOCKER_COMPOSE) run -e  KAFKA_RUNNER_TEST_FEATUREBASE_HOST=pilosa:10101 \
+				-e  KAFKA_RUNNER_TEST_FEATUREBASEGRPC_HOST=pilosa:20101 \
+				-e  KAFKA_RUNNER_TEST_KAFKA_HOST=kafka:9092 \
+                -e  KAFKA_RUNNER_TEST_REGISTRY_HOST=schema-registry:8081 \
+				-T cli-test bash -c "set -o pipefail; go test -v -mod=vendor -tags=odbc,dynamic -run TestKafkaRunner -covermode=atomic -coverpkg=$(TPKG) -coverprofile=/testdata/$(PROJECT)_base_coverage.out"
+
diff --git a/idk/docker-compose.yml b/idk/docker-compose.yml
index 87348df07..6fe634b7f 100644
--- a/idk/docker-compose.yml
+++ b/idk/docker-compose.yml
@@ -155,3 +155,19 @@ services:
     depends_on:
       - postgres
 
+  cli-test:
+    build:
+      context: ../.
+      dockerfile: ./idk/Dockerfile-cli-test
+    environment:
+      KAFKA_RUNNER_TEST_FEATUREBASE_HOST: pilosa:10101
+      KAFKA_RUNNER_TEST_FEATUREBASEGRPC_HOST: pilosa:20101
+      KAFKA_RUNNER_TEST_KAFKA_HOST: kafka:9092
+      KAFKA_RUNNER_TEST_REGISTRY_HOST: schema-registry:8081
+    volumes:
+      - ./testenv/certs:/certs
+      - ./docker-sasl/ssl_keys:/ssl_keys
+      - ./testdata:/testdata
+    depends_on:
+      - schema-registry
+
diff --git a/idk/ingest.go b/idk/ingest.go
index 1bb834d4f..a2710f19e 100644
--- a/idk/ingest.go
+++ b/idk/ingest.go
@@ -309,51 +309,13 @@ func (m *Main) clone() (*Main, error) {
 
 	index = schema.Index(m.Index)
 
-	// use a copy (schema race condition issu)
+	// use a copy (schema race condition issues)
 	mClone := *m
 	mClone.index = index
 
 	return &mClone, nil
 }
 
-// func (m *Main) clone() (*Main, error) {
-// 	var index *pilosaclient.Index
-// 	noOpSchemaManager := false
-
-// 	// If you have a schema manager, it does it's thing. Otherwise, it's a no
-// 	// opt manager. Then you get a default schema. If you get a default schema,
-// 	// you get a default index. This seems fine for IDK. However, for the CLI /
-// 	// SQL kafka runner, we use the m.index to create the dax.Table. If m.index
-// 	// is set to default values, then keys is false even when we don't want it
-// 	// to be. This was causing issue in buildBulkInsert in the batch package of
-// 	// the CLI.
-// 	switch m.SchemaManager.(type) {
-// 	case *nopSchemaManager:
-// 		noOpSchemaManager = true
-// 	}
-
-// 	schema, err := m.SchemaManager.Schema()
-// 	if err != nil {
-// 		return nil, err
-// 	}
-
-// 	if noOpSchemaManager && len(m.PrimaryKeyFields) > 0 {
-// 		// if IDK has PrimaryKeyFields, it expects keyed index
-// 		keys := pilosaclient.OptIndexKeys(true)
-// 		// most queries don't work with this set to false so set to true
-// 		exists := pilosaclient.OptIndexTrackExistence(true)
-// 		index = schema.Index(m.Index, keys, exists)
-// 	} else {
-// 		index = schema.Index(m.Index)
-// 	}
-
-// 	// use a copy (schema race condition issues)
-// 	mClone := *m
-// 	mClone.index = index
-
-// 	return &mClone, nil
-// }
-
 func (m *Main) runIngester(c int, l *msgCounter) error {
 	m.log.Printf("start ingester %d", c)
 	// TODO: actually implement cancellation and graceful shutdown
@@ -727,6 +689,13 @@ initialFetch:
 
 func (m *Main) Setup() (onFinishRun func(), err error) {
 	if m.basic {
+		// set up SchemaManager which is required for some logic in fbsql
+		// ingest. basic setup doesn't currently support tls so the tls config
+		// is ignored below
+		if _, err = m.setupClient(); err != nil {
+			return nil, errors.Wrap(err, "setting up client")
+		}
+
 		return m.basicSetup()
 	}
 	return m.setup()
@@ -744,8 +713,6 @@ func (m *Main) basicSetup() (onFinishRun func(), err error) {
 		return nil, errors.Wrap(err, "validating configuration")
 	}
 
-	_, err = m.setupClient()
-
 	// setup logging
 	var f *logger.FileWriter
 	var logOut io.Writer = os.Stderr
@@ -2203,7 +2170,7 @@ func (m *Main) newBatch(clientFields []*pilosaclient.Field) (pilosabatch.RecordB
 	ii := pilosaclient.FromClientIndex(m.index)
 	tbl := pilosacore.IndexInfoToTable(ii)
 
-	// Fields. // this is giving bad fieldInfos
+	// Fields.
 	fields := pilosaclient.FromClientFields(clientFields)
 
 	// If a custom Batcher has been defined, use that. Otherwise default to
diff --git a/idk/kafka/source.go b/idk/kafka/source.go
index 00adf0bff..2c5e418e7 100644
--- a/idk/kafka/source.go
+++ b/idk/kafka/source.go
@@ -48,7 +48,8 @@ type Source struct {
 	highmarks     []confluent.TopicPartition
 	client        *confluent.Consumer
 	recordChannel chan recordWithError
-	ConfigMap     *confluent.ConfigMap
+
+	ConfigMap *confluent.ConfigMap
 
 	// lastSchemaID and lastSchema keep track of the most recent
 	// schema in use. We expect this not to change often, but when it
@@ -189,10 +190,6 @@ func (s *Source) toPDKRecord(vals map[string]interface{}) []interface{} {
 	return data
 }
 
-func (s *Source) CommitMessages(recs []confluent.TopicPartition) ([]confluent.TopicPartition, error) {
-	return s.client.CommitOffsets(recs)
-}
-
 type Record struct {
 	src        *Source
 	topic      string
@@ -209,10 +206,6 @@ func (r *Record) StreamOffset() (string, uint64) {
 
 var _ idk.OffsetStreamRecord = &Record{}
 
-func (r *Record) Schema() interface{} {
-	return r.avroSchema
-}
-
 func (r *Record) Commit(ctx context.Context) error {
 	r.src.mu.Lock()
 	defer r.src.mu.Unlock()
@@ -232,9 +225,10 @@ func (r *Record) Commit(ctx context.Context) error {
 	p := int32(-1)
 	s := ""
 	r.src.highmarks = r.src.highmarks[:0]
-
 	// sort by increasing partition, decreasing offset
+
 	for _, x := range section {
+
 		if s != *x.Topic || p != x.Partition {
 			r.src.highmarks = append(r.src.highmarks, x)
 		}
@@ -246,6 +240,7 @@ func (r *Record) Commit(ctx context.Context) error {
 	if err != nil {
 		return errors.Wrap(err, "failed to commit messages")
 	}
+
 	if r.src.Verbose {
 		for _, o := range committedOffsets {
 			r.src.Log.Debugf("t: %v p: %v o: %v", *o.Topic, o.Partition, o.Offset)
@@ -254,7 +249,6 @@ func (r *Record) Commit(ctx context.Context) error {
 
 	r.src.spool = remaining
 	r.src.spoolBase = idx
-
 	return nil
 }
 
@@ -262,13 +256,21 @@ func (r *Record) Data() []interface{} {
 	return r.data
 }
 
+func (r *Record) Schema() interface{} {
+	return r.avroSchema
+}
+
+func (s *Source) CommitMessages(recs []confluent.TopicPartition) ([]confluent.TopicPartition, error) {
+	return s.client.CommitOffsets(recs)
+}
+
 // Open initializes the kafka source. (i.e. creating and configuring a consumer)
 // The configuration options for the confluentinc/confluent-kafka-go/kafka
 // libarary are: https://github.com/confluentinc/librdkafka/blob/master/CONFIGURATION.md
 func (s *Source) Open() error {
 	cfg, err := common.SetupConfluent(&s.ConfluentCommand)
 	if err != nil {
-		return err
+		return errors.Wrap(err, "setting up confluent command")
 	}
 	s.ConfigMap = cfg
 
@@ -284,8 +286,6 @@ func (s *Source) Open() error {
 		}
 		s.httpClient = getHTTPClient(tlsConfig)
 	}
-
-	// group
 	if s.Group != "" {
 		err = s.ConfigMap.SetKey("group.id", s.Group)
 		if err != nil {
@@ -321,8 +321,6 @@ func (s *Source) Open() error {
 		return errors.Wrap(err, "new consumer")
 	}
 
-	// by default, Kafka will use the stored offset (the latest committed message) and continue on from there.
-	// to skip old msgs, use rebalanceCbSkipOld to manually set offset to the end
 	err = cl.SubscribeTopics(s.Topics, nil)
 	if err != nil {
 		return errors.Wrap(err, "subscribe topics")
@@ -335,7 +333,6 @@ func (s *Source) Open() error {
 	s.client = cl
 	s.opened = true
 	s.wg.Add(1)
-
 	go func() {
 		s.generator()
 	}()
@@ -343,6 +340,23 @@ func (s *Source) Open() error {
 	return nil
 }
 
+func (s *Source) cleanRegistryURL() error {
+	// We can't immediately url.Parse the RegistryURL because parsing
+	// a host without a scheme is invalid. First we'll check for a
+	// scheme and add the default http:// if needed.
+	if !strings.Contains(s.SchemaRegistryURL, "://") {
+		s.SchemaRegistryURL = "http://" + s.SchemaRegistryURL
+	}
+
+	SchemaRegistryURL, err := url.Parse(s.SchemaRegistryURL)
+	if err != nil {
+		return errors.Wrap(err, "parse registry URL")
+	}
+
+	s.SchemaRegistryURL = SchemaRegistryURL.String()
+	return nil
+}
+
 func (c *Source) generator() {
 	defer func() {
 		close(c.recordChannel)
@@ -460,23 +474,6 @@ func (s *Source) Close() error {
 	return nil
 }
 
-func (s *Source) cleanRegistryURL() error {
-	// We can't immediately url.Parse the RegistryURL because parsing
-	// a host without a scheme is invalid. First we'll check for a
-	// scheme and add the default http:// if needed.
-	if !strings.Contains(s.SchemaRegistryURL, "://") {
-		s.SchemaRegistryURL = "http://" + s.SchemaRegistryURL
-	}
-
-	SchemaRegistryURL, err := url.Parse(s.SchemaRegistryURL)
-	if err != nil {
-		return errors.Wrap(err, "parse registry URL")
-	}
-
-	s.SchemaRegistryURL = SchemaRegistryURL.String()
-	return nil
-}
-
 // TODO change name
 func (s *Source) decodeAvroValueWithSchemaRegistry(val []byte) (interface{}, avro.Schema, error) {
 	if len(val) < 6 || val[0] != 0 {
diff --git a/idk/kafka_sasl/source.go b/idk/kafka_sasl/source.go
index feb063c4e..d5958cd9f 100644
--- a/idk/kafka_sasl/source.go
+++ b/idk/kafka_sasl/source.go
@@ -1,10 +1,8 @@
 package kafka_sasl
 
 import (
-	"bytes"
 	"context"
 	"encoding/json"
-	"fmt"
 	"io"
 	"os"
 	"sort"
@@ -14,7 +12,6 @@ import (
 
 	confluent "github.com/confluentinc/confluent-kafka-go/kafka"
 	"github.com/featurebasedb/featurebase/v3/idk"
-	"github.com/featurebasedb/featurebase/v3/idk/common"
 	"github.com/featurebasedb/featurebase/v3/logger"
 	"github.com/pkg/errors"
 )
@@ -24,14 +21,13 @@ import (
 // achieve concurrency, create multiple Sources.
 type Source struct {
 	idk.ConfluentCommand
-	Topics               []string
-	Group                string
-	Log                  logger.Logger
-	Timeout              time.Duration
-	SkipOld              bool
-	Verbose              bool
-	AllowMissingFields   bool
-	consumerCloseTimeout time.Duration
+	Topics             []string
+	Group              string
+	Log                logger.Logger
+	Timeout            time.Duration
+	SkipOld            bool
+	Verbose            bool
+	AllowMissingFields bool
 
 	// Header is a file referencing a file containing JSON header configuration.
 	Header string
@@ -63,15 +59,14 @@ type Source struct {
 // NewSource gets a new Source
 func NewSource() *Source {
 	src := &Source{
-		ConfluentCommand:     idk.ConfluentCommand{},
-		Topics:               []string{"test"},
-		Group:                "group0",
-		Log:                  logger.NopLogger,
-		recordChannel:        make(chan recordWithError),
-		quit:                 make(chan struct{}),
-		ConfigMap:            &confluent.ConfigMap{},
-		highmarks:            make([]confluent.TopicPartition, 0),
-		consumerCloseTimeout: 30,
+		ConfluentCommand: idk.ConfluentCommand{},
+		Topics:           []string{"test"},
+		Group:            "group0",
+		Log:              logger.NopLogger,
+		recordChannel:    make(chan recordWithError),
+		quit:             make(chan struct{}),
+		ConfigMap:        &confluent.ConfigMap{},
+		highmarks:        make([]confluent.TopicPartition, 0),
 	}
 	src.KafkaBootstrapServers = []string{"localhost:9092"}
 
@@ -90,9 +85,8 @@ func (s *Source) Record() (idk.Record, error) {
 	case context.DeadlineExceeded:
 		return nil, idk.ErrFlush
 	default:
-		return nil, errors.Wrap(rec.Err, "failed to fetch record from Kafka")
+		return nil, errors.Wrap(rec.Err, "failed to fetch record from Kafka Confluent")
 	}
-
 	if rec.Record == nil {
 		return nil, idk.ErrFlush
 	}
@@ -171,6 +165,7 @@ func (r *Record) Schema() interface{} { return nil }
 func (r *Record) Commit(ctx context.Context) error {
 	r.src.mu.Lock()
 	defer r.src.mu.Unlock()
+
 	idx, base := r.idx, r.src.spoolBase
 	if idx < base {
 		return errors.New("cannot commit a record that has already been committed")
@@ -184,6 +179,7 @@ func (r *Record) Commit(ctx context.Context) error {
 		}
 		return section[i].Offset > section[j].Offset
 	})
+	// calculate the high marks
 	p := int32(-1)
 	s := ""
 	r.src.highmarks = r.src.highmarks[:0]
@@ -195,17 +191,11 @@ func (r *Record) Commit(ctx context.Context) error {
 		}
 		p = x.Partition
 		s = *x.Topic
-
 	}
-	committedOffsets, err := r.src.CommitMessages(r.src.highmarks)
+	_, err := r.src.CommitMessages(r.src.highmarks)
 	if err != nil {
 		return errors.Wrap(err, "failed to commit messages")
 	}
-	if r.src.Verbose {
-		for _, o := range committedOffsets {
-			r.src.Log.Debugf("t: %v p: %v o: %v", *o.Topic, o.Partition, o.Offset)
-		}
-	}
 
 	r.src.spool = remaining
 	r.src.spoolBase = idx
@@ -219,17 +209,12 @@ func (r *Record) Data() []interface{} {
 
 // Open initializes the kafka source.
 func (s *Source) Open() error {
-	cfg, err := common.SetupConfluent(&s.ConfluentCommand)
-	if err != nil {
-		return err
-	}
-	s.ConfigMap = cfg
-
 	if len(s.Header) == 0 && len(s.HeaderFields) == 0 {
-		return errors.New("needs header specification file (file or fields)")
+		return errors.New("needs header specification (from file or from existing fields)")
 	}
 
 	var headerData []byte
+	var err error
 	if s.Header != "" {
 		headerData, err = os.ReadFile(s.Header)
 		if err != nil {
@@ -256,30 +241,21 @@ func (s *Source) Open() error {
 			return err
 		}
 	}
-
 	// when there is no initial offset in Kafka or if the current offset does not exist any more,
 	// use this as starting offset:
 	//  "earliest": automatically reset the offset to the earliest offset
 	//  "latest": automatically reset the offset to the latest offset
-	offset := "earliest"
-	if s.SkipOld {
-		offset = "latest"
-	}
-	err = s.ConfigMap.SetKey("auto.offset.reset", offset)
+	err = s.ConfigMap.SetKey("auto.offset.reset", "earliest")
 	if err != nil {
 		return err
 	}
-	if s.Verbose {
-		buf := bytes.NewBufferString("Confluent Config Map:")
-		encoder := json.NewEncoder(buf)
-		err = encoder.Encode(s.ConfigMap)
+	if s.SkipOld {
+		err = s.ConfigMap.SetKey("auto.offset.reset", "latest")
 		if err != nil {
 			return err
 		}
-		s.Log.Debugf(buf.String())
-		stv, iv := confluent.LibraryVersion()
-		s.Log.Debugf("version:(%v) %v", iv, stv)
 	}
+
 	cl, err := confluent.NewConsumer(s.ConfigMap)
 	if err != nil {
 		return errors.Wrap(err, "new consumer")
@@ -292,10 +268,6 @@ func (s *Source) Open() error {
 		return errors.Wrap(err, "subscribe topics")
 	}
 
-	if s.Verbose {
-		s.Log.Debugf("subscribed to %v", s.Topics)
-	}
-
 	s.client = cl
 	s.opened = true
 	s.wg.Add(1)
@@ -317,9 +289,6 @@ func (c *Source) generator() {
 		select {
 
 		case <-c.quit:
-			if c.Verbose {
-				c.Log.Debugf("source quit")
-			}
 			return
 		default:
 			ev := c.client.Poll(100)
@@ -333,9 +302,6 @@ func (c *Source) generator() {
 			case confluent.AssignedPartitions:
 				err := c.client.Assign(e.Partitions)
 				if err != nil {
-					if c.Verbose {
-						c.Log.Debugf("quit AssignedParitions")
-					}
 					return
 				}
 				// If we received an `RevokedPartitions` event, we need to revoke this
@@ -344,9 +310,6 @@ func (c *Source) generator() {
 			case confluent.RevokedPartitions:
 				err := c.client.Unassign()
 				if err != nil {
-					if c.Verbose {
-						c.Log.Debugf("quit RevokeParkitions")
-					}
 					return
 				}
 
@@ -358,9 +321,6 @@ func (c *Source) generator() {
 				select {
 				case c.recordChannel <- msg:
 				case <-c.quit:
-					if c.Verbose {
-						c.Log.Debugf("source quit Error")
-					}
 					return
 				}
 
@@ -377,17 +337,9 @@ func (c *Source) generator() {
 				select {
 				case c.recordChannel <- msg:
 				case <-c.quit:
-					if c.Verbose {
-						c.Log.Debugf("source quit Message")
-					}
 					return
 				}
-			case confluent.OffsetsCommitted:
-				c.Log.Debugf("commited %s", e)
 			default:
-				if c.Verbose {
-					c.Log.Debugf("ignored %#v", ev)
-				}
 				continue // consumer doesn't care about all event types (e.g. OffsetsCommitted)
 			}
 		}
@@ -398,26 +350,10 @@ func (c *Source) generator() {
 func (s *Source) Close() error {
 	if s.client != nil {
 		if s.opened { // only close opened sources
-			var err error
-			closedReturned := make(chan error, 1)
-			// send quit message to polling routine & wait for it to exit
 			s.quit <- struct{}{}
 			s.wg.Wait()
-			s.Log.Debugf("Trying to close consumer %s...", s.client.String())
-			go func() {
-				closedReturned <- s.client.Close()
-			}()
-			start := time.Now()
-			select {
-			case err = <-closedReturned:
-				if err == nil {
-					s.Log.Debugf("Successfully closed consumer %s!", s.client.String())
-					s.opened = false
-				}
-			case <-time.After(time.Duration(s.consumerCloseTimeout * 1000 * 1000 * 1000)):
-				err = fmt.Errorf("unable to properly close consumer %s after %f seconds", s.client.String(), time.Since(start).Seconds())
-			}
-
+			err := s.client.Close()
+			s.opened = false
 			return errors.Wrap(err, "closing kafka consumer")
 		}
 	}
diff --git a/wire_response.go b/wire_response.go
index 90eac3a1e..dfb08bc00 100644
--- a/wire_response.go
+++ b/wire_response.go
@@ -5,7 +5,6 @@ import (
 	"encoding/json"
 	"fmt"
 	"log"
-	"sort"
 	"strings"
 	"time"
 
@@ -211,15 +210,6 @@ func (ii IDSet) String() string {
 	return sb.String()
 }
 
-// SortedInt64Slice returns the values in a IDSet field in a int64 slice that is
-// sorted
-func (ii IDSet) SortedInt64Slice() []int64 {
-	var idSetSlice = make([]int64, len(ii))
-	copy(idSetSlice, ii)
-	sort.Slice(idSetSlice, func(i, j int) bool { return idSetSlice[i] < idSetSlice[j] })
-	return idSetSlice
-}
-
 // StringSet is a return type specific to SQLResponse types.
 type StringSet []string
 
@@ -237,15 +227,6 @@ func (ss StringSet) String() string {
 	return sb.String()
 }
 
-// SortedStringSlice returns the values in a StringSet field in a string slice
-// that is sorted
-func (ss StringSet) SortedStringSlice() []string {
-	var stringSetSlice = make([]string, len(ss))
-	copy(stringSetSlice, ss)
-	sort.Strings(stringSetSlice)
-	return stringSetSlice
-}
-
 // ShowColumnsResponse returns a structure which is specific to a `SHOW COLUMNS`
 // statement, derived from the results in the WireQueryResponse. This is kind of
 // a crude way to unmarshal a WireQueryResponse into a type which is specific to