WL#12459, DevAPI: Support connection-attributes.

Author: soklakov

CHANGES

@@ -3,6 +3,8 @@
 
 Version 8.0.16
 
+  - WL#12459, DevAPI: Support connection-attributes.
+
   - Fix for Bug#25650385, GETBYTE() RETURNS ERROR FOR BINARY() FLD.
 
   - Fix for Bug#27784363, MYSQL 8.0 JDBC DRIVER THROWS NUMBERFORMATEXCEPTION FOR TEXT DATA

src/main/core-api/java/com/mysql/cj/conf/PropertyDefinitions.java

@@ -708,8 +708,6 @@ public enum AuthMech { // xdevapi.auth
                 new BooleanPropertyDefinition(PropertyKey.readOnlyPropagatesToServer, DEFAULT_VALUE_TRUE, RUNTIME_MODIFIABLE,
                         Messages.getString("ConnectionProperties.readOnlyPropagatesToServer"), "5.1.35", CATEGORY_PERFORMANCE, Integer.MIN_VALUE),
 
-                // TODO improve X DevAPI properties descriptions
-
                 new BooleanPropertyDefinition(PropertyKey.xdevapiUseAsyncProtocol, DEFAULT_VALUE_FALSE, RUNTIME_NOT_MODIFIABLE,
                         Messages.getString("ConnectionProperties.useAsyncProtocol"), "6.0.0", CATEGORY_XDEVAPI, Integer.MIN_VALUE),
                 new EnumPropertyDefinition<>(PropertyKey.xdevapiSSLMode, XdevapiSslMode.REQUIRED, RUNTIME_MODIFIABLE,
@@ -726,6 +724,8 @@ public enum AuthMech { // xdevapi.auth
                         "8.0.8", CATEGORY_XDEVAPI, Integer.MIN_VALUE),
                 new IntegerPropertyDefinition(PropertyKey.xdevapiConnectTimeout, 10000, RUNTIME_MODIFIABLE,
                         Messages.getString("ConnectionProperties.xdevapiConnectTimeout"), "8.0.13", CATEGORY_XDEVAPI, Integer.MIN_VALUE, 0, Integer.MAX_VALUE),
+                new StringPropertyDefinition(PropertyKey.xdevapiConnectionAttributes, DEFAULT_VALUE_NULL_STRING, RUNTIME_NOT_MODIFIABLE,
+                        Messages.getString("ConnectionProperties.xdevapiConnectionAttributes"), "8.0.16", CATEGORY_XDEVAPI, Integer.MIN_VALUE),
 
                 new BooleanPropertyDefinition(PropertyKey.enableEscapeProcessing, DEFAULT_VALUE_TRUE, RUNTIME_MODIFIABLE,
                         Messages.getString("ConnectionProperties.enableEscapeProcessing"), "6.0.1", CATEGORY_PERFORMANCE, Integer.MIN_VALUE),

src/main/core-api/java/com/mysql/cj/conf/PropertyKey.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2018, Oracle and/or its affiliates. All rights reserved.
+ * Copyright (c) 2018, 2019, Oracle and/or its affiliates. All rights reserved.
  *
  * This program is free software; you can redistribute it and/or modify it under
  * the terms of the GNU General Public License, version 2.0, as published by the
@@ -239,6 +239,7 @@ public enum PropertyKey {
     xdevapiAsyncResponseTimeout("xdevapi.asyncResponseTimeout", "xdevapiAsyncResponseTimeout", true), //
     xdevapiAuth("xdevapi.auth", "xdevapiAuth", true), //
     xdevapiConnectTimeout("xdevapi.connect-timeout", "xdevapiConnectTimeout", true), //
+    xdevapiConnectionAttributes("xdevapi.connection-attributes", "xdevapiConnectionAttributes", true), //
     xdevapiSSLMode("xdevapi.ssl-mode", "xdevapiSSLMode", true), //
     xdevapiSSLTrustStoreUrl("xdevapi.ssl-truststore", "xdevapiSSLTruststore", true), //
     xdevapiSSLTrustStoreType("xdevapi.ssl-truststore-type", "xdevapiSSLTruststoreType", true), //

src/main/protocol-impl/java/com/mysql/cj/protocol/x/XMessageBuilder.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2015, 2018, Oracle and/or its affiliates. All rights reserved.
+ * Copyright (c) 2015, 2019, Oracle and/or its affiliates. All rights reserved.
  *
  * This program is free software; you can redistribute it and/or modify it under
  * the terms of the GNU General Public License, version 2.0, as published by the
@@ -69,6 +69,7 @@
 import com.mysql.cj.x.protobuf.MysqlxCrud.Update;
 import com.mysql.cj.x.protobuf.MysqlxCrud.UpdateOperation;
 import com.mysql.cj.x.protobuf.MysqlxCrud.UpdateOperation.UpdateType;
+import com.mysql.cj.x.protobuf.MysqlxDatatypes;
 import com.mysql.cj.x.protobuf.MysqlxDatatypes.Any;
 import com.mysql.cj.x.protobuf.MysqlxDatatypes.Object.Builder;
 import com.mysql.cj.x.protobuf.MysqlxDatatypes.Object.ObjectField;
@@ -96,11 +97,24 @@ public XMessage buildCapabilitiesGet() {
         return new XMessage(CapabilitiesGet.getDefaultInstance());
     }
 
-    public XMessage buildCapabilitiesSet(String name, Object value) {
-        Any v = ExprUtil.argObjectToScalarAny(value);
-        Capability cap = Capability.newBuilder().setName(name).setValue(v).build();
-        Capabilities caps = Capabilities.newBuilder().addCapabilities(cap).build();
-        return new XMessage(CapabilitiesSet.newBuilder().setCapabilities(caps).build());
+    @SuppressWarnings("unchecked")
+    public XMessage buildCapabilitiesSet(Map<String, Object> keyValuePair) {
+        Capabilities.Builder capsB = Capabilities.newBuilder();
+        keyValuePair.forEach((k, v) -> {
+            Any val;
+            if (XServerCapabilities.KEY_SESSION_CONNECT_ATTRS.equals(k)) {
+                MysqlxDatatypes.Object.Builder attrB = MysqlxDatatypes.Object.newBuilder();
+                ((Map<String, String>) v)
+                        .forEach((name, value) -> attrB.addFld(ObjectField.newBuilder().setKey(name).setValue(ExprUtil.buildAny(value)).build()));
+                val = Any.newBuilder().setType(Any.Type.OBJECT).setObj(attrB).build();
+
+            } else {
+                val = ExprUtil.argObjectToScalarAny(v);
+            }
+            Capability cap = Capability.newBuilder().setName(k).setValue(val).build();
+            capsB.addCapabilities(cap);
+        });
+        return new XMessage(CapabilitiesSet.newBuilder().setCapabilities(capsB).build());
     }
 
     public XMessage buildDocInsert(String schemaName, String collectionName, List<String> json, boolean upsert) {

src/main/protocol-impl/java/com/mysql/cj/protocol/x/XProtocol.java

@@ -36,12 +36,15 @@
 import java.io.InputStream;
 import java.nio.channels.CompletionHandler;
 import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.CompletableFuture;
 
 import com.mysql.cj.CharsetMapping;
+import com.mysql.cj.Constants;
+import com.mysql.cj.Messages;
 import com.mysql.cj.QueryResult;
 import com.mysql.cj.Session;
 import com.mysql.cj.TransactionEventHandler;
@@ -61,6 +64,7 @@
 import com.mysql.cj.exceptions.ExceptionInterceptor;
 import com.mysql.cj.exceptions.FeatureNotAvailableException;
 import com.mysql.cj.exceptions.SSLParamsException;
+import com.mysql.cj.exceptions.WrongArgumentException;
 import com.mysql.cj.protocol.AbstractProtocol;
 import com.mysql.cj.protocol.ColumnDefinition;
 import com.mysql.cj.protocol.ExportControlled;
@@ -110,6 +114,8 @@ public class XProtocol extends AbstractProtocol<XMessage> implements Protocol<XM
 
     public String defaultSchemaName;
 
+    private Map<String, Object> clientCapabilities = new HashMap<>();
+
     public XProtocol(String host, int port, String defaultSchema, PropertySet propertySet) {
 
         this.defaultSchemaName = defaultSchema;
@@ -168,16 +174,14 @@ public ServerSession getServerSession() {
     }
 
     /**
-     * Set a capability of current session. Must be done before authentication ({@link #changeUser(String, String, String)}).
+     * Set client capabilities of current session. Must be done before authentication ({@link #changeUser(String, String, String)}).
      * 
-     * @param name
-     *            capability name
-     * @param value
-     *            capability value
+     * @param keyValuePair
+     *            capabilities name/value map
      */
-    public void setCapability(String name, Object value) {
-        ((XServerCapabilities) getServerSession().getCapabilities()).setCapability(name, value);
-        this.sender.send(((XMessageBuilder) this.messageBuilder).buildCapabilitiesSet(name, value));
+    public void sendCapabilities(Map<String, Object> keyValuePair) {
+        keyValuePair.forEach((k, v) -> ((XServerCapabilities) getServerSession().getCapabilities()).setCapability(k, v));
+        this.sender.send(((XMessageBuilder) this.messageBuilder).buildCapabilitiesSet(keyValuePair));
         readOk();
     }
 
@@ -187,13 +191,29 @@ public void negotiateSSLConnection(int packLength) {
             throw new CJConnectionFeatureNotAvailableException();
         }
 
-        if (!((XServerCapabilities) this.serverSession.getCapabilities()).hasCapability("tls")) {
+        if (!((XServerCapabilities) this.serverSession.getCapabilities()).hasCapability(XServerCapabilities.KEY_TLS)) {
             throw new CJCommunicationsException("A secure connection is required but the server is not configured with SSL.");
         }
 
         // the message reader is async and is always "reading". we need to stop it to use the socket for the TLS handshake
         this.reader.stopAfterNextMessage();
-        setCapability("tls", true);
+
+        this.clientCapabilities.put(XServerCapabilities.KEY_TLS, true);
+
+        try {
+            sendCapabilities(this.clientCapabilities);
+        } catch (XProtocolError e) {
+            // XProtocolError: ERROR 5002 (HY000) Capability 'session_connect_attrs' doesn't exist
+            // happens when connecting to xplugin which doesn't support this feature.
+            // Just ignore this error and retry with reduced clientCapabilities
+            if (e.getErrorCode() != 5002 && !e.getMessage().contains(XServerCapabilities.KEY_SESSION_CONNECT_ATTRS)) {
+                throw e;
+            }
+            this.clientCapabilities.remove(XServerCapabilities.KEY_SESSION_CONNECT_ATTRS);
+            this.reader.start();
+            this.reader.stopAfterNextMessage();
+            sendCapabilities(this.clientCapabilities);
+        }
 
         try {
             this.socketConnection.performTlsHandshake(null); //(this.serverSession);
@@ -237,6 +257,13 @@ public void beforeHandshake() {
 
         this.serverSession.setCapabilities(readServerCapabilities());
 
+        // connection attributes
+        String attributes = this.propertySet.getStringProperty(PropertyKey.xdevapiConnectionAttributes).getValue();
+        if (attributes == null || !attributes.equalsIgnoreCase("false")) {
+            Map<String, String> attMap = getConnectionAttributesMap("true".equalsIgnoreCase(attributes) ? "" : attributes);
+            this.clientCapabilities.put(XServerCapabilities.KEY_SESSION_CONNECT_ATTRS, attMap);
+        }
+
         // Override common SSL properties with xdevapi ones to provide unified logic in ExportControlled via common SSL properties
         RuntimeProperty<XdevapiSslMode> xdevapiSslMode = this.propertySet.<XdevapiSslMode>getEnumProperty(PropertyKey.xdevapiSSLMode);
         if (xdevapiSslMode.isExplicitlySet()) {
@@ -277,9 +304,53 @@ public void beforeHandshake() {
 
         if (xdevapiSslMode.getValue() != XdevapiSslMode.DISABLED) {
             negotiateSSLConnection(0);
+        } else if (this.clientCapabilities.size() > 0) {
+            try {
+                sendCapabilities(this.clientCapabilities);
+            } catch (XProtocolError e) {
+                // XProtocolError: ERROR 5002 (HY000) Capability 'session_connect_attrs' doesn't exist
+                // happens when connecting to xplugin which doesn't support this feature. Just ignore this error.
+                if (e.getErrorCode() != 5002 && !e.getMessage().contains(XServerCapabilities.KEY_SESSION_CONNECT_ATTRS)) {
+                    throw e;
+                }
+                this.clientCapabilities.remove(XServerCapabilities.KEY_SESSION_CONNECT_ATTRS);
+            }
         }
     }
 
+    private Map<String, String> getConnectionAttributesMap(String attStr) {
+        Map<String, String> attMap = new HashMap<>();
+
+        if (attStr != null) {
+            if (attStr.startsWith("[") && attStr.endsWith("]")) {
+                attStr = attStr.substring(1, attStr.length() - 1);
+            }
+            if (!StringUtils.isNullOrEmpty(attStr)) {
+                String[] pairs = attStr.split(",");
+                for (String pair : pairs) {
+                    String[] kv = pair.split("=");
+                    String key = kv[0].trim();
+                    String value = kv.length > 1 ? kv[1].trim() : "";
+                    if (key.startsWith("_")) {
+                        throw ExceptionFactory.createException(WrongArgumentException.class, Messages.getString("Protocol.WrongAttributeName"));
+                    } else if (attMap.put(key, value) != null) {
+                        throw ExceptionFactory.createException(WrongArgumentException.class,
+                                Messages.getString("Protocol.DuplicateAttribute", new Object[] { key }));
+                    }
+                }
+            }
+        }
+
+        attMap.put("_platform", Constants.OS_ARCH);
+        attMap.put("_os", Constants.OS_NAME + "-" + Constants.OS_VERSION);
+        attMap.put("_client_name", Constants.CJ_NAME);
+        attMap.put("_client_version", Constants.CJ_VERSION);
+        attMap.put("_client_license", Constants.CJ_LICENSE);
+        attMap.put("_runtime_version", Constants.JVM_VERSION);
+        attMap.put("_runtime_vendor", Constants.JVM_VENDOR);
+        return attMap;
+    }
+
     private String currUser = null, currPassword = null, currDatabase = null; // TODO remove these variables after implementing mysql_reset_connection() in reset() method
 
     @Override
@@ -616,7 +687,14 @@ public void reset() {
         send(((XMessageBuilder) this.messageBuilder).buildSessionReset(), 0);
         readOk();
 
-        // TODO changeUser() call should be removed after proper implementation of session reset in xplugin 
+        // TODO should be removed after proper implementation of session reset in xplugin
+        if (this.clientCapabilities.containsKey(XServerCapabilities.KEY_SESSION_CONNECT_ATTRS)) {
+            Map<String, Object> reducedClientCapabilities = new HashMap<>();
+            reducedClientCapabilities.put(XServerCapabilities.KEY_SESSION_CONNECT_ATTRS,
+                    this.clientCapabilities.get(XServerCapabilities.KEY_SESSION_CONNECT_ATTRS));
+            sendCapabilities(reducedClientCapabilities);
+        }
+        // TODO should be removed after proper implementation of session reset in xplugin
         this.authProvider.changeUser(null, this.currUser, this.currPassword, this.currDatabase);
     }
 

src/main/protocol-impl/java/com/mysql/cj/protocol/x/XServerCapabilities.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2018, Oracle and/or its affiliates. All rights reserved.
+ * Copyright (c) 2018, 2019, Oracle and/or its affiliates. All rights reserved.
  *
  * This program is free software; you can redistribute it and/or modify it under
  * the terms of the GNU General Public License, version 2.0, as published by the
@@ -42,37 +42,46 @@ public class XServerCapabilities implements ServerCapabilities {
 
     private Map<String, Any> capabilities;
 
+    static String KEY_SESSION_CONNECT_ATTRS = "session_connect_attrs";
+    static String KEY_TLS = "tls";
+    static String KEY_NODE_TYPE = "node_type";
+    static String KEY_CLIENT_PWD_EXPIRE_OK = "client.pwd_expire_ok";
+    static String KEY_AUTHENTICATION_MECHANISMS = "authentication.mechanisms";
+    static String KEY_DOC_FORMATS = "doc.formats";
+
     public XServerCapabilities(Map<String, Any> capabilities) {
         this.capabilities = capabilities;
     }
 
     public void setCapability(String name, Object value) {
-        this.capabilities.put(name, ExprUtil.argObjectToScalarAny(value));
+        if (!KEY_SESSION_CONNECT_ATTRS.equals(name)) {
+            this.capabilities.put(name, ExprUtil.argObjectToScalarAny(value));
+        }
     }
 
     public boolean hasCapability(String name) {
         return this.capabilities.containsKey(name);
     }
 
     public String getNodeType() {
-        return this.capabilities.get("node_type").getScalar().getVString().getValue().toStringUtf8();
+        return this.capabilities.get(KEY_NODE_TYPE).getScalar().getVString().getValue().toStringUtf8();
     }
 
     public boolean getTls() {
-        return hasCapability("tls") ? this.capabilities.get("tls").getScalar().getVBool() : false;
+        return hasCapability(KEY_TLS) ? this.capabilities.get(KEY_TLS).getScalar().getVBool() : false;
     }
 
     public boolean getClientPwdExpireOk() {
-        return this.capabilities.get("client.pwd_expire_ok").getScalar().getVBool();
+        return this.capabilities.get(KEY_CLIENT_PWD_EXPIRE_OK).getScalar().getVBool();
     }
 
     public List<String> getAuthenticationMechanisms() {
-        return this.capabilities.get("authentication.mechanisms").getArray().getValueList().stream()
+        return this.capabilities.get(KEY_AUTHENTICATION_MECHANISMS).getArray().getValueList().stream()
                 .map(v -> v.getScalar().getVString().getValue().toStringUtf8()).collect(Collectors.toList());
     }
 
     public String getDocFormats() {
-        return this.capabilities.get("doc.formats").getScalar().getVString().getValue().toStringUtf8();
+        return this.capabilities.get(KEY_DOC_FORMATS).getScalar().getVString().getValue().toStringUtf8();
     }
 
     @Override

src/main/resources/com/mysql/cj/LocalizedErrorMessages.properties

@@ -346,7 +346,8 @@ Protocol.7=' :\n\n
 Protocol.8=Invalid socket timeout value or state
 Protocol.SlowQuery=Slow query (exceeded {0} {1}, duration: {2} {1}):
 Protocol.ServerSlowQuery=The server processing the query has indicated that the query was marked "slow". 
-
+Protocol.DuplicateAttribute=Duplicate key "{0}" used in "xdevapi.connection-attributes".
+Protocol.WrongAttributeName=Key names in "xdevapi.connection-attributes" cannot start with "_".
 
 RandomBalanceStrategy.0=No hosts configured
 
@@ -908,6 +909,7 @@ ConnectionProperties.asyncResponseTimeout=Timeout (in seconds) for getting serve
 ConnectionProperties.auth=Authentication mechanism to use with the X Protocol. Allowed values are "SHA256_MEMORY", "MYSQL41", "PLAIN", and "EXTERNAL". Value is case insensitive. If the property is not set, the mechanism is chosen depending on the connection type: "PLAIN" is used for TLS connections and "SHA256_MEMORY" or "MYSQL41" is used for unencrypted connections.
 
 ConnectionProperties.xdevapiConnectTimeout=X DevAPI specific timeout for socket connect (in milliseconds), with '0' being no timeout. Defaults to '10000'. If "xdevapi.connect-timeout" is not set explicitly and "connectTimeout" is, "xdevapi.connect-timeout" takes up the value of "connectTimeout". If "xdevapi.useAsyncProtocol=true", both "xdevapi.connect-timeout" and "connectTimeout" are ignored."  
+ConnectionProperties.xdevapiConnectionAttributes=An X DevAPI-specific comma-delimited list of user-defined key=value pairs (in addition to standard X Protocol-defined key=value pairs) to be passed to MySQL Server for display as connection attributes in PERFORMANCE_SCHEMA tables session_account_connect_attrs and session_connect_attrs. Example usage: xdevapi.connection-attributes=key1=value1,key2=value2 or xdevapi.connection-attributes=[key1=value1,key2=value2]. This functionality is available for use with MySQL Server version 8.0.16 or later only. Earlier versions of X Protocol do not support connection attributes, causing this configuration option to be ignored. For situations where Session creation/initialization speed is critical, setting xdevapi.connection-attributes=false will cause connection attribute processing to be bypassed.
 
 ConnectionProperties.unknown=Property is not defined in Connector/J but used in connection URL.