integration tests (confluentinc#9)

* integration tests
moved integration tests from example/ to kafka/

* improvements based on code review feedback

* fixed Offset.Set() API bug

* fixed code that caused crash in integration tests

* changed CC and PC to consumerCtrl and producerCtrl for readability

* integration tests
moved integration tests from example/ to kafka/

improvements based on code review feedback

fixed Offset.Set() API bug

fixed code that caused crash in integration tests

changed CC and PC to consumerCtrl and producerCtrl for readability

* improvements based on code review feedback

* improvements based on code review feedback

* fix code formatting

Author: hqin

kafka/README

@@ -15,7 +15,7 @@ Testing
 =======
 
 
-Some of the tests included in this directory, the benchmark tests in particular,
+Some of the tests included in this directory, the benchmark and integration tests in particular,
 require an existing Kafka cluster and a testconf.json configuration file to
 provide tests with bootstrap brokers, topic name, etc.
 
@@ -39,7 +39,10 @@ To run benchmark tests:
 
     $ go test -bench .
 
+For the code coverage:
 
+    $ go test -coverprofile=coverage.out -bench=.
+    $ go tool cover -func=coverage.out
 
 
 Build tags (static linking)

kafka/config_test.go

@@ -0,0 +1,89 @@
+/**
+ * Copyright 2016 Confluent Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package kafka
+
+import (
+	"fmt"
+	"testing"
+)
+
+// A custom type with Stringer interface to be used to test config map APIs
+type HostPortType struct {
+	Host string
+	Port int
+}
+
+// implements String() interface
+func (hp HostPortType) String() string {
+	return fmt.Sprintf("%s:%d", hp.Host, hp.Port)
+}
+
+//Test config map APIs
+func TestConfigMapAPIs(t *testing.T) {
+	config := &ConfigMap{}
+
+	// set a good key via SetKey()
+	err := config.SetKey("bootstrap.servers", testconf.Brokers)
+	if err != nil {
+		t.Errorf("Failed to set key via SetKey(). Error: %s\n", err)
+	}
+
+	// test custom Stringer type
+	hostPort := HostPortType{Host: "localhost", Port: 9092}
+	err = config.SetKey("bootstrap.servers", hostPort)
+	if err != nil {
+		t.Errorf("Failed to set custom Stringer type via SetKey(). Error: %s\n", err)
+	}
+
+	// test boolean type
+	err = config.SetKey("{topic}.produce.offset.report", true)
+	if err != nil {
+		t.Errorf("Failed to set key via SetKey(). Error: %s\n", err)
+	}
+
+	// test offset literal string
+	err = config.SetKey("{topic}.auto.offset.reset", "earliest")
+	if err != nil {
+		t.Errorf("Failed to set key via SetKey(). Error: %s\n", err)
+	}
+
+	//test offset constant
+	err = config.SetKey("{topic}.auto.offset.reset", OffsetBeginning)
+	if err != nil {
+		t.Errorf("Failed to set key via SetKey(). Error: %s\n", err)
+	}
+
+	//test integer offset
+	err = config.SetKey("{topic}.message.timeout.ms", 10)
+	if err != nil {
+		t.Errorf("Failed to set integer value via SetKey(). Error: %s\n", err)
+	}
+
+	// set a good key-value pair via Set()
+	err = config.Set("group.id=test.id")
+	if err != nil {
+		t.Errorf("Failed to set key-value pair via Set(). Error: %s\n", err)
+	}
+
+	// negative test cases
+	// set a bad key-value pair via Set()
+	err = config.Set("group.id:test.id")
+	if err == nil {
+		t.Errorf("Expected failure when setting invalid key-value pair via Set()")
+	}
+
+}

kafka/event_test.go

@@ -0,0 +1,40 @@
+/**
+ * Copyright 2016 Confluent Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package kafka
+
+import (
+	"testing"
+)
+
+// TestEventAPIs dry-tests the public event related APIs, no broker is needed.
+func TestEventAPIs(t *testing.T) {
+	assignedPartitions := AssignedPartitions{}
+	t.Logf("%s\n", assignedPartitions.String())
+
+	revokedPartitions := RevokedPartitions{}
+	t.Logf("%s\n", revokedPartitions.String())
+
+	topic := "test"
+	partition := PartitionEOF{Topic: &topic}
+	t.Logf("%s\n", partition.String())
+
+	partition = PartitionEOF{}
+	t.Logf("%s\n", partition.String())
+
+	committedOffsets := OffsetsCommitted{}
+	t.Logf("%s\n", committedOffsets.String())
+}