Reject delivery.report.only.error=true on producer creation (#306)

.. to avoid a cgoDr memory leak per successful message.

kafka/producer.go

@@ -390,7 +390,15 @@ func NewProducer(conf *ConfigMap) (*Producer, error) {
 	// the original is not mutated.
 	confCopy := conf.clone()
 
-	v, err := confCopy.extract("go.batch.producer", false)
+	v, err := confCopy.extract("delivery.report.only.error", false)
+	if v == true {
+		// FIXME: The filtering of successful DRs must be done in
+		//        the Go client to avoid cgoDr memory leaks.
+		return nil, newErrorFromString(ErrUnsupportedFeature,
+			"delivery.report.only.error=true is not currently supported by the Go client")
+	}
+
+	v, err = confCopy.extract("go.batch.producer", false)
 	if err != nil {
 		return nil, err
 	}

kafka/producer_test.go

@@ -216,3 +216,14 @@ func TestProducerBufferSafety(t *testing.T) {
 
 	p.Close()
 }
+
+// TestProducerInvalidConfig verifies that invalid configuration is handled correctly.
+func TestProducerInvalidConfig(t *testing.T) {
+
+	_, err := NewProducer(&ConfigMap{
+		"delivery.report.only.error": true,
+	})
+	if err == nil {
+		t.Fatalf("Expected NewProducer() to fail with delivery.report.only.error set")
+	}
+}