JVMCBC-1068: Explicitly handle FeatureNotAvailable for Magma on CE

Magma buckets are only available on EE, so we need to make sure that
if a bucket is created on CE with magma set as the storage backend
a FeatureNotAvailable is returned.

Change-Id: I682968db6b1744265b4784acfd99cf79f252368e
Reviewed-on: https://review.couchbase.org/c/couchbase-jvm-clients/+/171251
Tested-by: Build Bot <build@couchbase.com>
Reviewed-by: Michael Reiche <michael.reiche@couchbase.com>

core-io/src/main/java/com/couchbase/client/core/error/FeatureNotAvailableException.java

@@ -57,7 +57,11 @@ public static FeatureNotAvailableException collections() {
 
   public static FeatureNotAvailableException collectionsForMemcached() {
     return new FeatureNotAvailableException("Non-Default collections are not supported for memcached buckets.");
+  }
 
+  public static FeatureNotAvailableException communityEdition(final String feature) {
+    return new FeatureNotAvailableException("The feature [" + feature + "] is not supported in the Community " +
+      "Edition (CE) of Couchbase Server.");
   }
 
 }

core-io/src/main/java/com/couchbase/client/core/io/netty/manager/NonChunkedManagerMessageHandler.java

@@ -18,6 +18,7 @@
 
 import com.couchbase.client.core.deps.io.netty.handler.codec.http.HttpResponseStatus;
 import com.couchbase.client.core.endpoint.BaseEndpoint;
+import com.couchbase.client.core.error.FeatureNotAvailableException;
 import com.couchbase.client.core.error.HttpStatusCodeException;
 import com.couchbase.client.core.error.QuotaLimitedException;
 import com.couchbase.client.core.error.RateLimitedException;
@@ -42,6 +43,12 @@ protected Exception failRequestWith(HttpResponseStatus status, String content, N
       content
     );
 
+    if (status.equals(HttpResponseStatus.BAD_REQUEST)) {
+      if (content.contains("Magma is supported in enterprise edition only")) {
+        return FeatureNotAvailableException.communityEdition("Storage Backend: Magma");
+      }
+    }
+
     if (status.equals(HttpResponseStatus.TOO_MANY_REQUESTS)) {
       if (content.contains("num_concurrent_requests")
         || content.contains("ingress")