return specific schema registry unavailable error.

Author: hchienjo

avroregistry/errors.go

@@ -0,0 +1,15 @@
+package avroregistry
+
+import (
+	"fmt"
+)
+
+// UnavailableError reports an error when the schema registry is unavailable.
+type UnavailableError struct {
+	Cause error
+}
+
+// Error implements the error interface.
+func (m *UnavailableError) Error() string {
+	return fmt.Sprintf("schema registry unavailability caused by: %v", m.Cause)
+}

avroregistry/registry.go

@@ -190,7 +190,7 @@ func (r *Registry) doRequest(req *http.Request, result interface{}) error {
 		resp, err := http.DefaultClient.Do(req)
 		if err != nil {
 			if !attempt.More() || !isTemporaryError(err) {
-				return err
+				return &UnavailableError{err}
 			}
 			continue
 		}
@@ -201,7 +201,7 @@ func (r *Registry) doRequest(req *http.Request, result interface{}) error {
 		if !attempt.More() {
 			return err
 		}
-		if err, ok := err.(*apiError); ok && err.StatusCode/100 != 5 {
+		if err, ok := err.(*apiError); ok && err.StatusCode != http.StatusInternalServerError {
 			// It's not a 5xx error. We want to retry on 5xx
 			// errors, because the Confluent Avro registry
 			// can occasionally return them as a matter of

avroregistry/registry_test.go

@@ -36,6 +36,31 @@ func TestRegister(t *testing.T) {
 	c.Assert(id1, qt.Equals, id)
 }
 
+func TestSchemaRegistryUnavailableError(t *testing.T) {
+	c := qt.New(t)
+	ctx := context.Background()
+
+	testServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+
+	}))
+
+	// close the server
+	testServer.Close()
+
+	registry, err := avroregistry.New(avroregistry.Params{
+		ServerURL:     testServer.URL,
+		RetryStrategy: noRetry,
+	})
+	c.Assert(err, qt.IsNil)
+
+	type R struct {
+		X int
+	}
+
+	_, err = registry.Register(ctx, randomString(), schemaOf(nil, R{}))
+	c.Assert(err, qt.ErrorAs, &avroregistry.UnavailableError{})
+}
+
 func TestRegisterWithEmptyStruct(t *testing.T) {
 	c := qt.New(t)