Added support for DeleteTopicsRequest V1

delete_topics_request.go

@@ -3,6 +3,7 @@ package sarama
 import "time"
 
 type DeleteTopicsRequest struct {
+	Version int16
 	Topics  []string
 	Timeout time.Duration
 }
@@ -25,6 +26,7 @@ func (d *DeleteTopicsRequest) decode(pd packetDecoder, version int16) (err error
 		return err
 	}
 	d.Timeout = time.Duration(timeout) * time.Millisecond
+	d.Version = version
 	return nil
 }
 
@@ -33,9 +35,14 @@ func (d *DeleteTopicsRequest) key() int16 {
 }
 
 func (d *DeleteTopicsRequest) version() int16 {
-	return 0
+	return d.Version
 }
 
 func (d *DeleteTopicsRequest) requiredVersion() KafkaVersion {
-	return V0_10_1_0
+	switch d.Version {
+	case 1:
+		return V0_11_0_0
+	default:
+		return V0_10_1_0
+	}
 }

delete_topics_request_test.go

@@ -12,8 +12,19 @@ var deleteTopicsRequest = []byte{
 	0, 0, 0, 100,
 }
 
-func TestDeleteTopicsRequest(t *testing.T) {
+func TestDeleteTopicsRequestV0(t *testing.T) {
 	req := &DeleteTopicsRequest{
+		Version: 0,
+		Topics:  []string{"topic", "other"},
+		Timeout: 100 * time.Millisecond,
+	}
+
+	testRequest(t, "", req, deleteTopicsRequest)
+}
+
+func TestDeleteTopicsRequestV1(t *testing.T) {
+	req := &DeleteTopicsRequest{
+		Version: 1,
 		Topics:  []string{"topic", "other"},
 		Timeout: 100 * time.Millisecond,
 	}