diff --git a/core/src/main/java/org/keycloak/representations/idm/RealmRepresentation.java b/core/src/main/java/org/keycloak/representations/idm/RealmRepresentation.java
index 11ee08ab76..e552c58954 100755
--- a/core/src/main/java/org/keycloak/representations/idm/RealmRepresentation.java
+++ b/core/src/main/java/org/keycloak/representations/idm/RealmRepresentation.java
@@ -54,6 +54,8 @@ public class RealmRepresentation {
     protected Boolean registrationEmailAsUsername;
     protected Boolean rememberMe;
     protected Boolean verifyEmail;
+    protected Boolean loginWithEmailAllowed;
+    protected Boolean duplicateEmailsAllowed;
     protected Boolean resetPasswordAllowed;
     protected Boolean editUsernameAllowed;
 
@@ -418,6 +420,22 @@ public class RealmRepresentation {
     public void setVerifyEmail(Boolean verifyEmail) {
         this.verifyEmail = verifyEmail;
     }
+    
+    public Boolean isLoginWithEmailAllowed() {
+        return loginWithEmailAllowed;
+    }
+
+    public void setLoginWithEmailAllowed(Boolean loginWithEmailAllowed) {
+        this.loginWithEmailAllowed = loginWithEmailAllowed;
+    }
+    
+    public Boolean isDuplicateEmailsAllowed() {
+        return duplicateEmailsAllowed;
+    }
+
+    public void setDuplicateEmailsAllowed(Boolean duplicateEmailsAllowed) {
+        this.duplicateEmailsAllowed = duplicateEmailsAllowed;
+    }
 
     public Boolean isResetPasswordAllowed() {
         return resetPasswordAllowed;
diff --git a/federation/ldap/src/main/java/org/keycloak/storage/ldap/mappers/UserAttributeLDAPStorageMapper.java b/federation/ldap/src/main/java/org/keycloak/storage/ldap/mappers/UserAttributeLDAPStorageMapper.java
index 900969ce30..8a9cf7d306 100644
--- a/federation/ldap/src/main/java/org/keycloak/storage/ldap/mappers/UserAttributeLDAPStorageMapper.java
+++ b/federation/ldap/src/main/java/org/keycloak/storage/ldap/mappers/UserAttributeLDAPStorageMapper.java
@@ -164,7 +164,7 @@ public class UserAttributeLDAPStorageMapper extends AbstractLDAPStorageMapper {
 
     // throw ModelDuplicateException if there is different user in model with same email
     protected void checkDuplicateEmail(String userModelAttrName, String email, RealmModel realm, KeycloakSession session, UserModel user) {
-        if (email == null) return;
+        if (email == null || realm.isDuplicateEmailsAllowed()) return;
         if (UserModel.EMAIL.equalsIgnoreCase(userModelAttrName)) {
             // lowercase before search
             email = KeycloakModelUtils.toLowerCaseSafe(email);
diff --git a/model/infinispan/src/main/java/org/keycloak/models/cache/infinispan/RealmAdapter.java b/model/infinispan/src/main/java/org/keycloak/models/cache/infinispan/RealmAdapter.java
index 2cca4472fa..0b36d671d3 100755
--- a/model/infinispan/src/main/java/org/keycloak/models/cache/infinispan/RealmAdapter.java
+++ b/model/infinispan/src/main/java/org/keycloak/models/cache/infinispan/RealmAdapter.java
@@ -305,6 +305,30 @@ public class RealmAdapter implements CachedRealmModel {
         updated.setVerifyEmail(verifyEmail);
     }
 
+    @Override
+    public boolean isLoginWithEmailAllowed() {
+        if (isUpdated()) return updated.isLoginWithEmailAllowed();
+        return cached.isLoginWithEmailAllowed();
+    }
+
+    @Override
+    public void setLoginWithEmailAllowed(boolean loginWithEmailAllowed) {
+        getDelegateForUpdate();
+        updated.setLoginWithEmailAllowed(loginWithEmailAllowed);
+    }
+    
+    @Override
+    public boolean isDuplicateEmailsAllowed() {
+        if (isUpdated()) return updated.isDuplicateEmailsAllowed();
+        return cached.isDuplicateEmailsAllowed();
+    }
+
+    @Override
+    public void setDuplicateEmailsAllowed(boolean duplicateEmailsAllowed) {
+        getDelegateForUpdate();
+        updated.setDuplicateEmailsAllowed(duplicateEmailsAllowed);
+    }
+
     @Override
     public boolean isResetPasswordAllowed() {
         if (isUpdated()) return updated.isResetPasswordAllowed();
diff --git a/model/infinispan/src/main/java/org/keycloak/models/cache/infinispan/entities/CachedRealm.java b/model/infinispan/src/main/java/org/keycloak/models/cache/infinispan/entities/CachedRealm.java
index 48689e6b7d..5e4b5792b3 100755
--- a/model/infinispan/src/main/java/org/keycloak/models/cache/infinispan/entities/CachedRealm.java
+++ b/model/infinispan/src/main/java/org/keycloak/models/cache/infinispan/entities/CachedRealm.java
@@ -34,9 +34,6 @@ import org.keycloak.models.RealmModel;
 import org.keycloak.models.RequiredActionProviderModel;
 import org.keycloak.models.RequiredCredentialModel;
 
-import java.security.PrivateKey;
-import java.security.PublicKey;
-import java.security.cert.X509Certificate;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.HashMap;
@@ -61,6 +58,8 @@ public class CachedRealm extends AbstractExtendableRevisioned {
     protected boolean registrationEmailAsUsername;
     protected boolean rememberMe;
     protected boolean verifyEmail;
+    protected boolean loginWithEmailAllowed;
+    protected boolean duplicateEmailsAllowed;
     protected boolean resetPasswordAllowed;
     protected boolean identityFederationEnabled;
     protected boolean editUsernameAllowed;
@@ -150,6 +149,8 @@ public class CachedRealm extends AbstractExtendableRevisioned {
         registrationEmailAsUsername = model.isRegistrationEmailAsUsername();
         rememberMe = model.isRememberMe();
         verifyEmail = model.isVerifyEmail();
+        loginWithEmailAllowed = model.isLoginWithEmailAllowed();
+        duplicateEmailsAllowed = model.isDuplicateEmailsAllowed();
         resetPasswordAllowed = model.isResetPasswordAllowed();
         identityFederationEnabled = model.isIdentityFederationEnabled();
         editUsernameAllowed = model.isEditUsernameAllowed();
@@ -340,6 +341,14 @@ public class CachedRealm extends AbstractExtendableRevisioned {
     public boolean isVerifyEmail() {
         return verifyEmail;
     }
+    
+    public boolean isLoginWithEmailAllowed() {
+        return loginWithEmailAllowed;
+    }
+    
+    public boolean isDuplicateEmailsAllowed() {
+        return duplicateEmailsAllowed;
+    }
 
     public boolean isResetPasswordAllowed() {
         return resetPasswordAllowed;
diff --git a/model/jpa/src/main/java/org/keycloak/models/jpa/JpaUserProvider.java b/model/jpa/src/main/java/org/keycloak/models/jpa/JpaUserProvider.java
index 8e1d2eb5a1..dbe367ff49 100755
--- a/model/jpa/src/main/java/org/keycloak/models/jpa/JpaUserProvider.java
+++ b/model/jpa/src/main/java/org/keycloak/models/jpa/JpaUserProvider.java
@@ -480,7 +480,12 @@ public class JpaUserProvider implements UserProvider, UserCredentialStore {
         query.setParameter("email", email.toLowerCase());
         query.setParameter("realmId", realm.getId());
         List<UserEntity> results = query.getResultList();
-        return results.isEmpty() ? null : new UserAdapter(session, realm, em, results.get(0));
+        
+        if (results.isEmpty()) return null;
+        
+        ensureEmailConstraint(results, realm);
+        
+        return new UserAdapter(session, realm, em, results.get(0));
     }
 
      @Override
@@ -880,7 +885,25 @@ public class JpaUserProvider implements UserProvider, UserCredentialStore {
         return toModel(results.get(0));
     }
 
-
-
-
+    // Could override this to provide a custom behavior.
+    protected void ensureEmailConstraint(List<UserEntity> users, RealmModel realm) {
+        UserEntity user = users.get(0);
+        
+        if (users.size() > 1) {
+            // Realm settings have been changed from allowing duplicate emails to not allowing them
+            // but duplicates haven't been removed.
+            throw new ModelDuplicateException("Multiple users with email '" + user.getEmail() + "' exist in Keycloak.");
+        }
+        
+        if (realm.isDuplicateEmailsAllowed()) {
+            return;
+        }
+     
+        if (user.getEmail() != null && !user.getEmail().equals(user.getEmailConstraint())) {
+            // Realm settings have been changed from allowing duplicate emails to not allowing them.
+            // We need to update the email constraint to reflect this change in the user entities.
+            user.setEmailConstraint(user.getEmail());
+            em.persist(user);
+        }  
+    }
 }
diff --git a/model/jpa/src/main/java/org/keycloak/models/jpa/RealmAdapter.java b/model/jpa/src/main/java/org/keycloak/models/jpa/RealmAdapter.java
index f11ba9090d..32f34a5727 100755
--- a/model/jpa/src/main/java/org/keycloak/models/jpa/RealmAdapter.java
+++ b/model/jpa/src/main/java/org/keycloak/models/jpa/RealmAdapter.java
@@ -169,6 +169,7 @@ public class RealmAdapter implements RealmModel, JpaModel<RealmEntity> {
     @Override
     public void setRegistrationEmailAsUsername(boolean registrationEmailAsUsername) {
         realm.setRegistrationEmailAsUsername(registrationEmailAsUsername);
+        if (registrationEmailAsUsername) realm.setDuplicateEmailsAllowed(false);
         em.flush();
     }
 
@@ -347,6 +348,33 @@ public class RealmAdapter implements RealmModel, JpaModel<RealmEntity> {
         realm.setVerifyEmail(verifyEmail);
         em.flush();
     }
+    
+    @Override
+    public boolean isLoginWithEmailAllowed() {
+        return realm.isLoginWithEmailAllowed();
+    }
+
+    @Override
+    public void setLoginWithEmailAllowed(boolean loginWithEmailAllowed) {
+        realm.setLoginWithEmailAllowed(loginWithEmailAllowed);
+        if (loginWithEmailAllowed) realm.setDuplicateEmailsAllowed(false);
+        em.flush();
+    }
+    
+    @Override
+    public boolean isDuplicateEmailsAllowed() {
+        return realm.isDuplicateEmailsAllowed();
+    }
+
+    @Override
+    public void setDuplicateEmailsAllowed(boolean duplicateEmailsAllowed) {
+        realm.setDuplicateEmailsAllowed(duplicateEmailsAllowed);
+        if (duplicateEmailsAllowed) {
+            realm.setLoginWithEmailAllowed(false);
+            realm.setRegistrationEmailAsUsername(false);
+        }
+        em.flush();
+    }
 
     @Override
     public boolean isResetPasswordAllowed() {
diff --git a/model/jpa/src/main/java/org/keycloak/models/jpa/UserAdapter.java b/model/jpa/src/main/java/org/keycloak/models/jpa/UserAdapter.java
index a95548bfe1..a80dec947b 100755
--- a/model/jpa/src/main/java/org/keycloak/models/jpa/UserAdapter.java
+++ b/model/jpa/src/main/java/org/keycloak/models/jpa/UserAdapter.java
@@ -276,7 +276,7 @@ public class UserAdapter implements UserModel, JpaModel<UserEntity> {
     @Override
     public void setEmail(String email) {
         email = KeycloakModelUtils.toLowerCaseSafe(email);
-        user.setEmail(email);
+        user.setEmail(email, realm.isDuplicateEmailsAllowed());
     }
 
     @Override
diff --git a/model/jpa/src/main/java/org/keycloak/models/jpa/entities/RealmEntity.java b/model/jpa/src/main/java/org/keycloak/models/jpa/entities/RealmEntity.java
index eef1d91246..cc62c8a55c 100755
--- a/model/jpa/src/main/java/org/keycloak/models/jpa/entities/RealmEntity.java
+++ b/model/jpa/src/main/java/org/keycloak/models/jpa/entities/RealmEntity.java
@@ -73,6 +73,10 @@ public class RealmEntity {
     protected boolean verifyEmail;
     @Column(name="RESET_PASSWORD_ALLOWED")
     protected boolean resetPasswordAllowed;
+    @Column(name="LOGIN_WITH_EMAIL_ALLOWED")
+    protected boolean loginWithEmailAllowed;
+    @Column(name="DUPLICATE_EMAILS_ALLOWED")
+    protected boolean duplicateEmailsAllowed;
     @Column(name="REMEMBER_ME")
     protected boolean rememberMe;
 
@@ -287,6 +291,22 @@ public class RealmEntity {
     public void setVerifyEmail(boolean verifyEmail) {
         this.verifyEmail = verifyEmail;
     }
+    
+    public boolean isLoginWithEmailAllowed() {
+        return loginWithEmailAllowed;
+    }
+
+    public void setLoginWithEmailAllowed(boolean loginWithEmailAllowed) {
+        this.loginWithEmailAllowed = loginWithEmailAllowed;
+    }
+    
+    public boolean isDuplicateEmailsAllowed() {
+        return duplicateEmailsAllowed;
+    }
+
+    public void setDuplicateEmailsAllowed(boolean duplicateEmailsAllowed) {
+        this.duplicateEmailsAllowed = duplicateEmailsAllowed;
+    }
 
     public boolean isResetPasswordAllowed() {
         return resetPasswordAllowed;
diff --git a/model/jpa/src/main/java/org/keycloak/models/jpa/entities/UserEntity.java b/model/jpa/src/main/java/org/keycloak/models/jpa/entities/UserEntity.java
index 9b5ff1957d..6f2f3ffeba 100755
--- a/model/jpa/src/main/java/org/keycloak/models/jpa/entities/UserEntity.java
+++ b/model/jpa/src/main/java/org/keycloak/models/jpa/entities/UserEntity.java
@@ -78,7 +78,7 @@ public class UserEntity {
     @Column(name = "EMAIL_VERIFIED")
     protected boolean emailVerified;
 
-    // Hack just to workaround the fact that on MS-SQL you can't have unique constraint with multiple NULL values TODO: Find better solution (like unique index with 'where' but that's proprietary)
+    // This is necessary to be able to dynamically switch unique email constraints on and off in the realm settings
     @Column(name = "EMAIL_CONSTRAINT")
     protected String emailConstraint = KeycloakModelUtils.generateId();
 
@@ -144,9 +144,9 @@ public class UserEntity {
         return email;
     }
 
-    public void setEmail(String email) {
+    public void setEmail(String email, boolean allowDuplicate) {
         this.email = email;
-        this.emailConstraint = email != null ? email : KeycloakModelUtils.generateId();
+        this.emailConstraint = email == null || allowDuplicate ? KeycloakModelUtils.generateId() : email;
     }
 
     public boolean isEnabled() {
diff --git a/model/jpa/src/main/resources/META-INF/jpa-changelog-2.5.0.xml b/model/jpa/src/main/resources/META-INF/jpa-changelog-2.5.0.xml
index 4aee29165b..8f561e52f8 100755
--- a/model/jpa/src/main/resources/META-INF/jpa-changelog-2.5.0.xml
+++ b/model/jpa/src/main/resources/META-INF/jpa-changelog-2.5.0.xml
@@ -96,5 +96,16 @@
         <addUniqueConstraint columnNames="NAME,CLIENT_REALM_CONSTRAINT" constraintName="UK_J3RWUVD56ONTGSUHOGM184WW2-2" tableName="KEYCLOAK_ROLE"/>
         <modifyDataType tableName="KEYCLOAK_ROLE" columnName="DESCRIPTION" newDataType="NVARCHAR(255)"/>
     </changeSet>
+    
+    <changeSet author="slawomir@dabek.name" id="2.5.0-duplicate-email-support">
+        <addColumn tableName="REALM">
+            <column name="LOGIN_WITH_EMAIL_ALLOWED" type="BOOLEAN" defaultValueBoolean="true">
+                <constraints nullable="false"/>
+            </column>
+            <column name="DUPLICATE_EMAILS_ALLOWED" type="BOOLEAN" defaultValueBoolean="false">
+                <constraints nullable="false"/>
+            </column>
+        </addColumn>
+    </changeSet>
 
 </databaseChangeLog>
\ No newline at end of file
diff --git a/model/mongo/src/main/java/org/keycloak/connections/mongo/updater/impl/updates/Update2_5_0.java b/model/mongo/src/main/java/org/keycloak/connections/mongo/updater/impl/updates/Update2_5_0.java
index a46947bde4..c95617b4a6 100644
--- a/model/mongo/src/main/java/org/keycloak/connections/mongo/updater/impl/updates/Update2_5_0.java
+++ b/model/mongo/src/main/java/org/keycloak/connections/mongo/updater/impl/updates/Update2_5_0.java
@@ -17,8 +17,10 @@
 
 package org.keycloak.connections.mongo.updater.impl.updates;
 
+import com.mongodb.BasicDBObject;
+import com.mongodb.DBCollection;
+import com.mongodb.DBCursor;
 import org.keycloak.models.KeycloakSession;
-import org.keycloak.models.LDAPConstants;
 import org.keycloak.provider.ProviderFactory;
 import org.keycloak.storage.UserStorageProvider;
 
@@ -40,6 +42,16 @@ public class Update2_5_0 extends AbstractMigrateUserFedToComponent {
         for (ProviderFactory factory : factories) {
             portUserFedToComponent(factory.getId());
         }
+        
+        DBCollection realms = db.getCollection("realms");
+        try (DBCursor realmsCursor = realms.find()) {
+            while (realmsCursor.hasNext()) {
+                BasicDBObject realm = (BasicDBObject) realmsCursor.next();
+                realm.append("loginWithEmailAllowed", true);
+                realm.append("duplicateEmailsAllowed", false);
+                realms.save(realm);
+            }
+        }
     }
 
 }
diff --git a/model/mongo/src/main/java/org/keycloak/models/mongo/keycloak/adapters/MongoUserProvider.java b/model/mongo/src/main/java/org/keycloak/models/mongo/keycloak/adapters/MongoUserProvider.java
index dd4d7a6d82..9100631760 100755
--- a/model/mongo/src/main/java/org/keycloak/models/mongo/keycloak/adapters/MongoUserProvider.java
+++ b/model/mongo/src/main/java/org/keycloak/models/mongo/keycloak/adapters/MongoUserProvider.java
@@ -60,6 +60,7 @@ import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.regex.Pattern;
+import org.keycloak.models.mongo.keycloak.entities.UserEntity;
 
 /**
  * @author <a href="mailto:mposolda@redhat.com">Marek Posolda</a>
@@ -111,13 +112,13 @@ public class MongoUserProvider implements UserProvider, UserCredentialStore {
                 .and("email").is(email.toLowerCase())
                 .and("realmId").is(realm.getId())
                 .get();
-        MongoUserEntity user = getMongoStore().loadSingleEntity(MongoUserEntity.class, query, invocationContext);
+        List<MongoUserEntity> users = getMongoStore().loadEntities(MongoUserEntity.class, query, invocationContext);
 
-        if (user == null) {
-            return null;
-        } else {
-            return new UserAdapter(session, realm, user, invocationContext);
-        }
+        if (users.isEmpty()) return null;
+        
+        ensureEmailConstraint(users, realm);
+        
+        return new UserAdapter(session, realm, users.get(0), invocationContext);
     }
 
     @Override
@@ -817,4 +818,26 @@ public class MongoUserProvider implements UserProvider, UserCredentialStore {
         if (update) getMongoStore().updateEntity(mongoUser, invocationContext);
         return credModel;
     }
+
+    // Could override this to provide a custom behavior.
+    protected void ensureEmailConstraint(List<MongoUserEntity> users, RealmModel realm) {
+        MongoUserEntity user = users.get(0);
+        
+        if (users.size() > 1) {
+            // Realm settings have been changed from allowing duplicate emails to not allowing them
+            // but duplicates haven't been removed.
+            throw new ModelDuplicateException("Multiple users with email '" + user.getEmail() + "' exist in Keycloak.");
+        }
+        
+        if (realm.isDuplicateEmailsAllowed()) {
+            return;
+        }
+     
+        if (user.getEmail() != null && user.getEmailIndex() == null) {
+            // Realm settings have been changed from allowing duplicate emails to not allowing them.
+            // We need to update the email index to reflect this change in the user entities.
+            user.setEmail(user.getEmail(), false);
+            getMongoStore().updateEntity(user, invocationContext);
+        }  
+    }
 }
diff --git a/model/mongo/src/main/java/org/keycloak/models/mongo/keycloak/adapters/RealmAdapter.java b/model/mongo/src/main/java/org/keycloak/models/mongo/keycloak/adapters/RealmAdapter.java
index cf8fd56d26..62d3be8fb3 100755
--- a/model/mongo/src/main/java/org/keycloak/models/mongo/keycloak/adapters/RealmAdapter.java
+++ b/model/mongo/src/main/java/org/keycloak/models/mongo/keycloak/adapters/RealmAdapter.java
@@ -157,12 +157,15 @@ public class RealmAdapter extends AbstractMongoAdapter<MongoRealmEntity> impleme
         updateRealm();
     }
 
+    @Override
     public boolean isRegistrationEmailAsUsername() {
         return realm.isRegistrationEmailAsUsername();
     }
 
+    @Override
     public void setRegistrationEmailAsUsername(boolean registrationEmailAsUsername) {
         realm.setRegistrationEmailAsUsername(registrationEmailAsUsername);
+        if (registrationEmailAsUsername) realm.setDuplicateEmailsAllowed(false);
         updateRealm();
     }
 
@@ -266,6 +269,33 @@ public class RealmAdapter extends AbstractMongoAdapter<MongoRealmEntity> impleme
         realm.setVerifyEmail(verifyEmail);
         updateRealm();
     }
+    
+    @Override
+    public boolean isLoginWithEmailAllowed() {
+        return realm.isLoginWithEmailAllowed();
+    }
+
+    @Override
+    public void setLoginWithEmailAllowed(boolean loginWithEmailAllowed) {
+        realm.setLoginWithEmailAllowed(loginWithEmailAllowed);
+        if (loginWithEmailAllowed) realm.setDuplicateEmailsAllowed(false);
+        updateRealm();
+    }
+
+    @Override
+    public boolean isDuplicateEmailsAllowed() {
+        return realm.isDuplicateEmailsAllowed();
+    }
+
+    @Override
+    public void setDuplicateEmailsAllowed(boolean duplicateEmailsAllowed) {
+        realm.setDuplicateEmailsAllowed(duplicateEmailsAllowed);
+        if (duplicateEmailsAllowed) {
+            realm.setLoginWithEmailAllowed(false);
+            realm.setRegistrationEmailAsUsername(false);
+        }
+        updateRealm();
+    }
 
     @Override
     public boolean isResetPasswordAllowed() {
diff --git a/model/mongo/src/main/java/org/keycloak/models/mongo/keycloak/adapters/UserAdapter.java b/model/mongo/src/main/java/org/keycloak/models/mongo/keycloak/adapters/UserAdapter.java
index e5440cc4cd..9282df017b 100755
--- a/model/mongo/src/main/java/org/keycloak/models/mongo/keycloak/adapters/UserAdapter.java
+++ b/model/mongo/src/main/java/org/keycloak/models/mongo/keycloak/adapters/UserAdapter.java
@@ -124,8 +124,7 @@ public class UserAdapter extends AbstractMongoAdapter<MongoUserEntity> implement
     @Override
     public void setEmail(String email) {
         email = KeycloakModelUtils.toLowerCaseSafe(email);
-
-        user.setEmail(email);
+        user.setEmail(email, realm.isDuplicateEmailsAllowed());
         updateUser();
     }
 
diff --git a/model/mongo/src/main/java/org/keycloak/models/mongo/keycloak/entities/MongoUserEntity.java b/model/mongo/src/main/java/org/keycloak/models/mongo/keycloak/entities/MongoUserEntity.java
index ae9d5a6689..909391b811 100755
--- a/model/mongo/src/main/java/org/keycloak/models/mongo/keycloak/entities/MongoUserEntity.java
+++ b/model/mongo/src/main/java/org/keycloak/models/mongo/keycloak/entities/MongoUserEntity.java
@@ -29,13 +29,6 @@ import org.keycloak.connections.mongo.api.context.MongoStoreInvocationContext;
 @MongoCollection(collectionName = "users")
 public class MongoUserEntity extends UserEntity implements MongoIdentifiableEntity {
 
-    public String getEmailIndex() {
-        return getEmail() != null ? getRealmId() + "//" + getEmail() : null;
-    }
-
-    public void setEmailIndex(String ignored) {
-    }
-
     @Override
     public void afterRemove(MongoStoreInvocationContext context) {
         // Remove all consents of this user
diff --git a/model/mongo/src/main/java/org/keycloak/models/mongo/keycloak/entities/RealmEntity.java b/model/mongo/src/main/java/org/keycloak/models/mongo/keycloak/entities/RealmEntity.java
index 71512167ac..07df0a833a 100755
--- a/model/mongo/src/main/java/org/keycloak/models/mongo/keycloak/entities/RealmEntity.java
+++ b/model/mongo/src/main/java/org/keycloak/models/mongo/keycloak/entities/RealmEntity.java
@@ -37,6 +37,8 @@ public class RealmEntity extends AbstractIdentifiableEntity {
     protected boolean registrationEmailAsUsername;
     private boolean rememberMe;
     private boolean verifyEmail;
+    private boolean loginWithEmailAllowed;
+    private boolean duplicateEmailsAllowed;
     private boolean resetPasswordAllowed;
     private String passwordPolicy;
 
@@ -186,6 +188,22 @@ public class RealmEntity extends AbstractIdentifiableEntity {
     public void setVerifyEmail(boolean verifyEmail) {
         this.verifyEmail = verifyEmail;
     }
+    
+    public boolean isLoginWithEmailAllowed() {
+        return loginWithEmailAllowed;
+    }
+
+    public void setLoginWithEmailAllowed(boolean loginWithEmailAllowed) {
+        this.loginWithEmailAllowed = loginWithEmailAllowed;
+    }
+    
+    public boolean isDuplicateEmailsAllowed() {
+        return duplicateEmailsAllowed;
+    }
+
+    public void setDuplicateEmailsAllowed(boolean duplicateEmailsAllowed) {
+        this.duplicateEmailsAllowed = duplicateEmailsAllowed;
+    }
 
     public boolean isResetPasswordAllowed() {
         return resetPasswordAllowed;
diff --git a/model/mongo/src/main/java/org/keycloak/models/mongo/keycloak/entities/UserEntity.java b/model/mongo/src/main/java/org/keycloak/models/mongo/keycloak/entities/UserEntity.java
index 56e8a88cb3..2dd5395338 100755
--- a/model/mongo/src/main/java/org/keycloak/models/mongo/keycloak/entities/UserEntity.java
+++ b/model/mongo/src/main/java/org/keycloak/models/mongo/keycloak/entities/UserEntity.java
@@ -31,6 +31,7 @@ public class UserEntity extends AbstractIdentifiableEntity {
     private String firstName;
     private String lastName;
     private String email;
+    private String emailIndex;
     private boolean emailVerified;
     private boolean enabled;
 
@@ -82,11 +83,25 @@ public class UserEntity extends AbstractIdentifiableEntity {
     public String getEmail() {
         return email;
     }
-
+    
+    @Deprecated // called upon deserialization only
     public void setEmail(String email) {
         this.email = email;
     }
 
+    public void setEmail(String email, boolean allowDuplicate) {
+        this.email = email;
+        this.emailIndex = email == null || allowDuplicate ? null : getRealmId() + "//" + email;
+    }
+    
+    public void setEmailIndex(String index) {
+        this.emailIndex = index;
+    }
+    
+    public String getEmailIndex() {
+        return emailIndex;
+    }
+
     public boolean isEmailVerified() {
         return emailVerified;
     }
diff --git a/server-spi-private/src/main/java/org/keycloak/models/utils/KeycloakModelUtils.java b/server-spi-private/src/main/java/org/keycloak/models/utils/KeycloakModelUtils.java
index cba151e25a..a258cd77a6 100755
--- a/server-spi-private/src/main/java/org/keycloak/models/utils/KeycloakModelUtils.java
+++ b/server-spi-private/src/main/java/org/keycloak/models/utils/KeycloakModelUtils.java
@@ -188,14 +188,14 @@ public final class KeycloakModelUtils {
     }
 
     /**
-     * Try to find user by username or email
+     * Try to find user by username or email for authentication
      *
      * @param realm    realm
      * @param username username or email of user
      * @return found user
      */
     public static UserModel findUserByNameOrEmail(KeycloakSession session, RealmModel realm, String username) {
-        if (username.indexOf('@') != -1) {
+        if (realm.isLoginWithEmailAllowed() && username.indexOf('@') != -1) {
             UserModel user = session.users().getUserByEmail(username, realm);
             if (user != null) {
                 return user;
diff --git a/server-spi-private/src/main/java/org/keycloak/models/utils/ModelToRepresentation.java b/server-spi-private/src/main/java/org/keycloak/models/utils/ModelToRepresentation.java
index 27dd6dc64f..0a323e0cbd 100755
--- a/server-spi-private/src/main/java/org/keycloak/models/utils/ModelToRepresentation.java
+++ b/server-spi-private/src/main/java/org/keycloak/models/utils/ModelToRepresentation.java
@@ -292,6 +292,8 @@ public class ModelToRepresentation {
         rep.setAdminEventsDetailsEnabled(realm.isAdminEventsDetailsEnabled());
 
         rep.setVerifyEmail(realm.isVerifyEmail());
+        rep.setLoginWithEmailAllowed(realm.isLoginWithEmailAllowed());
+        rep.setDuplicateEmailsAllowed(realm.isDuplicateEmailsAllowed());
         rep.setResetPasswordAllowed(realm.isResetPasswordAllowed());
         rep.setEditUsernameAllowed(realm.isEditUsernameAllowed());
         rep.setRevokeRefreshToken(realm.isRevokeRefreshToken());
diff --git a/server-spi-private/src/main/java/org/keycloak/models/utils/RepresentationToModel.java b/server-spi-private/src/main/java/org/keycloak/models/utils/RepresentationToModel.java
index d8b934fef9..96ffe8cb96 100755
--- a/server-spi-private/src/main/java/org/keycloak/models/utils/RepresentationToModel.java
+++ b/server-spi-private/src/main/java/org/keycloak/models/utils/RepresentationToModel.java
@@ -184,6 +184,8 @@ public class RepresentationToModel {
             newRealm.setRegistrationEmailAsUsername(rep.isRegistrationEmailAsUsername());
         if (rep.isRememberMe() != null) newRealm.setRememberMe(rep.isRememberMe());
         if (rep.isVerifyEmail() != null) newRealm.setVerifyEmail(rep.isVerifyEmail());
+        if (rep.isLoginWithEmailAllowed() != null) newRealm.setLoginWithEmailAllowed(rep.isLoginWithEmailAllowed());
+        if (rep.isDuplicateEmailsAllowed() != null) newRealm.setDuplicateEmailsAllowed(rep.isDuplicateEmailsAllowed());
         if (rep.isResetPasswordAllowed() != null) newRealm.setResetPasswordAllowed(rep.isResetPasswordAllowed());
         if (rep.isEditUsernameAllowed() != null) newRealm.setEditUsernameAllowed(rep.isEditUsernameAllowed());
         if (rep.getLoginTheme() != null) newRealm.setLoginTheme(rep.getLoginTheme());
@@ -785,6 +787,8 @@ public class RepresentationToModel {
         if (rep.isRegistrationEmailAsUsername() != null) realm.setRegistrationEmailAsUsername(rep.isRegistrationEmailAsUsername());
         if (rep.isRememberMe() != null) realm.setRememberMe(rep.isRememberMe());
         if (rep.isVerifyEmail() != null) realm.setVerifyEmail(rep.isVerifyEmail());
+        if (rep.isLoginWithEmailAllowed() != null) realm.setLoginWithEmailAllowed(rep.isLoginWithEmailAllowed());
+        if (rep.isDuplicateEmailsAllowed() != null) realm.setDuplicateEmailsAllowed(rep.isDuplicateEmailsAllowed());
         if (rep.isResetPasswordAllowed() != null) realm.setResetPasswordAllowed(rep.isResetPasswordAllowed());
         if (rep.isEditUsernameAllowed() != null) realm.setEditUsernameAllowed(rep.isEditUsernameAllowed());
         if (rep.getSslRequired() != null) realm.setSslRequired(SslRequired.valueOf(rep.getSslRequired().toUpperCase()));
diff --git a/server-spi/src/main/java/org/keycloak/models/RealmModel.java b/server-spi/src/main/java/org/keycloak/models/RealmModel.java
index 3149f50308..7640aad83c 100755
--- a/server-spi/src/main/java/org/keycloak/models/RealmModel.java
+++ b/server-spi/src/main/java/org/keycloak/models/RealmModel.java
@@ -23,10 +23,6 @@ import org.keycloak.provider.ProviderEvent;
 import org.keycloak.storage.UserStorageProvider;
 import org.keycloak.storage.UserStorageProviderModel;
 
-import java.security.Key;
-import java.security.PrivateKey;
-import java.security.PublicKey;
-import java.security.cert.X509Certificate;
 import java.util.Collections;
 import java.util.LinkedList;
 import java.util.List;
@@ -149,6 +145,14 @@ public interface RealmModel extends RoleContainerModel {
     boolean isVerifyEmail();
 
     void setVerifyEmail(boolean verifyEmail);
+    
+    boolean isLoginWithEmailAllowed();
+
+    void setLoginWithEmailAllowed(boolean loginWithEmailAllowed);
+    
+    boolean isDuplicateEmailsAllowed();
+
+    void setDuplicateEmailsAllowed(boolean duplicateEmailsAllowed);
 
     boolean isResetPasswordAllowed();
 
diff --git a/services/src/main/java/org/keycloak/authentication/authenticators/broker/IdpCreateUserIfUniqueAuthenticator.java b/services/src/main/java/org/keycloak/authentication/authenticators/broker/IdpCreateUserIfUniqueAuthenticator.java
index 9c35844f1c..317cb64873 100644
--- a/services/src/main/java/org/keycloak/authentication/authenticators/broker/IdpCreateUserIfUniqueAuthenticator.java
+++ b/services/src/main/java/org/keycloak/authentication/authenticators/broker/IdpCreateUserIfUniqueAuthenticator.java
@@ -119,7 +119,7 @@ public class IdpCreateUserIfUniqueAuthenticator extends AbstractIdpAuthenticator
     // Could be overriden to detect duplication based on other criterias (firstName, lastName, ...)
     protected ExistingUserInfo checkExistingUser(AuthenticationFlowContext context, String username, SerializedBrokeredIdentityContext serializedCtx, BrokeredIdentityContext brokerContext) {
 
-        if (brokerContext.getEmail() != null) {
+        if (brokerContext.getEmail() != null && !context.getRealm().isDuplicateEmailsAllowed()) {
             UserModel existingUser = context.getSession().users().getUserByEmail(brokerContext.getEmail(), context.getRealm());
             if (existingUser != null) {
                 return new ExistingUserInfo(existingUser.getId(), UserModel.EMAIL, existingUser.getEmail());
diff --git a/services/src/main/java/org/keycloak/authentication/authenticators/resetcred/ResetCredentialChooseUser.java b/services/src/main/java/org/keycloak/authentication/authenticators/resetcred/ResetCredentialChooseUser.java
index d0919cc9e8..19edea653a 100755
--- a/services/src/main/java/org/keycloak/authentication/authenticators/resetcred/ResetCredentialChooseUser.java
+++ b/services/src/main/java/org/keycloak/authentication/authenticators/resetcred/ResetCredentialChooseUser.java
@@ -80,10 +80,11 @@ public class ResetCredentialChooseUser implements Authenticator, AuthenticatorFa
             context.failureChallenge(AuthenticationFlowError.INVALID_USER, challenge);
             return;
         }
-
-        UserModel user = context.getSession().users().getUserByUsername(username, context.getRealm());
-        if (user == null && username.contains("@")) {
-            user =  context.getSession().users().getUserByEmail(username, context.getRealm());
+        
+        RealmModel realm = context.getRealm();
+        UserModel user = context.getSession().users().getUserByUsername(username, realm);
+        if (user == null && realm.isLoginWithEmailAllowed() && username.contains("@")) {
+            user =  context.getSession().users().getUserByEmail(username, realm);
         }
 
         context.getClientSession().setNote(AbstractUsernameFormAuthenticator.ATTEMPTED_USERNAME, username);
diff --git a/services/src/main/java/org/keycloak/authentication/forms/RegistrationProfile.java b/services/src/main/java/org/keycloak/authentication/forms/RegistrationProfile.java
index 355bbe2ad8..08319a39f4 100755
--- a/services/src/main/java/org/keycloak/authentication/forms/RegistrationProfile.java
+++ b/services/src/main/java/org/keycloak/authentication/forms/RegistrationProfile.java
@@ -83,7 +83,7 @@ public class RegistrationProfile implements FormAction, FormActionFactory {
             emailValid = false;
         }
 
-        if (emailValid && context.getSession().users().getUserByEmail(email, context.getRealm()) != null) {
+        if (emailValid && !context.getRealm().isDuplicateEmailsAllowed() && context.getSession().users().getUserByEmail(email, context.getRealm()) != null) {
             eventError = Errors.EMAIL_IN_USE;
             formData.remove(Validation.FIELD_EMAIL);
             context.getEvent().detail(Details.EMAIL, email);
diff --git a/services/src/main/java/org/keycloak/authentication/forms/RegistrationUserCreation.java b/services/src/main/java/org/keycloak/authentication/forms/RegistrationUserCreation.java
index 7b52d32b26..90dee70808 100755
--- a/services/src/main/java/org/keycloak/authentication/forms/RegistrationUserCreation.java
+++ b/services/src/main/java/org/keycloak/authentication/forms/RegistrationUserCreation.java
@@ -86,7 +86,7 @@ public class RegistrationUserCreation implements FormAction, FormActionFactory {
                 context.validationError(formData, errors);
                 return;
             }
-            if (email != null && context.getSession().users().getUserByEmail(email, context.getRealm()) != null) {
+            if (email != null && !context.getRealm().isDuplicateEmailsAllowed() && context.getSession().users().getUserByEmail(email, context.getRealm()) != null) {
                 context.error(Errors.EMAIL_IN_USE);
                 formData.remove(Validation.FIELD_EMAIL);
                 errors.add(new FormMessage(RegistrationPage.FIELD_EMAIL, Messages.EMAIL_EXISTS));
diff --git a/services/src/main/java/org/keycloak/authentication/requiredactions/UpdateProfile.java b/services/src/main/java/org/keycloak/authentication/requiredactions/UpdateProfile.java
index fcc5df7f43..ca0185e1c5 100644
--- a/services/src/main/java/org/keycloak/authentication/requiredactions/UpdateProfile.java
+++ b/services/src/main/java/org/keycloak/authentication/requiredactions/UpdateProfile.java
@@ -104,16 +104,18 @@ public class UpdateProfile implements RequiredActionProvider, RequiredActionFact
         boolean emailChanged = oldEmail != null ? !oldEmail.equals(email) : email != null;
 
         if (emailChanged) {
-            UserModel userByEmail = session.users().getUserByEmail(email, realm);
+            if (!realm.isDuplicateEmailsAllowed()) {
+                UserModel userByEmail = session.users().getUserByEmail(email, realm);
 
-            // check for duplicated email
-            if (userByEmail != null && !userByEmail.getId().equals(user.getId())) {
-                Response challenge = context.form()
-                        .setError(Messages.EMAIL_EXISTS)
-                        .setFormData(formData)
-                        .createResponse(UserModel.RequiredAction.UPDATE_PROFILE);
-                context.challenge(challenge);
-                return;
+                // check for duplicated email
+                if (userByEmail != null && !userByEmail.getId().equals(user.getId())) {
+                    Response challenge = context.form()
+                            .setError(Messages.EMAIL_EXISTS)
+                            .setFormData(formData)
+                            .createResponse(UserModel.RequiredAction.UPDATE_PROFILE);
+                    context.challenge(challenge);
+                    return;
+                }
             }
 
             user.setEmail(email);
diff --git a/services/src/main/java/org/keycloak/forms/login/freemarker/model/RealmBean.java b/services/src/main/java/org/keycloak/forms/login/freemarker/model/RealmBean.java
index 4219f97419..cd6156861e 100755
--- a/services/src/main/java/org/keycloak/forms/login/freemarker/model/RealmBean.java
+++ b/services/src/main/java/org/keycloak/forms/login/freemarker/model/RealmBean.java
@@ -64,6 +64,10 @@ public class RealmBean {
     public boolean isRegistrationEmailAsUsername() {
         return realm.isRegistrationEmailAsUsername();
     }
+    
+    public boolean isLoginWithEmailAllowed() {
+        return realm.isLoginWithEmailAllowed();
+    }
 
     public boolean isResetPasswordAllowed() {
         return realm.isResetPasswordAllowed();
diff --git a/services/src/main/java/org/keycloak/partialimport/AbstractPartialImport.java b/services/src/main/java/org/keycloak/partialimport/AbstractPartialImport.java
index ec42600f2a..a811c7eeaa 100644
--- a/services/src/main/java/org/keycloak/partialimport/AbstractPartialImport.java
+++ b/services/src/main/java/org/keycloak/partialimport/AbstractPartialImport.java
@@ -42,7 +42,7 @@ public abstract class AbstractPartialImport<T> implements PartialImport<T> {
     public abstract String getName(T resourceRep);
     public abstract String getModelId(RealmModel realm, KeycloakSession session, T resourceRep);
     public abstract boolean exists(RealmModel realm, KeycloakSession session, T resourceRep);
-    public abstract String existsMessage(T resourceRep);
+    public abstract String existsMessage(RealmModel realm, T resourceRep);
     public abstract ResourceType getResourceType();
     public abstract void remove(RealmModel realm, KeycloakSession session, T resourceRep);
     public abstract void create(RealmModel realm, KeycloakSession session, T resourceRep);
@@ -59,7 +59,7 @@ public abstract class AbstractPartialImport<T> implements PartialImport<T> {
                 switch (partialImportRep.getPolicy()) {
                     case SKIP: toSkip.add(resourceRep); break;
                     case OVERWRITE: toOverwrite.add(resourceRep); break;
-                    default: throw existsError(existsMessage(resourceRep));
+                    default: throw existsError(existsMessage(realm, resourceRep));
                 }
             }
         }
diff --git a/services/src/main/java/org/keycloak/partialimport/ClientsPartialImport.java b/services/src/main/java/org/keycloak/partialimport/ClientsPartialImport.java
index b7e46dfbc1..308d634cc3 100755
--- a/services/src/main/java/org/keycloak/partialimport/ClientsPartialImport.java
+++ b/services/src/main/java/org/keycloak/partialimport/ClientsPartialImport.java
@@ -56,7 +56,7 @@ public class ClientsPartialImport extends AbstractPartialImport<ClientRepresenta
     }
 
     @Override
-    public String existsMessage(ClientRepresentation clientRep) {
+    public String existsMessage(RealmModel realm, ClientRepresentation clientRep) {
         return "Client id '" + getName(clientRep) + "' already exists";
     }
 
diff --git a/services/src/main/java/org/keycloak/partialimport/GroupsPartialImport.java b/services/src/main/java/org/keycloak/partialimport/GroupsPartialImport.java
index cdb57ad7bc..f0f405d71e 100644
--- a/services/src/main/java/org/keycloak/partialimport/GroupsPartialImport.java
+++ b/services/src/main/java/org/keycloak/partialimport/GroupsPartialImport.java
@@ -59,7 +59,7 @@ public class GroupsPartialImport extends AbstractPartialImport<GroupRepresentati
     }
 
     @Override
-    public String existsMessage(GroupRepresentation groupRep) {
+    public String existsMessage(RealmModel realm, GroupRepresentation groupRep) {
         return "Group '" + groupRep.getPath() + "' already exists";
     }
 
diff --git a/services/src/main/java/org/keycloak/partialimport/IdentityProvidersPartialImport.java b/services/src/main/java/org/keycloak/partialimport/IdentityProvidersPartialImport.java
index cdcb89c031..b68d116117 100644
--- a/services/src/main/java/org/keycloak/partialimport/IdentityProvidersPartialImport.java
+++ b/services/src/main/java/org/keycloak/partialimport/IdentityProvidersPartialImport.java
@@ -55,7 +55,7 @@ public class IdentityProvidersPartialImport extends AbstractPartialImport<Identi
     }
 
     @Override
-    public String existsMessage(IdentityProviderRepresentation idpRep) {
+    public String existsMessage(RealmModel realm, IdentityProviderRepresentation idpRep) {
         return "Identity Provider '" + getName(idpRep) + "' already exists.";
     }
 
diff --git a/services/src/main/java/org/keycloak/partialimport/PartialImportManager.java b/services/src/main/java/org/keycloak/partialimport/PartialImportManager.java
index 5ec3eec43e..107ea43b22 100644
--- a/services/src/main/java/org/keycloak/partialimport/PartialImportManager.java
+++ b/services/src/main/java/org/keycloak/partialimport/PartialImportManager.java
@@ -19,8 +19,10 @@ package org.keycloak.partialimport;
 
 import org.keycloak.events.admin.OperationType;
 import org.keycloak.models.KeycloakSession;
+import org.keycloak.models.ModelDuplicateException;
 import org.keycloak.models.RealmModel;
 import org.keycloak.representations.idm.PartialImportRepresentation;
+import org.keycloak.services.ErrorResponse;
 import org.keycloak.services.resources.admin.AdminEventBuilder;
 
 import javax.ws.rs.core.Response;
@@ -86,7 +88,11 @@ public class PartialImportManager {
         }
 
         if (session.getTransactionManager().isActive()) {
-            session.getTransactionManager().commit();
+            try {
+                session.getTransactionManager().commit();
+            } catch (ModelDuplicateException e) {
+                return ErrorResponse.exists(e.getLocalizedMessage());
+            }
         }
 
         return Response.ok(results).build();
diff --git a/services/src/main/java/org/keycloak/partialimport/RealmRolesPartialImport.java b/services/src/main/java/org/keycloak/partialimport/RealmRolesPartialImport.java
index 9c53709052..c820d12ce0 100644
--- a/services/src/main/java/org/keycloak/partialimport/RealmRolesPartialImport.java
+++ b/services/src/main/java/org/keycloak/partialimport/RealmRolesPartialImport.java
@@ -73,7 +73,7 @@ public class RealmRolesPartialImport extends AbstractPartialImport<RoleRepresent
     }
 
     @Override
-    public String existsMessage(RoleRepresentation roleRep) {
+    public String existsMessage(RealmModel realm, RoleRepresentation roleRep) {
         return "Realm role '" + getName(roleRep) + "' already exists.";
     }
 
diff --git a/services/src/main/java/org/keycloak/partialimport/UsersPartialImport.java b/services/src/main/java/org/keycloak/partialimport/UsersPartialImport.java
index cce0fec4da..c03f0bed3c 100755
--- a/services/src/main/java/org/keycloak/partialimport/UsersPartialImport.java
+++ b/services/src/main/java/org/keycloak/partialimport/UsersPartialImport.java
@@ -60,10 +60,12 @@ public class UsersPartialImport extends AbstractPartialImport<UserRepresentation
         String userName = user.getUsername();
         if (userName != null) {
             return session.users().getUserByUsername(userName, realm).getId();
-        } else {
+        } else if (!realm.isDuplicateEmailsAllowed()) {
             String email = user.getEmail();
             return session.users().getUserByEmail(email, realm).getId();
         }
+        
+        return null;
     }
 
     @Override
@@ -76,13 +78,13 @@ public class UsersPartialImport extends AbstractPartialImport<UserRepresentation
     }
 
     private boolean userEmailExists(RealmModel realm, KeycloakSession session, UserRepresentation user) {
-        return (user.getEmail() != null) &&
+        return (user.getEmail() != null) && !realm.isDuplicateEmailsAllowed() &&
                (session.users().getUserByEmail(user.getEmail(), realm) != null);
     }
 
     @Override
-    public String existsMessage(UserRepresentation user) {
-        if (user.getEmail() == null) {
+    public String existsMessage(RealmModel realm, UserRepresentation user) {
+        if (user.getEmail() == null || !realm.isDuplicateEmailsAllowed()) {
             return "User with user name " + getName(user) + " already exists.";
         }
 
@@ -97,12 +99,13 @@ public class UsersPartialImport extends AbstractPartialImport<UserRepresentation
     @Override
     public void remove(RealmModel realm, KeycloakSession session, UserRepresentation user) {
         UserModel userModel = session.users().getUserByUsername(user.getUsername(), realm);
-        if (userModel == null) {
+        if (userModel == null && !realm.isDuplicateEmailsAllowed()) {
             userModel = session.users().getUserByEmail(user.getEmail(), realm);
         }
-
-        boolean success = new UserManager(session).removeUser(realm, userModel);
-        if (!success) throw new RuntimeException("Unable to overwrite user " + getName(user));
+        if (userModel != null) {
+            boolean success = new UserManager(session).removeUser(realm, userModel);
+            if (!success) throw new RuntimeException("Unable to overwrite user " + getName(user));
+        }
     }
 
     @Override
diff --git a/services/src/main/java/org/keycloak/services/managers/RealmManager.java b/services/src/main/java/org/keycloak/services/managers/RealmManager.java
index 93aafd67f1..c6476234fe 100755
--- a/services/src/main/java/org/keycloak/services/managers/RealmManager.java
+++ b/services/src/main/java/org/keycloak/services/managers/RealmManager.java
@@ -220,6 +220,7 @@ public class RealmManager implements RealmImporter {
         realm.setFailureFactor(30);
         realm.setSslRequired(SslRequired.EXTERNAL);
         realm.setOTPPolicy(OTPPolicy.DEFAULT_POLICY);
+        realm.setLoginWithEmailAllowed(true);
 
         realm.setEventsListeners(Collections.singleton("jboss-logging"));
 
diff --git a/services/src/main/java/org/keycloak/services/resources/AccountService.java b/services/src/main/java/org/keycloak/services/resources/AccountService.java
index 75cc5a7134..f610cd09a7 100755
--- a/services/src/main/java/org/keycloak/services/resources/AccountService.java
+++ b/services/src/main/java/org/keycloak/services/resources/AccountService.java
@@ -400,7 +400,7 @@ public class AccountService extends AbstractSecuredLocalService {
             String email = formData.getFirst("email");
             String oldEmail = user.getEmail();
             boolean emailChanged = oldEmail != null ? !oldEmail.equals(email) : email != null;
-            if (emailChanged) {
+            if (emailChanged && !realm.isDuplicateEmailsAllowed()) {
                 UserModel existing = session.users().getUserByEmail(email, realm);
                 if (existing != null && !existing.getId().equals(user.getId())) {
                     throw new ModelDuplicateException(Messages.EMAIL_EXISTS);
@@ -419,9 +419,11 @@ public class AccountService extends AbstractSecuredLocalService {
             }
 
             if (realm.isRegistrationEmailAsUsername()) {
-                UserModel existing = session.users().getUserByEmail(email, realm);
-                if (existing != null && !existing.getId().equals(user.getId())) {
-                    throw new ModelDuplicateException(Messages.USERNAME_EXISTS);
+                if (!realm.isDuplicateEmailsAllowed()) {
+                    UserModel existing = session.users().getUserByEmail(email, realm);
+                    if (existing != null && !existing.getId().equals(user.getId())) {
+                        throw new ModelDuplicateException(Messages.USERNAME_EXISTS);
+                    }
                 }
                 user.setUsername(email);
             }
diff --git a/services/src/main/java/org/keycloak/services/resources/admin/RealmAdminResource.java b/services/src/main/java/org/keycloak/services/resources/admin/RealmAdminResource.java
index 55427f7c53..8a6999849c 100644
--- a/services/src/main/java/org/keycloak/services/resources/admin/RealmAdminResource.java
+++ b/services/src/main/java/org/keycloak/services/resources/admin/RealmAdminResource.java
@@ -302,6 +302,7 @@ public class RealmAdminResource {
                 }
             }
 
+            boolean wasDuplicateEmailsAllowed = realm.isDuplicateEmailsAllowed();
             RepresentationToModel.updateRealm(rep, realm, session);
 
             // Refresh periodic sync tasks for configured federationProviders
@@ -312,6 +313,12 @@ public class RealmAdminResource {
             }
 
             adminEvent.operation(OperationType.UPDATE).representation(StripSecretsUtils.strip(rep)).success();
+            
+            if (rep.isDuplicateEmailsAllowed() != null && rep.isDuplicateEmailsAllowed() != wasDuplicateEmailsAllowed) {
+                UserCache cache = session.getProvider(UserCache.class);
+                if (cache != null) cache.clear();
+            }
+            
             return Response.noContent().build();
         } catch (PatternSyntaxException e) {
             return ErrorResponse.error("Specified regex pattern(s) is invalid.", Response.Status.BAD_REQUEST);
diff --git a/services/src/main/java/org/keycloak/services/resources/admin/UsersResource.java b/services/src/main/java/org/keycloak/services/resources/admin/UsersResource.java
index 98d664166e..49e719b831 100755
--- a/services/src/main/java/org/keycloak/services/resources/admin/UsersResource.java
+++ b/services/src/main/java/org/keycloak/services/resources/admin/UsersResource.java
@@ -208,7 +208,7 @@ public class UsersResource {
         if (session.users().getUserByUsername(rep.getUsername(), realm) != null) {
             return ErrorResponse.exists("User exists with same username");
         }
-        if (rep.getEmail() != null && session.users().getUserByEmail(rep.getEmail(), realm) != null) {
+        if (rep.getEmail() != null && !realm.isDuplicateEmailsAllowed() && session.users().getUserByEmail(rep.getEmail(), realm) != null) {
             return ErrorResponse.exists("User exists with same email");
         }
 
diff --git a/testsuite/integration-arquillian/tests/base/src/test/java/org/keycloak/testsuite/account/AccountTest.java b/testsuite/integration-arquillian/tests/base/src/test/java/org/keycloak/testsuite/account/AccountTest.java
index b9afd9bfce..77019aa144 100755
--- a/testsuite/integration-arquillian/tests/base/src/test/java/org/keycloak/testsuite/account/AccountTest.java
+++ b/testsuite/integration-arquillian/tests/base/src/test/java/org/keycloak/testsuite/account/AccountTest.java
@@ -489,6 +489,12 @@ public class AccountTest extends TestRealmKeycloakTest {
         testRealm().update(testRealm);
     }
 
+    private void setDuplicateEmailsAllowed(boolean allowed) {
+        RealmRepresentation testRealm = testRealm().toRepresentation();
+        testRealm.setDuplicateEmailsAllowed(allowed);
+        testRealm().update(testRealm);
+    }
+
     @Test
     public void changeUsername() {
         // allow to edit the username in realm
@@ -599,7 +605,7 @@ public class AccountTest extends TestRealmKeycloakTest {
 
     // KEYCLOAK-1534
     @Test
-    public void changeEmailToExisting() {
+    public void changeEmailToExistingForbidden() {
         profilePage.open();
         loginPage.login("test-user@localhost", "password");
 
@@ -633,6 +639,24 @@ public class AccountTest extends TestRealmKeycloakTest {
         profilePage.updateProfile("Tom", "Brady", "test-user@localhost");
         events.expectAccount(EventType.UPDATE_PROFILE).assertEvent();
     }
+ 
+    @Test
+    public void changeEmailToExistingAllowed() {
+        setDuplicateEmailsAllowed(true); 
+        
+        profilePage.open();
+        loginPage.login("test-user@localhost", "password");
+
+        events.expectLogin().client("account").detail(Details.REDIRECT_URI, ACCOUNT_REDIRECT).assertEvent();
+
+        Assert.assertEquals("test-user@localhost", profilePage.getUsername());
+        Assert.assertEquals("test-user@localhost", profilePage.getEmail());
+
+        // Change to the email, which some other user has
+        profilePage.updateProfile("New first", "New last", "test-user-no-access@localhost");
+
+        Assert.assertEquals("Your account has been updated.", profilePage.getSuccess());
+    }
 
     @Test
     public void setupTotp() {
diff --git a/testsuite/integration-arquillian/tests/base/src/test/java/org/keycloak/testsuite/admin/partialimport/PartialImportTest.java b/testsuite/integration-arquillian/tests/base/src/test/java/org/keycloak/testsuite/admin/partialimport/PartialImportTest.java
index e970b9119b..e6f73480a9 100644
--- a/testsuite/integration-arquillian/tests/base/src/test/java/org/keycloak/testsuite/admin/partialimport/PartialImportTest.java
+++ b/testsuite/integration-arquillian/tests/base/src/test/java/org/keycloak/testsuite/admin/partialimport/PartialImportTest.java
@@ -87,6 +87,7 @@ public class PartialImportTest extends AbstractAuthTest {
     public void initAdminEvents() {
         RealmRepresentation realmRep = RealmBuilder.edit(testRealmResource().toRepresentation()).testEventListener().build();
         realmId = realmRep.getId();
+        realmRep.setDuplicateEmailsAllowed(false);
         adminClient.realm(realmRep.getRealm()).update(realmRep);
 
         piRep = new PartialImportRepresentation();
@@ -321,6 +322,40 @@ public class PartialImportTest extends AbstractAuthTest {
         }
     }
 
+    @Test
+    public void testAddUsersWithDuplicateEmailsForbidden() {
+        assertAdminEvents.clear();
+
+        setFail();
+        addUsers();
+        
+        UserRepresentation user = createUserRepresentation(USER_PREFIX + 999, USER_PREFIX + 1 + "@foo.com", "foo", "bar", true);
+        piRep.getUsers().add(user);
+
+        Response response = testRealmResource().partialImport(piRep);
+        assertEquals(409, response.getStatus());
+    }
+    
+    @Test
+    public void testAddUsersWithDuplicateEmailsAllowed() {
+        
+        RealmRepresentation realmRep = new RealmRepresentation();
+        realmRep.setDuplicateEmailsAllowed(true);
+        adminClient.realm(realmId).update(realmRep);
+                
+        assertAdminEvents.clear();
+
+        setFail();
+        addUsers();
+        doImport();
+        
+        UserRepresentation user = createUserRepresentation(USER_PREFIX + 999, USER_PREFIX + 1 + "@foo.com", "foo", "bar", true);
+        piRep.setUsers(Arrays.asList(user));
+        
+        PartialImportResults results = doImport();
+        assertEquals(1, results.getAdded());
+    }
+
     @Test
     public void testAddUsersWithTermsAndConditions() {
         assertAdminEvents.clear();
diff --git a/testsuite/integration-arquillian/tests/base/src/test/java/org/keycloak/testsuite/admin/realm/RealmTest.java b/testsuite/integration-arquillian/tests/base/src/test/java/org/keycloak/testsuite/admin/realm/RealmTest.java
index 26cf87b153..d0a8bde8f6 100755
--- a/testsuite/integration-arquillian/tests/base/src/test/java/org/keycloak/testsuite/admin/realm/RealmTest.java
+++ b/testsuite/integration-arquillian/tests/base/src/test/java/org/keycloak/testsuite/admin/realm/RealmTest.java
@@ -412,6 +412,8 @@ public class RealmTest extends AbstractAdminTest {
         if (realm.isRegistrationEmailAsUsername() != null) assertEquals(realm.isRegistrationEmailAsUsername(), storedRealm.isRegistrationEmailAsUsername());
         if (realm.isRememberMe() != null) assertEquals(realm.isRememberMe(), storedRealm.isRememberMe());
         if (realm.isVerifyEmail() != null) assertEquals(realm.isVerifyEmail(), storedRealm.isVerifyEmail());
+        if (realm.isLoginWithEmailAllowed() != null) assertEquals(realm.isLoginWithEmailAllowed(), storedRealm.isLoginWithEmailAllowed());
+        if (realm.isDuplicateEmailsAllowed() != null) assertEquals(realm.isDuplicateEmailsAllowed(), storedRealm.isDuplicateEmailsAllowed());
         if (realm.isResetPasswordAllowed() != null) assertEquals(realm.isResetPasswordAllowed(), storedRealm.isResetPasswordAllowed());
         if (realm.isEditUsernameAllowed() != null) assertEquals(realm.isEditUsernameAllowed(), storedRealm.isEditUsernameAllowed());
         if (realm.getSslRequired() != null) assertEquals(realm.getSslRequired(), storedRealm.getSslRequired());
diff --git a/testsuite/integration-arquillian/tests/base/src/test/java/org/keycloak/testsuite/exportimport/ExportImportTest.java b/testsuite/integration-arquillian/tests/base/src/test/java/org/keycloak/testsuite/exportimport/ExportImportTest.java
index 2abf0447d8..5d5b3438df 100755
--- a/testsuite/integration-arquillian/tests/base/src/test/java/org/keycloak/testsuite/exportimport/ExportImportTest.java
+++ b/testsuite/integration-arquillian/tests/base/src/test/java/org/keycloak/testsuite/exportimport/ExportImportTest.java
@@ -99,9 +99,9 @@ public class ExportImportTest extends AbstractExportImportTest {
 
         testRealmExportImport();
 
-        // There should be 3 files in target directory (1 realm, 3 user)
+        // There should be 3 files in target directory (1 realm, 4 user)
         File[] files = new File(targetDirPath).listFiles();
-        assertEquals(4, files.length);
+        assertEquals(5, files.length);
     }
 
     @Test
diff --git a/testsuite/integration-arquillian/tests/base/src/test/java/org/keycloak/testsuite/forms/RegisterTest.java b/testsuite/integration-arquillian/tests/base/src/test/java/org/keycloak/testsuite/forms/RegisterTest.java
index 707b765290..1a2149d775 100644
--- a/testsuite/integration-arquillian/tests/base/src/test/java/org/keycloak/testsuite/forms/RegisterTest.java
+++ b/testsuite/integration-arquillian/tests/base/src/test/java/org/keycloak/testsuite/forms/RegisterTest.java
@@ -62,12 +62,12 @@ public class RegisterTest extends TestRealmKeycloakTest {
     }
 
     @Test
-    public void registerExistingUser() {
+    public void registerExistingUsernameForbidden() {
         loginPage.open();
         loginPage.clickRegister();
         registerPage.assertCurrent();
 
-        registerPage.register("firstName", "lastName", "registerExistingUser@email", "test-user@localhost", "password", "password");
+        registerPage.register("firstName", "lastName", "registerExistingUser@email", "roleRichUser", "password", "password");
 
         registerPage.assertCurrent();
         assertEquals("Username already exists.", registerPage.getError());
@@ -80,10 +80,57 @@ public class RegisterTest extends TestRealmKeycloakTest {
         assertEquals("", registerPage.getPassword());
         assertEquals("", registerPage.getPasswordConfirm());
 
-        events.expectRegister("test-user@localhost", "registerExistingUser@email")
+        events.expectRegister("roleRichUser", "registerExistingUser@email")
                 .removeDetail(Details.EMAIL)
                 .user((String) null).error("username_in_use").assertEvent();
     }
+ 
+    @Test
+    public void registerExistingEmailForbidden() {
+        loginPage.open();
+        loginPage.clickRegister();
+        registerPage.assertCurrent();
+
+        registerPage.register("firstName", "lastName", "test-user@localhost", "registerExistingUser", "password", "password");
+
+        registerPage.assertCurrent();
+        assertEquals("Email already exists.", registerPage.getError());
+
+        // assert form keeps form fields on error
+        assertEquals("firstName", registerPage.getFirstName());
+        assertEquals("lastName", registerPage.getLastName());
+        assertEquals("", registerPage.getEmail());
+        assertEquals("registerExistingUser", registerPage.getUsername());
+        assertEquals("", registerPage.getPassword());
+        assertEquals("", registerPage.getPasswordConfirm());
+
+        events.expectRegister("registerExistingUser", "registerExistingUser@email")
+                .removeDetail(Details.EMAIL)
+                .user((String) null).error("email_in_use").assertEvent();
+    }
+ 
+    @Test
+    public void registerExistingEmailAllowed() {
+        setDuplicateEmailsAllowed(true);
+                
+        loginPage.open();
+        loginPage.clickRegister();
+        registerPage.assertCurrent();
+
+        registerPage.register("firstName", "lastName", "test-user@localhost", "registerExistingEmailUser", "password", "password");
+
+        assertEquals(RequestType.AUTH_RESPONSE, appPage.getRequestType());
+
+        String userId = events.expectRegister("registerExistingEmailUser", "test-user@localhost").assertEvent().getUserId();
+        events.expectLogin().detail("username", "registerexistingemailuser").user(userId).assertEvent();
+
+        UserRepresentation user = getUser(userId);
+        Assert.assertNotNull(user);
+        assertEquals("registerexistingemailuser", user.getUsername());
+        assertEquals("test-user@localhost", user.getEmail());
+        assertEquals("firstName", user.getFirstName());
+        assertEquals("lastName", user.getLastName());
+    }
 
     @Test
     public void registerUserInvalidPasswordConfirm() {
@@ -397,5 +444,11 @@ public class RegisterTest extends TestRealmKeycloakTest {
         realm.setRegistrationEmailAsUsername(value);
         testRealm().update(realm);
     }
+    
+    private void setDuplicateEmailsAllowed(boolean allowed) {
+        RealmRepresentation testRealm = testRealm().toRepresentation();
+        testRealm.setDuplicateEmailsAllowed(allowed);
+        testRealm().update(testRealm);
+    }
 
 }
diff --git a/testsuite/integration-arquillian/tests/base/src/test/java/org/keycloak/testsuite/oauth/AccessTokenDuplicateEmailsNotCleanedUpTest.java b/testsuite/integration-arquillian/tests/base/src/test/java/org/keycloak/testsuite/oauth/AccessTokenDuplicateEmailsNotCleanedUpTest.java
new file mode 100644
index 0000000000..d49c4b8ae9
--- /dev/null
+++ b/testsuite/integration-arquillian/tests/base/src/test/java/org/keycloak/testsuite/oauth/AccessTokenDuplicateEmailsNotCleanedUpTest.java
@@ -0,0 +1,100 @@
+/*
+ * Copyright 2016 Red Hat, Inc. and/or its affiliates
+ * and other contributors as indicated by the @author tags.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.keycloak.testsuite.oauth;
+
+import java.util.List;
+
+import static org.junit.Assert.assertEquals;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.keycloak.OAuth2Constants;
+import org.keycloak.representations.AccessToken;
+import org.keycloak.representations.idm.RealmRepresentation;
+import org.keycloak.testsuite.AbstractKeycloakTest;
+import org.keycloak.testsuite.AssertEvents;
+import static org.keycloak.testsuite.admin.AbstractAdminTest.loadJson;
+import static org.keycloak.testsuite.admin.ApiUtil.findUserByUsername;
+import org.keycloak.testsuite.util.OAuthClient;
+import org.openqa.selenium.By;
+
+/**
+ * @author <a href="mailto:slawomir@dabek.name">Slawomir Dabek</a>
+ */
+public class AccessTokenDuplicateEmailsNotCleanedUpTest extends AbstractKeycloakTest {
+
+    @Rule
+    public AssertEvents events = new AssertEvents(this);
+
+    @Override
+    public void beforeAbstractKeycloakTest() throws Exception {
+        super.beforeAbstractKeycloakTest();
+    }
+
+    @Before
+    public void clientConfiguration() {
+        oauth.clientId("test-app");
+        oauth.realm("test-duplicate-emails");
+
+        RealmRepresentation realmRep = new RealmRepresentation();
+        // change realm settings to allow login with email after having imported users with duplicate email addresses
+        realmRep.setLoginWithEmailAllowed(true);
+        adminClient.realm("test-duplicate-emails").update(realmRep);
+    }
+
+    @Override
+    public void addTestRealms(List<RealmRepresentation> testRealms) {
+        RealmRepresentation realm = loadJson(getClass().getResourceAsStream("/testrealm-duplicate-emails.json"), RealmRepresentation.class);
+        testRealms.add(realm);
+    }
+
+    @Test
+    public void loginWithNonDuplicateEmail() throws Exception {
+        oauth.doLogin("non-duplicate-email-user@localhost", "password");
+
+        String code = oauth.getCurrentQuery().get(OAuth2Constants.CODE);
+        OAuthClient.AccessTokenResponse response = oauth.doAccessTokenRequest(code, "password");
+
+        assertEquals(200, response.getStatusCode());
+        
+        AccessToken token = oauth.verifyToken(response.getAccessToken());
+        
+        assertEquals(findUserByUsername(adminClient.realm("test-duplicate-emails"), "non-duplicate-email-user").getId(), token.getSubject());
+    }
+
+    @Test
+    public void loginWithDuplicateEmail() throws Exception {
+        oauth.doLogin("duplicate-email-user@localhost", "password");
+
+        assertEquals("Username already exists.", driver.findElement(By.xpath("//span[@class='kc-feedback-text']")).getText());
+    }
+    
+    @Test
+    public void loginWithUserHavingDuplicateEmailByUsername() throws Exception {
+        oauth.doLogin("duplicate-email-user1", "password");
+
+        String code = oauth.getCurrentQuery().get(OAuth2Constants.CODE);
+        OAuthClient.AccessTokenResponse response = oauth.doAccessTokenRequest(code, "password");
+
+        assertEquals(200, response.getStatusCode());
+        
+        AccessToken token = oauth.verifyToken(response.getAccessToken());
+        
+        assertEquals(findUserByUsername(adminClient.realm("test-duplicate-emails"), "duplicate-email-user1").getId(), token.getSubject());
+        assertEquals("duplicate-email-user@localhost", token.getEmail());
+    }
+}
diff --git a/testsuite/integration-arquillian/tests/base/src/test/java/org/keycloak/testsuite/oauth/AccessTokenDuplicateEmailsTest.java b/testsuite/integration-arquillian/tests/base/src/test/java/org/keycloak/testsuite/oauth/AccessTokenDuplicateEmailsTest.java
new file mode 100644
index 0000000000..0f56bcd3f7
--- /dev/null
+++ b/testsuite/integration-arquillian/tests/base/src/test/java/org/keycloak/testsuite/oauth/AccessTokenDuplicateEmailsTest.java
@@ -0,0 +1,128 @@
+/*
+ * Copyright 2016 Red Hat, Inc. and/or its affiliates
+ * and other contributors as indicated by the @author tags.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.keycloak.testsuite.oauth;
+
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.keycloak.OAuth2Constants;
+import org.keycloak.representations.AccessToken;
+import org.keycloak.representations.idm.RealmRepresentation;
+import org.keycloak.testsuite.AbstractKeycloakTest;
+import org.keycloak.testsuite.AssertEvents;
+import org.keycloak.testsuite.util.OAuthClient;
+import java.util.List;
+
+import static org.junit.Assert.assertEquals;
+import static org.keycloak.testsuite.admin.AbstractAdminTest.loadJson;
+import static org.keycloak.testsuite.admin.ApiUtil.findUserByUsername;
+import org.keycloak.testsuite.arquillian.AuthServerTestEnricher;
+import org.openqa.selenium.By;
+
+/**
+ * @author <a href="mailto:slawomir@dabek.name">Slawomir Dabek</a>
+ */
+public class AccessTokenDuplicateEmailsTest extends AbstractKeycloakTest {
+
+    @Rule
+    public AssertEvents events = new AssertEvents(this);
+
+
+    @Override
+    public void beforeAbstractKeycloakTest() throws Exception {
+        super.beforeAbstractKeycloakTest();
+    }
+
+    @Before
+    public void clientConfiguration() {
+        oauth.clientId("test-app");
+        oauth.realm("test-duplicate-emails");
+    }
+
+    @Override
+    public void addTestRealms(List<RealmRepresentation> testRealms) {
+        RealmRepresentation realm = loadJson(getClass().getResourceAsStream("/testrealm-duplicate-emails.json"), RealmRepresentation.class);
+        testRealms.add(realm);
+    }
+    
+    @Test
+    public void loginFormUsernameLabel() throws Exception {
+        oauth.openLoginForm();
+        oauth.redirectUri(AuthServerTestEnricher.getAuthServerContextRoot() + "/does/not/matter/");
+        
+        assertEquals("Username", driver.findElement(By.xpath("//label[@for='username']")).getText());
+    }
+
+    @Test
+    public void loginWithNonDuplicateEmailUser() throws Exception {
+        oauth.doLogin("non-duplicate-email-user", "password");
+
+        String code = oauth.getCurrentQuery().get(OAuth2Constants.CODE);
+        OAuthClient.AccessTokenResponse response = oauth.doAccessTokenRequest(code, "password");
+
+        assertEquals(200, response.getStatusCode());
+        
+        AccessToken token = oauth.verifyToken(response.getAccessToken());
+        
+        assertEquals(findUserByUsername(adminClient.realm("test-duplicate-emails"), "non-duplicate-email-user").getId(), token.getSubject());
+        assertEquals("non-duplicate-email-user@localhost", token.getEmail());
+    }
+
+    @Test
+    public void loginWithFirstDuplicateEmailUser() throws Exception {
+        oauth.doLogin("duplicate-email-user1", "password");
+
+        String code = oauth.getCurrentQuery().get(OAuth2Constants.CODE);
+        OAuthClient.AccessTokenResponse response = oauth.doAccessTokenRequest(code, "password");
+
+        assertEquals(200, response.getStatusCode());
+        
+        AccessToken token = oauth.verifyToken(response.getAccessToken());
+        
+        assertEquals(findUserByUsername(adminClient.realm("test-duplicate-emails"), "duplicate-email-user1").getId(), token.getSubject());
+        assertEquals("duplicate-email-user@localhost", token.getEmail());
+    }
+    
+    @Test
+    public void loginWithSecondDuplicateEmailUser() throws Exception {
+        oauth.doLogin("duplicate-email-user2", "password");
+
+        String code = oauth.getCurrentQuery().get(OAuth2Constants.CODE);
+        OAuthClient.AccessTokenResponse response = oauth.doAccessTokenRequest(code, "password");
+
+        assertEquals(200, response.getStatusCode());
+        
+        AccessToken token = oauth.verifyToken(response.getAccessToken());
+        
+        assertEquals(findUserByUsername(adminClient.realm("test-duplicate-emails"), "duplicate-email-user2").getId(), token.getSubject());
+        assertEquals("duplicate-email-user@localhost", token.getEmail());
+    }
+
+    @Test
+    public void loginWithNonDuplicateEmail() throws Exception {
+        oauth.doLogin("non-duplicate-email-user@localhost", "password");
+
+        assertEquals("Invalid username or password.", driver.findElement(By.xpath("//span[@class='kc-feedback-text']")).getText());
+    }
+    
+    @Test
+    public void loginWithDuplicateEmail() throws Exception {
+        oauth.doLogin("duplicate-email-user@localhost", "password");
+
+        assertEquals("Invalid username or password.", driver.findElement(By.xpath("//span[@class='kc-feedback-text']")).getText());
+    }
+}
diff --git a/testsuite/integration-arquillian/tests/base/src/test/java/org/keycloak/testsuite/oauth/AccessTokenNoEmailLoginTest.java b/testsuite/integration-arquillian/tests/base/src/test/java/org/keycloak/testsuite/oauth/AccessTokenNoEmailLoginTest.java
new file mode 100644
index 0000000000..316c09d57e
--- /dev/null
+++ b/testsuite/integration-arquillian/tests/base/src/test/java/org/keycloak/testsuite/oauth/AccessTokenNoEmailLoginTest.java
@@ -0,0 +1,83 @@
+/*
+ * Copyright 2016 Red Hat, Inc. and/or its affiliates
+ * and other contributors as indicated by the @author tags.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.keycloak.testsuite.oauth;
+
+import java.util.List;
+
+import static org.junit.Assert.assertEquals;
+import org.junit.Before;
+import org.junit.Test;
+import org.keycloak.OAuth2Constants;
+import org.keycloak.representations.AccessToken;
+import org.keycloak.representations.idm.RealmRepresentation;
+import org.keycloak.testsuite.AbstractKeycloakTest;
+import static org.keycloak.testsuite.admin.AbstractAdminTest.loadJson;
+import static org.keycloak.testsuite.admin.ApiUtil.findUserByUsername;
+import org.keycloak.testsuite.util.OAuthClient;
+import org.openqa.selenium.By;
+
+/**
+ * @author <a href="mailto:slawomir@dabek.name">Slawomir Dabek</a>
+ */
+public class AccessTokenNoEmailLoginTest extends AbstractKeycloakTest {
+    
+    @Override
+    public void beforeAbstractKeycloakTest() throws Exception {
+        super.beforeAbstractKeycloakTest();
+    }
+
+    @Before
+    public void clientConfiguration() {
+        oauth.clientId("test-app");
+    }
+
+    @Override
+    public void addTestRealms(List<RealmRepresentation> testRealms) {
+        RealmRepresentation realm = loadJson(getClass().getResourceAsStream("/testrealm.json"), RealmRepresentation.class);
+        realm.setLoginWithEmailAllowed(false);
+        testRealms.add(realm);
+    }
+    
+    @Test
+    public void loginFormUsernameLabel() throws Exception {
+        oauth.openLoginForm();
+        
+        assertEquals("Username", driver.findElement(By.xpath("//label[@for='username']")).getText());
+    }
+
+    @Test
+    public void loginWithUsername() throws Exception {
+        oauth.doLogin("non-duplicate-email-user", "password");
+
+        String code = oauth.getCurrentQuery().get(OAuth2Constants.CODE);
+        OAuthClient.AccessTokenResponse response = oauth.doAccessTokenRequest(code, "password");
+
+        assertEquals(200, response.getStatusCode());
+
+        AccessToken token = oauth.verifyToken(response.getAccessToken());
+
+        assertEquals(findUserByUsername(adminClient.realm("test"), "non-duplicate-email-user").getId(), token.getSubject());
+        assertEquals("non-duplicate-email-user@localhost", token.getEmail());
+    }
+
+    @Test
+    public void loginWithEmail() throws Exception {
+        oauth.doLoginGrant("non-duplicate-email-user@localhost", "password");
+        
+        assertEquals("Invalid username or password.", driver.findElement(By.xpath("//span[@class='kc-feedback-text']")).getText());
+    }
+}
diff --git a/testsuite/integration-arquillian/tests/base/src/test/java/org/keycloak/testsuite/oauth/AccessTokenTest.java b/testsuite/integration-arquillian/tests/base/src/test/java/org/keycloak/testsuite/oauth/AccessTokenTest.java
index 7115f7481e..92e68cb046 100755
--- a/testsuite/integration-arquillian/tests/base/src/test/java/org/keycloak/testsuite/oauth/AccessTokenTest.java
+++ b/testsuite/integration-arquillian/tests/base/src/test/java/org/keycloak/testsuite/oauth/AccessTokenTest.java
@@ -93,6 +93,7 @@ import static org.keycloak.testsuite.admin.ApiUtil.findUserByUsername;
 import static org.keycloak.testsuite.admin.ApiUtil.findUserByUsernameId;
 import static org.keycloak.testsuite.util.OAuthClient.AUTH_SERVER_ROOT;
 import static org.keycloak.testsuite.util.ProtocolMapperUtil.createRoleNameMapper;
+import org.openqa.selenium.By;
 
 /**
  * @author <a href="mailto:sthorger@redhat.com">Stian Thorgersen</a>
@@ -135,6 +136,13 @@ public class AccessTokenTest extends AbstractKeycloakTest {
         testRealms.add(realm);
 
     }
+    
+    @Test
+    public void loginFormUsernameOrEmailLabel() throws Exception {
+        oauth.openLoginForm();
+        
+        assertEquals("Username or email", driver.findElement(By.xpath("//label[@for='username']")).getText());
+    }
 
     @Test
     public void accessTokenRequest() throws Exception {
diff --git a/testsuite/integration-arquillian/tests/base/src/test/resources/testrealm-duplicate-emails.json b/testsuite/integration-arquillian/tests/base/src/test/resources/testrealm-duplicate-emails.json
new file mode 100644
index 0000000000..560c1d297d
--- /dev/null
+++ b/testsuite/integration-arquillian/tests/base/src/test/resources/testrealm-duplicate-emails.json
@@ -0,0 +1,142 @@
+{
+  "id": "test-duplicate-emails",
+  "realm": "test-duplicate-emails",
+  "enabled": true,
+  "sslRequired": "external",
+  "registrationAllowed": true,
+  "resetPasswordAllowed": true,
+  "editUsernameAllowed" : true,
+  "loginWithEmailAllowed": false,
+  "duplicateEmailsAllowed": true,
+  "privateKey": "MIICXAIBAAKBgQCrVrCuTtArbgaZzL1hvh0xtL5mc7o0NqPVnYXkLvgcwiC3BjLGw1tGEGoJaXDuSaRllobm53JBhjx33UNv+5z/UMG4kytBWxheNVKnL6GgqlNabMaFfPLPCF8kAgKnsi79NMo+n6KnSY8YeUmec/p2vjO2NjsSAVcWEQMVhJ31LwIDAQABAoGAfmO8gVhyBxdqlxmIuglbz8bcjQbhXJLR2EoS8ngTXmN1bo2L90M0mUKSdc7qF10LgETBzqL8jYlQIbt+e6TH8fcEpKCjUlyq0Mf/vVbfZSNaVycY13nTzo27iPyWQHK5NLuJzn1xvxxrUeXI6A2WFpGEBLbHjwpx5WQG9A+2scECQQDvdn9NE75HPTVPxBqsEd2z10TKkl9CZxu10Qby3iQQmWLEJ9LNmy3acvKrE3gMiYNWb6xHPKiIqOR1as7L24aTAkEAtyvQOlCvr5kAjVqrEKXalj0Tzewjweuxc0pskvArTI2Oo070h65GpoIKLc9jf+UA69cRtquwP93aZKtW06U8dQJAF2Y44ks/mK5+eyDqik3koCI08qaC8HYq2wVl7G2QkJ6sbAaILtcvD92ToOvyGyeE0flvmDZxMYlvaZnaQ0lcSQJBAKZU6umJi3/xeEbkJqMfeLclD27XGEFoPeNrmdx0q10Azp4NfJAY+Z8KRyQCR2BEG+oNitBOZ+YXF9KCpH3cdmECQHEigJhYg+ykOvr1aiZUMFT72HU0jnmQe2FVekuG+LJUt2Tm7GtMjTFoGpf0JwrVuZN39fOYAlo+nTixgeW7X8Y=",
+  "publicKey": "MIGfMA0GCSqGSIb3DQEBAQUAA4GNADCBiQKBgQCrVrCuTtArbgaZzL1hvh0xtL5mc7o0NqPVnYXkLvgcwiC3BjLGw1tGEGoJaXDuSaRllobm53JBhjx33UNv+5z/UMG4kytBWxheNVKnL6GgqlNabMaFfPLPCF8kAgKnsi79NMo+n6KnSY8YeUmec/p2vjO2NjsSAVcWEQMVhJ31LwIDAQAB",
+  "requiredCredentials": [ "password" ],
+  "defaultRoles": [ "user" ],
+  "smtpServer": {
+    "from": "auto@keycloak.org",
+    "host": "localhost",
+    "port":"3025"
+  },
+  "users" : [
+    {
+      "username" : "non-duplicate-email-user",
+      "enabled": true,
+      "email" : "non-duplicate-email-user@localhost",
+      "firstName": "Brian",
+      "lastName": "Cohen",
+      "credentials" : [
+        { "type" : "password",
+          "value" : "password" }
+      ],
+      "realmRoles": ["user", "offline_access"],
+      "clientRoles": {
+        "test-app": [ "customer-user" ],
+        "account": [ "view-profile", "manage-account" ]
+      }
+    },
+    {
+      "username" : "duplicate-email-user1",
+      "enabled": true,
+      "email" : "duplicate-email-user@localhost",
+      "firstName": "Agent",
+      "lastName": "Smith",
+      "credentials" : [
+        { "type" : "password",
+          "value" : "password" }
+      ],
+      "realmRoles": ["user", "offline_access"],
+      "clientRoles": {
+        "test-app": [ "customer-user" ],
+        "account": [ "view-profile", "manage-account" ]
+      }
+    },
+    {
+      "username" : "duplicate-email-user2",
+      "enabled": true,
+      "email" : "duplicate-email-user@localhost",
+      "firstName": "Agent",
+      "lastName": "Smith",
+      "credentials" : [
+        { "type" : "password",
+          "value" : "password" }
+      ],
+      "realmRoles": ["user", "offline_access"],
+      "clientRoles": {
+        "test-app": [ "customer-user" ],
+        "account": [ "view-profile", "manage-account" ]
+      }
+    }
+  ],
+  "scopeMappings": [
+    {
+      "client": "test-app",
+      "roles": ["user"]
+    }
+  ],
+  "clients": [
+    {
+      "clientId": "test-app",
+      "enabled": true,
+      "baseUrl": "http://localhost:8180/auth/realms/master/app/auth",
+      "redirectUris": [
+        "http://localhost:8180/auth/realms/master/app/auth/*"
+      ],
+      "adminUrl": "http://localhost:8180/auth/realms/master/app/admin",
+      "secret": "password"
+    }
+  ],
+  "roles" : {
+    "realm" : [
+      {
+        "name": "user",
+        "description": "Have User privileges"
+      },
+      {
+        "name": "admin",
+        "description": "Have Administrator privileges"
+      },
+      {
+        "name": "customer-user-premium",
+        "description": "Have User Premium privileges"
+      },
+      {
+        "name": "sample-realm-role",
+        "description": "Sample realm role"
+      }
+    ],
+    "client" : {
+      "test-app" : [
+        {
+          "name": "customer-user",
+          "description": "Have Customer User privileges"
+        },
+        {
+          "name": "customer-admin",
+          "description": "Have Customer Admin privileges"
+        },
+        {
+          "name": "sample-client-role",
+          "description": "Sample client role"
+        },
+        {
+          "name": "customer-admin-composite-role",
+          "description": "Have Customer Admin privileges via composite role",
+          "composite" : true,
+          "composites" : {
+            "realm" : [ "customer-user-premium" ],
+            "client" : {
+              "test-app" : [ "customer-admin" ]
+            }
+          }
+        }
+      ]
+    }
+
+  },
+  "groups" : [],
+  "clientScopeMappings": {},
+  "internationalizationEnabled": true,
+  "supportedLocales": ["en", "de"],
+  "defaultLocale": "en",
+  "eventsListeners": ["jboss-logging", "event-queue"]
+}
diff --git a/testsuite/integration-arquillian/tests/base/src/test/resources/testrealm.json b/testsuite/integration-arquillian/tests/base/src/test/resources/testrealm.json
index b0e87679b3..969d9b53e2 100644
--- a/testsuite/integration-arquillian/tests/base/src/test/resources/testrealm.json
+++ b/testsuite/integration-arquillian/tests/base/src/test/resources/testrealm.json
@@ -100,6 +100,22 @@
       "clientRoles": {
         "test-app-scope": [ "test-app-allowed-by-scope", "test-app-disallowed-by-scope" ]
       }
+    },
+    {
+      "username" : "non-duplicate-email-user",
+      "enabled": true,
+      "email" : "non-duplicate-email-user@localhost",
+      "firstName": "Brian",
+      "lastName": "Cohen",
+      "credentials" : [
+        { "type" : "password",
+          "value" : "password" }
+      ],
+      "realmRoles": ["user", "offline_access"],
+      "clientRoles": {
+        "test-app": [ "customer-user" ],
+        "account": [ "view-profile", "manage-account" ]
+      }
     }
   ],
   "scopeMappings": [
diff --git a/themes/src/main/resources/theme/base/admin/messages/admin-messages_en.properties b/themes/src/main/resources/theme/base/admin/messages/admin-messages_en.properties
index 18ffe2a392..d29632ffb3 100644
--- a/themes/src/main/resources/theme/base/admin/messages/admin-messages_en.properties
+++ b/themes/src/main/resources/theme/base/admin/messages/admin-messages_en.properties
@@ -32,6 +32,10 @@ resetPasswordAllowed=Forgot password
 resetPasswordAllowed.tooltip=Show a link on login page for user to click on when they have forgotten their credentials.
 rememberMe=Remember Me
 rememberMe.tooltip=Show checkbox on login page to allow user to remain logged in between browser restarts until session expires.
+loginWithEmailAllowed=Login with email
+loginWithEmailAllowed.tooltip=Allow users to log in with their email address.
+duplicateEmailsAllowed=Duplicate emails
+duplicateEmailsAllowed.tooltip=Allow multiple users to have the same email address. Changing this setting will also clear the users cache. It is recommended to manually update email constraints of existing users in the database after switching off support for duplicate email addresses.
 verifyEmail=Verify email
 verifyEmail.tooltip=Require the user to verify their email address the first time they login.
 sslRequired=Require SSL
diff --git a/themes/src/main/resources/theme/base/admin/resources/partials/realm-login-settings.html b/themes/src/main/resources/theme/base/admin/resources/partials/realm-login-settings.html
index 9766c7a649..c5a084ac20 100755
--- a/themes/src/main/resources/theme/base/admin/resources/partials/realm-login-settings.html
+++ b/themes/src/main/resources/theme/base/admin/resources/partials/realm-login-settings.html
@@ -45,6 +45,20 @@
                 </div>
                 <kc-tooltip>{{:: 'verifyEmail.tooltip' | translate}}</kc-tooltip>
             </div>
+            <div class="form-group">
+                <label for="loginWithEmailAllowed" class="col-md-2 control-label">{{:: 'loginWithEmailAllowed' | translate}}</label>
+                <div class="col-md-6">
+                    <input ng-model="realm.loginWithEmailAllowed" name="loginWithEmailAllowed" id="loginWithEmailAllowed" onoffswitch on-text="{{:: 'onText' | translate}}" off-text="{{:: 'offText' | translate}}" />
+                </div>
+                <kc-tooltip>{{:: 'loginWithEmailAllowed.tooltip' | translate}}</kc-tooltip>
+            </div>
+            <div class="form-group" ng-show="!realm.loginWithEmailAllowed && !realm.registrationEmailAsUsername">
+                <label for="duplicateEmailsAllowed" class="col-md-2 control-label">{{:: 'duplicateEmailsAllowed' | translate}}</label>
+                <div class="col-md-6">
+                    <input ng-model="realm.duplicateEmailsAllowed" name="duplicateEmailsAllowed" id="duplicateEmailsAllowed" onoffswitch on-text="{{:: 'onText' | translate}}" off-text="{{:: 'offText' | translate}}" />
+                </div>
+                <kc-tooltip>{{:: 'duplicateEmailsAllowed.tooltip' | translate}}</kc-tooltip>
+            </div>
             <div class="form-group">
                 <label for="sslRequired" class="col-md-2 control-label">{{:: 'sslRequired' | translate}}</label>
                 <div class="col-md-2">
diff --git a/themes/src/main/resources/theme/base/login/login-reset-password.ftl b/themes/src/main/resources/theme/base/login/login-reset-password.ftl
index a561b2a65f..a0d118a40d 100755
--- a/themes/src/main/resources/theme/base/login/login-reset-password.ftl
+++ b/themes/src/main/resources/theme/base/login/login-reset-password.ftl
@@ -8,7 +8,7 @@
         <form id="kc-reset-password-form" class="${properties.kcFormClass!}" action="${url.loginAction}" method="post">
             <div class="${properties.kcFormGroupClass!}">
                 <div class="${properties.kcLabelWrapperClass!}">
-                    <label for="username" class="${properties.kcLabelClass!}"><#if !realm.registrationEmailAsUsername>${msg("usernameOrEmail")}<#else>${msg("email")}</#if></label>
+                    <label for="username" class="${properties.kcLabelClass!}"><#if !realm.loginWithEmailAllowed>${msg("username")}<#elseif !realm.registrationEmailAsUsername>${msg("usernameOrEmail")}<#else>${msg("email")}</#if></label>
                 </div>
                 <div class="${properties.kcInputWrapperClass!}">
                     <input type="text" id="username" name="username" class="${properties.kcInputClass!}" autofocus/>
diff --git a/themes/src/main/resources/theme/base/login/login.ftl b/themes/src/main/resources/theme/base/login/login.ftl
index bcaa952646..7c0212346a 100755
--- a/themes/src/main/resources/theme/base/login/login.ftl
+++ b/themes/src/main/resources/theme/base/login/login.ftl
@@ -9,7 +9,7 @@
             <form id="kc-form-login" class="${properties.kcFormClass!}" action="${url.loginAction}" method="post">
                 <div class="${properties.kcFormGroupClass!}">
                     <div class="${properties.kcLabelWrapperClass!}">
-                        <label for="username" class="${properties.kcLabelClass!}"><#if !realm.registrationEmailAsUsername>${msg("usernameOrEmail")}<#else>${msg("email")}</#if></label>
+                        <label for="username" class="${properties.kcLabelClass!}"><#if !realm.loginWithEmailAllowed>${msg("username")}<#elseif !realm.registrationEmailAsUsername>${msg("usernameOrEmail")}<#else>${msg("email")}</#if></label>
                     </div>
 
                     <div class="${properties.kcInputWrapperClass!}">