diff --git a/hazelcast-client/src/main/java/com/hazelcast/client/impl/DefaultClientExtension.java b/hazelcast-client/src/main/java/com/hazelcast/client/impl/DefaultClientExtension.java
index fd0cf0e6b9fb..816bc7af87c2 100644
--- a/hazelcast-client/src/main/java/com/hazelcast/client/impl/DefaultClientExtension.java
+++ b/hazelcast-client/src/main/java/com/hazelcast/client/impl/DefaultClientExtension.java
@@ -46,6 +46,7 @@
 import com.hazelcast.nio.SocketInterceptor;
 import com.hazelcast.partition.strategy.DefaultPartitioningStrategy;
 import com.hazelcast.spi.properties.GroupProperty;
+import com.hazelcast.spi.properties.HazelcastProperties;
 import com.hazelcast.spi.serialization.SerializationService;
 import com.hazelcast.util.function.Supplier;
 
@@ -89,6 +90,7 @@ public InternalSerializationService createSerializationService(byte version) {
             return builder
                     .setClassLoader(configClassLoader)
                     .setConfig(serializationConfig)
+                    .setProperties(new HazelcastProperties(config.getProperties()))
                     .setManagedContext(new HazelcastClientManagedContext(client, config.getManagedContext()))
                     .setPartitioningStrategy(partitioningStrategy)
                     .setHazelcastInstance(hazelcastInstance)
diff --git a/hazelcast-client/src/test/java/com/hazelcast/client/ClientDeserializationProtectionTest.java b/hazelcast-client/src/test/java/com/hazelcast/client/ClientDeserializationProtectionTest.java
new file mode 100644
index 000000000000..ffe342253c5a
--- /dev/null
+++ b/hazelcast-client/src/test/java/com/hazelcast/client/ClientDeserializationProtectionTest.java
@@ -0,0 +1,91 @@
+/*
+ * Copyright (c) 2008-2018, Hazelcast, Inc. All Rights Reserved.
+ *
+ * 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 com.hazelcast.client;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.fail;
+
+import java.io.IOException;
+import java.io.Serializable;
+
+import org.junit.After;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+
+import com.hazelcast.client.config.ClientConfig;
+import com.hazelcast.client.test.TestAwareClientFactory;
+import com.hazelcast.core.HazelcastInstance;
+import com.hazelcast.nio.serialization.HazelcastSerializationException;
+import com.hazelcast.spi.properties.GroupProperty;
+import com.hazelcast.test.HazelcastParallelClassRunner;
+import com.hazelcast.test.annotation.ParallelTest;
+import com.hazelcast.test.annotation.QuickTest;
+
+/**
+ * Tests if deserialization blacklisting works for clients
+ */
+@RunWith(HazelcastParallelClassRunner.class)
+@Category({ QuickTest.class, ParallelTest.class })
+public class ClientDeserializationProtectionTest {
+
+    private final TestAwareClientFactory factory = new TestAwareClientFactory();
+
+    @After
+    public void killAllHazelcastInstances() throws IOException {
+        factory.terminateAll();
+    }
+
+    /**
+     * <pre>
+     * Given: Serialization filter is configured with a blacklist on the client.
+     * When: Blacklisted class is deserialized.
+     * Then: The deserialization fails.
+     * </pre>
+     */
+    @Test
+    public void test() throws Exception {
+        HazelcastInstance hz = factory.newHazelcastInstance(null);
+        ClientConfig config = new ClientConfig();
+        config.setProperty(GroupProperty.SERIALIZATION_FILTER_ENABLED.getName(), "true");
+        config.setProperty(GroupProperty.SERIALIZATION_FILTER_BLACKLIST_CLASSES.getName(), TestDeserialized.class.getName());
+        HazelcastInstance client = factory.newHazelcastClient(config);
+
+        hz.getMap("test").put("test", new TestDeserialized());
+        try {
+            client.getMap("test").get("test");
+            fail("Deserialization should fail");
+        } catch (HazelcastSerializationException s) {
+            assertEquals("SecurityException was expected as a cause of failed deserialization", SecurityException.class,
+                    s.getCause().getClass());
+            assertFalse("Untrusted deserialization was possible", TestDeserialized.IS_DESERIALIZED);
+        }
+    }
+
+    public static class TestDeserialized implements Serializable {
+        private static final long serialVersionUID = 1L;
+        public static volatile boolean IS_DESERIALIZED = false;
+
+        private void writeObject(java.io.ObjectOutputStream out) throws IOException {
+        }
+
+        private void readObject(java.io.ObjectInputStream in) throws IOException, ClassNotFoundException {
+            IS_DESERIALIZED = true;
+        }
+    }
+}
diff --git a/hazelcast/src/main/java/com/hazelcast/config/ClassFilter.java b/hazelcast/src/main/java/com/hazelcast/config/ClassFilter.java
new file mode 100644
index 000000000000..0e7a8addd5c0
--- /dev/null
+++ b/hazelcast/src/main/java/com/hazelcast/config/ClassFilter.java
@@ -0,0 +1,159 @@
+/*
+ * Copyright (c) 2008-2018, Hazelcast, Inc. All Rights Reserved.
+ *
+ * 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 com.hazelcast.config;
+
+import static com.hazelcast.util.Preconditions.checkNotNull;
+import static java.util.Collections.unmodifiableSet;
+
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import com.hazelcast.logging.ILogger;
+import com.hazelcast.logging.Logger;
+
+/**
+ * Holds blacklist and whitelist configuration in java deserialization configuration.
+ */
+public class ClassFilter {
+
+    private static final String PROPERTY_CLASSNAME_LIMIT = "hazelcast.serialization.filter.classname.limit";
+    private static final int CLASSNAME_LIMIT = Integer.getInteger(PROPERTY_CLASSNAME_LIMIT, 10000);
+    private static final ILogger LOGGER = Logger.getLogger(ClassFilter.class);
+
+    private final Set<String> classes = Collections.newSetFromMap(new ConcurrentHashMap<String, Boolean>());
+    private final Set<String> packages = Collections.newSetFromMap(new ConcurrentHashMap<String, Boolean>());
+
+    private AtomicBoolean warningLogged = new AtomicBoolean();
+
+    /**
+     * Returns unmodifiable set of class names.
+     */
+    public Set<String> getClasses() {
+        return unmodifiableSet(classes);
+    }
+
+    /**
+     * Returns unmodifiable set of package names.
+     */
+    public Set<String> getPackages() {
+        return unmodifiableSet(packages);
+    }
+
+    public ClassFilter addClasses(String... names) {
+        checkNotNull(names);
+        for (String name : names) {
+            classes.add(name);
+        }
+        return this;
+    }
+
+    public ClassFilter setClasses(Collection<String> names) {
+        checkNotNull(names);
+        classes.clear();
+        classes.addAll(names);
+        return this;
+    }
+
+    public ClassFilter addPackages(String... names) {
+        checkNotNull(names);
+        for (String name : names) {
+            packages.add(name);
+        }
+        return this;
+    }
+
+    public ClassFilter setPackages(Collection<String> names) {
+        checkNotNull(names);
+        packages.clear();
+        packages.addAll(names);
+        return this;
+    }
+
+    public boolean isEmpty() {
+        return classes.isEmpty() && packages.isEmpty();
+    }
+
+    public boolean isListed(String className) {
+        if (classes.contains(className)) {
+            return true;
+        }
+        if (!packages.isEmpty()) {
+            int dotPosition = className.lastIndexOf(".");
+            if (dotPosition > 0) {
+                // String packageName = ;
+                return checkPackage(className, className.substring(0, dotPosition));
+            }
+        }
+        return false;
+    }
+
+    /**
+     * Checks if given class name is listed by package. If it's listed, then performance optimization is used and classname is
+     * added directly to {@code classes} collection.
+     *
+     * @param className Class name to be checked.
+     * @param packageName Package name of the checked class.
+     * @return {@code true} iff class is listed by-package
+     */
+    private boolean checkPackage(String className, String packageName) {
+        if (packages.contains(packageName)) {
+            if (classes.size() < CLASSNAME_LIMIT) {
+                // performance optimization
+                classes.add(className);
+            } else if (warningLogged.compareAndSet(false, true)) {
+                LOGGER.warning(String.format(
+                        "The class names collection size reached its limit. Optimizations for package names checks "
+                                + "will not optimize next usages. You can control the class names collection size limit by "
+                                + "setting system property '%s'. Actual value is %d.",
+                        PROPERTY_CLASSNAME_LIMIT, CLASSNAME_LIMIT));
+            }
+            return true;
+        }
+        return false;
+    }
+
+    @Override
+    public int hashCode() {
+        final int prime = 31;
+        int result = 1;
+        result = prime * result + ((classes == null) ? 0 : classes.hashCode());
+        result = prime * result + ((packages == null) ? 0 : packages.hashCode());
+        return result;
+    }
+
+    @Override
+    public boolean equals(Object obj) {
+        if (this == obj) {
+            return true;
+        }
+        if (obj == null || getClass() != obj.getClass()) {
+            return false;
+        }
+        ClassFilter other = (ClassFilter) obj;
+        return ((classes == null && other.classes == null) || (classes != null && classes.equals(other.classes)))
+                && ((packages == null && other.packages == null) || (packages != null && packages.equals(other.packages)));
+    }
+
+    @Override
+    public String toString() {
+        return "ClassFilter{classes=" + classes + ", packages=" + packages + "}";
+    }
+
+}
diff --git a/hazelcast/src/main/java/com/hazelcast/config/JavaSerializationFilterConfig.java b/hazelcast/src/main/java/com/hazelcast/config/JavaSerializationFilterConfig.java
new file mode 100644
index 000000000000..58d7cdf5b585
--- /dev/null
+++ b/hazelcast/src/main/java/com/hazelcast/config/JavaSerializationFilterConfig.java
@@ -0,0 +1,111 @@
+/*
+ * Copyright (c) 2008-2018, Hazelcast, Inc. All Rights Reserved.
+ *
+ * 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 com.hazelcast.config;
+
+import static com.hazelcast.spi.properties.GroupProperty.SERIALIZATION_FILTER_BLACKLIST_CLASSES;
+import static com.hazelcast.spi.properties.GroupProperty.SERIALIZATION_FILTER_BLACKLIST_PACKAGES;
+import static com.hazelcast.spi.properties.GroupProperty.SERIALIZATION_FILTER_ENABLED;
+import static com.hazelcast.spi.properties.GroupProperty.SERIALIZATION_FILTER_WHITELIST_CLASSES;
+import static com.hazelcast.spi.properties.GroupProperty.SERIALIZATION_FILTER_WHITELIST_PACKAGES;
+
+import com.hazelcast.spi.properties.HazelcastProperties;
+import com.hazelcast.util.StringUtil;
+
+/**
+ * Configuration for Serialization Filter.
+ */
+public final class JavaSerializationFilterConfig {
+
+    private final ClassFilter blacklist;
+    private final ClassFilter whitelist;
+
+    private JavaSerializationFilterConfig(ClassFilter blacklist, ClassFilter whitelist) {
+        if (blacklist == null) {
+            blacklist = new ClassFilter();
+            // default blacklist - some well-known vulnerable classes/packages
+            blacklist.addClasses(
+                    "com.sun.org.apache.xalan.internal.xsltc.trax.TemplatesImpl",
+                    "bsh.XThis",
+                    "org.apache.commons.beanutils.BeanComparator",
+                    "org.codehaus.groovy.runtime.ConvertedClosure",
+                    "org.codehaus.groovy.runtime.MethodClosure",
+                    "org.springframework.beans.factory.ObjectFactory",
+                    "com.sun.org.apache.xalan.internal.xsltc.trax.TemplatesImpl")
+            .addPackages(
+                    "org.apache.commons.collections.functors",
+                    "org.apache.commons.collections4.functors");
+        }
+        this.blacklist = blacklist;
+        this.whitelist = whitelist == null ? new ClassFilter() : whitelist;
+    }
+
+    public static JavaSerializationFilterConfig getInstance(HazelcastProperties hzProperties) {
+        if (hzProperties.getBoolean(SERIALIZATION_FILTER_ENABLED)) {
+            return new JavaSerializationFilterConfig(
+                    createClassFilter(hzProperties.getString(SERIALIZATION_FILTER_BLACKLIST_CLASSES),
+                            hzProperties.getString(SERIALIZATION_FILTER_BLACKLIST_PACKAGES)),
+                    createClassFilter(hzProperties.getString(SERIALIZATION_FILTER_WHITELIST_CLASSES),
+                            hzProperties.getString(SERIALIZATION_FILTER_WHITELIST_PACKAGES)));
+        }
+        return null;
+    }
+
+    public ClassFilter getBlacklist() {
+        return blacklist;
+    }
+
+    public ClassFilter getWhitelist() {
+        return whitelist;
+    }
+
+    @Override
+    public int hashCode() {
+        final int prime = 31;
+        int result = 1;
+        result = prime * result + ((blacklist == null) ? 0 : blacklist.hashCode());
+        result = prime * result + ((whitelist == null) ? 0 : whitelist.hashCode());
+        return result;
+    }
+
+    @Override
+    public boolean equals(Object obj) {
+        if (this == obj) {
+            return true;
+        }
+        if (obj == null || getClass() != obj.getClass()) {
+            return false;
+        }
+        JavaSerializationFilterConfig other = (JavaSerializationFilterConfig) obj;
+        return ((blacklist == null && other.blacklist == null) || (blacklist != null && blacklist.equals(other.blacklist)))
+                && ((whitelist == null && other.whitelist == null) || (whitelist != null && whitelist.equals(other.whitelist)));
+    }
+
+    @Override
+    public String toString() {
+        return "JavaSerializationFilterConfig{ blacklist=" + blacklist + ", whitelist=" + whitelist + "}";
+    }
+
+    private static ClassFilter createClassFilter(String csvClasses, String csvPackages) {
+        if (StringUtil.isNullOrEmpty(csvClasses) && StringUtil.isNullOrEmpty(csvPackages)) {
+            return null;
+        }
+        ClassFilter classFilter = new ClassFilter();
+        classFilter.addClasses(StringUtil.splitByComma(csvClasses, false))
+            .addPackages(StringUtil.splitByComma(csvPackages, false));
+        return classFilter;
+    }
+}
diff --git a/hazelcast/src/main/java/com/hazelcast/instance/DefaultNodeExtension.java b/hazelcast/src/main/java/com/hazelcast/instance/DefaultNodeExtension.java
index 2985f52b3d5c..8257a3c8745d 100644
--- a/hazelcast/src/main/java/com/hazelcast/instance/DefaultNodeExtension.java
+++ b/hazelcast/src/main/java/com/hazelcast/instance/DefaultNodeExtension.java
@@ -187,6 +187,7 @@ public InternalSerializationService createSerializationService() {
 
             ss = builder.setClassLoader(configClassLoader)
                     .setConfig(serializationConfig)
+                    .setProperties(node.getProperties())
                     .setManagedContext(hazelcastInstance.managedContext)
                     .setPartitioningStrategy(partitioningStrategy)
                     .setHazelcastInstance(hazelcastInstance)
diff --git a/hazelcast/src/main/java/com/hazelcast/internal/serialization/SerializationServiceBuilder.java b/hazelcast/src/main/java/com/hazelcast/internal/serialization/SerializationServiceBuilder.java
index 87a72d6b498d..7a372c511ca7 100644
--- a/hazelcast/src/main/java/com/hazelcast/internal/serialization/SerializationServiceBuilder.java
+++ b/hazelcast/src/main/java/com/hazelcast/internal/serialization/SerializationServiceBuilder.java
@@ -23,11 +23,13 @@
 import com.hazelcast.nio.serialization.ClassDefinition;
 import com.hazelcast.nio.serialization.DataSerializableFactory;
 import com.hazelcast.nio.serialization.PortableFactory;
+import com.hazelcast.spi.properties.HazelcastProperties;
 import com.hazelcast.spi.serialization.SerializationService;
 import com.hazelcast.util.function.Supplier;
 
 import java.nio.ByteOrder;
 
+@SuppressWarnings({"checkstyle:methodcount"})
 public interface SerializationServiceBuilder {
 
     SerializationServiceBuilder setVersion(byte version);
@@ -36,6 +38,15 @@ public interface SerializationServiceBuilder {
 
     SerializationServiceBuilder setClassLoader(ClassLoader classLoader);
 
+    /**
+     * Sets Hazelcast group properties configured for given instance.
+     *
+     * @deprecated This method was added to Hazelcast 3.10.x branch just to allow backporting of deserialization protection
+     *             feature. It's not part of Hazelcast 3.11+ API.
+     */
+    @Deprecated
+    SerializationServiceBuilder setProperties(HazelcastProperties properties);
+
     SerializationServiceBuilder setConfig(SerializationConfig config);
 
     SerializationServiceBuilder addDataSerializableFactory(int id, DataSerializableFactory factory);
diff --git a/hazelcast/src/main/java/com/hazelcast/internal/serialization/impl/AbstractSerializationService.java b/hazelcast/src/main/java/com/hazelcast/internal/serialization/impl/AbstractSerializationService.java
index c69af391c791..d481e7eee55e 100644
--- a/hazelcast/src/main/java/com/hazelcast/internal/serialization/impl/AbstractSerializationService.java
+++ b/hazelcast/src/main/java/com/hazelcast/internal/serialization/impl/AbstractSerializationService.java
@@ -88,19 +88,15 @@ public abstract class AbstractSerializationService implements InternalSerializat
     private final byte version;
     private final ILogger logger = Logger.getLogger(InternalSerializationService.class);
 
-    AbstractSerializationService(InputOutputFactory inputOutputFactory, byte version, ClassLoader classLoader,
-                                 ManagedContext managedContext, PartitioningStrategy globalPartitionStrategy,
-                                 int initialOutputBufferSize,
-                                 BufferPoolFactory bufferPoolFactory,
-                                 Supplier<RuntimeException> notActiveExceptionSupplier) {
-        this.inputOutputFactory = inputOutputFactory;
-        this.version = version;
-        this.classLoader = classLoader;
-        this.managedContext = managedContext;
-        this.globalPartitioningStrategy = globalPartitionStrategy;
-        this.outputBufferSize = initialOutputBufferSize;
-        this.bufferPoolThreadLocal = new BufferPoolThreadLocal(this, bufferPoolFactory,
-                notActiveExceptionSupplier);
+    AbstractSerializationService(Builder<?> builder) {
+        this.inputOutputFactory = builder.inputOutputFactory;
+        this.version = builder.version;
+        this.classLoader = builder.classLoader;
+        this.managedContext = builder.managedContext;
+        this.globalPartitioningStrategy = builder.globalPartitionStrategy;
+        this.outputBufferSize = builder.initialOutputBufferSize;
+        this.bufferPoolThreadLocal = new BufferPoolThreadLocal(this, builder.bufferPoolFactory,
+                builder.notActiveExceptionSupplier);
         this.nullSerializerAdapter = createSerializerAdapter(new ConstantSerializers.NullSerializer(), this);
     }
 
@@ -564,4 +560,64 @@ private SerializerAdapter lookupJavaSerializer(Class type) {
         }
         return null;
     }
+
+    public abstract static class Builder<T extends Builder<T>> {
+        private InputOutputFactory inputOutputFactory;
+        private byte version;
+        private ClassLoader classLoader;
+        private ManagedContext managedContext;
+        private PartitioningStrategy globalPartitionStrategy;
+        private int initialOutputBufferSize;
+        private BufferPoolFactory bufferPoolFactory;
+        private Supplier<RuntimeException> notActiveExceptionSupplier;
+
+        protected Builder() {
+        }
+
+        protected abstract T self();
+
+        public final T withInputOutputFactory(InputOutputFactory inputOutputFactory) {
+            this.inputOutputFactory = inputOutputFactory;
+            return self();
+        }
+
+        public final T withVersion(byte version) {
+            this.version = version;
+            return self();
+        }
+
+        public final T withClassLoader(ClassLoader classLoader) {
+            this.classLoader = classLoader;
+            return self();
+        }
+
+        public ClassLoader getClassLoader() {
+            return classLoader;
+        }
+
+        public final T withManagedContext(ManagedContext managedContext) {
+            this.managedContext = managedContext;
+            return self();
+        }
+
+        public final T withGlobalPartitionStrategy(PartitioningStrategy globalPartitionStrategy) {
+            this.globalPartitionStrategy = globalPartitionStrategy;
+            return self();
+        }
+
+        public final T withInitialOutputBufferSize(int initialOutputBufferSize) {
+            this.initialOutputBufferSize = initialOutputBufferSize;
+            return self();
+        }
+
+        public final T withBufferPoolFactory(BufferPoolFactory bufferPoolFactory) {
+            this.bufferPoolFactory = bufferPoolFactory;
+            return self();
+        }
+
+        public final T withNotActiveExceptionSupplier(Supplier<RuntimeException> notActiveExceptionSupplier) {
+            this.notActiveExceptionSupplier = notActiveExceptionSupplier;
+            return self();
+        }
+    }
 }
diff --git a/hazelcast/src/main/java/com/hazelcast/internal/serialization/impl/DefaultSerializationServiceBuilder.java b/hazelcast/src/main/java/com/hazelcast/internal/serialization/impl/DefaultSerializationServiceBuilder.java
index 02f74b74c8a3..7f2b1bbf2954 100644
--- a/hazelcast/src/main/java/com/hazelcast/internal/serialization/impl/DefaultSerializationServiceBuilder.java
+++ b/hazelcast/src/main/java/com/hazelcast/internal/serialization/impl/DefaultSerializationServiceBuilder.java
@@ -17,6 +17,7 @@
 package com.hazelcast.internal.serialization.impl;
 
 import com.hazelcast.config.GlobalSerializerConfig;
+import com.hazelcast.config.JavaSerializationFilterConfig;
 import com.hazelcast.config.SerializationConfig;
 import com.hazelcast.core.HazelcastInstance;
 import com.hazelcast.core.HazelcastInstanceAware;
@@ -29,6 +30,8 @@
 import com.hazelcast.internal.serialization.SerializationServiceBuilder;
 import com.hazelcast.internal.serialization.impl.bufferpool.BufferPoolFactoryImpl;
 import com.hazelcast.nio.ClassLoaderUtil;
+import com.hazelcast.nio.ClassNameFilter;
+import com.hazelcast.nio.SerializationClassNameFilter;
 import com.hazelcast.nio.serialization.ClassDefinition;
 import com.hazelcast.nio.serialization.DataSerializableFactory;
 import com.hazelcast.nio.serialization.HazelcastSerializationException;
@@ -36,6 +39,7 @@
 import com.hazelcast.nio.serialization.Serializer;
 import com.hazelcast.nio.serialization.SerializerHook;
 import com.hazelcast.spi.properties.GroupProperty;
+import com.hazelcast.spi.properties.HazelcastProperties;
 import com.hazelcast.util.StringUtil;
 import com.hazelcast.util.function.Supplier;
 
@@ -49,6 +53,7 @@
 import static java.nio.ByteOrder.LITTLE_ENDIAN;
 import static java.nio.ByteOrder.nativeOrder;
 
+@SuppressWarnings({"checkstyle:methodcount"})
 public class DefaultSerializationServiceBuilder implements SerializationServiceBuilder {
 
     // System property to override configured byte order for tests
@@ -87,6 +92,8 @@ public class DefaultSerializationServiceBuilder implements SerializationServiceB
 
     protected Supplier<RuntimeException> notActiveExceptionSupplier;
 
+    protected ClassNameFilter classNameFilter;
+
     @Override
     public SerializationServiceBuilder setVersion(byte version) {
         byte maxVersion = BuildInfoProvider.getBuildInfo().getSerializationVersion();
@@ -215,6 +222,14 @@ public SerializationServiceBuilder setInitialOutputBufferSize(int initialOutputB
         return this;
     }
 
+
+    @Override
+    public SerializationServiceBuilder setProperties(HazelcastProperties properties) {
+        JavaSerializationFilterConfig filterConfig = JavaSerializationFilterConfig.getInstance(properties);
+        classNameFilter = filterConfig == null ? null : new SerializationClassNameFilter(filterConfig);
+        return this;
+    }
+
     @Override
     public InternalSerializationService build() {
         initVersions();
@@ -272,10 +287,22 @@ protected InternalSerializationService createSerializationService(InputOutputFac
                                                                       Supplier<RuntimeException> notActiveExceptionSupplier) {
         switch (version) {
             case 1:
-                SerializationServiceV1 serializationServiceV1 = new SerializationServiceV1(inputOutputFactory, version,
-                        portableVersion, classLoader, dataSerializableFactories, portableFactories, managedContext,
-                        partitioningStrategy, initialOutputBufferSize, new BufferPoolFactoryImpl(), enableCompression,
-                        enableSharedObject, notActiveExceptionSupplier);
+                SerializationServiceV1 serializationServiceV1 = SerializationServiceV1.builder()
+                    .withInputOutputFactory(inputOutputFactory)
+                    .withVersion(version)
+                    .withPortableVersion(portableVersion)
+                    .withClassLoader(classLoader)
+                    .withDataSerializableFactories(dataSerializableFactories)
+                    .withPortableFactories(portableFactories)
+                    .withManagedContext(managedContext)
+                    .withGlobalPartitionStrategy(partitioningStrategy)
+                    .withInitialOutputBufferSize(initialOutputBufferSize)
+                    .withBufferPoolFactory(new BufferPoolFactoryImpl())
+                    .withEnableCompression(enableCompression)
+                    .withEnableSharedObject(enableSharedObject)
+                    .withNotActiveExceptionSupplier(notActiveExceptionSupplier)
+                    .withClassNameFilter(classNameFilter)
+                    .build();
                 serializationServiceV1.registerClassDefinitions(classDefinitions, checkClassDefErrors);
                 return serializationServiceV1;
 
diff --git a/hazelcast/src/main/java/com/hazelcast/internal/serialization/impl/JavaDefaultSerializers.java b/hazelcast/src/main/java/com/hazelcast/internal/serialization/impl/JavaDefaultSerializers.java
index 2f1236fb23e6..0472e320c8d7 100644
--- a/hazelcast/src/main/java/com/hazelcast/internal/serialization/impl/JavaDefaultSerializers.java
+++ b/hazelcast/src/main/java/com/hazelcast/internal/serialization/impl/JavaDefaultSerializers.java
@@ -18,6 +18,7 @@
 
 import com.hazelcast.nio.BufferObjectDataInput;
 import com.hazelcast.nio.ClassLoaderUtil;
+import com.hazelcast.nio.ClassNameFilter;
 import com.hazelcast.nio.ObjectDataInput;
 import com.hazelcast.nio.ObjectDataOutput;
 import com.hazelcast.nio.serialization.HazelcastSerializationException;
@@ -53,10 +54,12 @@ public static final class JavaSerializer extends SingletonSerializer<Object> {
 
         private final boolean shared;
         private final boolean gzipEnabled;
+        private final ClassNameFilter classFilter;
 
-        public JavaSerializer(boolean shared, boolean gzipEnabled) {
+        public JavaSerializer(boolean shared, boolean gzipEnabled, ClassNameFilter classFilter) {
             this.shared = shared;
             this.gzipEnabled = gzipEnabled;
+            this.classFilter = classFilter;
         }
 
         @Override
@@ -74,7 +77,7 @@ public Object read(final ObjectDataInput in) throws IOException {
 
         private Object read(InputStream in, ClassLoader classLoader) throws IOException {
             try {
-                ObjectInputStream objectInputStream = newObjectInputStream(classLoader, in);
+                ObjectInputStream objectInputStream = newObjectInputStream(classLoader, classFilter, in);
                 if (shared) {
                     return objectInputStream.readObject();
                 }
@@ -130,9 +133,11 @@ private void writeGzipped(OutputStream out, Object obj) throws IOException {
     public static final class ExternalizableSerializer extends SingletonSerializer<Externalizable> {
 
         private final boolean gzipEnabled;
+        private final ClassNameFilter classFilter;
 
-        public ExternalizableSerializer(boolean gzipEnabled) {
+        public ExternalizableSerializer(boolean gzipEnabled, ClassNameFilter classFilter) {
             this.gzipEnabled = gzipEnabled;
+            this.classFilter = classFilter;
         }
 
         @Override
@@ -167,7 +172,7 @@ private Externalizable readGzipped(InputStream in, String className, ClassLoader
 
         private Externalizable read(InputStream in, String className, ClassLoader classLoader) throws Exception {
             Externalizable ds = ClassLoaderUtil.newInstance(classLoader, className);
-            ObjectInputStream objectInputStream = newObjectInputStream(classLoader, in);
+            ObjectInputStream objectInputStream = newObjectInputStream(classLoader, classFilter, in);
             ds.readExternal(objectInputStream);
             return ds;
         }
diff --git a/hazelcast/src/main/java/com/hazelcast/internal/serialization/impl/SerializationServiceV1.java b/hazelcast/src/main/java/com/hazelcast/internal/serialization/impl/SerializationServiceV1.java
index e8ffcc93cb3c..384848a5686a 100644
--- a/hazelcast/src/main/java/com/hazelcast/internal/serialization/impl/SerializationServiceV1.java
+++ b/hazelcast/src/main/java/com/hazelcast/internal/serialization/impl/SerializationServiceV1.java
@@ -16,15 +16,13 @@
 
 package com.hazelcast.internal.serialization.impl;
 
-import com.hazelcast.core.ManagedContext;
 import com.hazelcast.core.PartitioningStrategy;
-import com.hazelcast.internal.serialization.InputOutputFactory;
 import com.hazelcast.internal.serialization.PortableContext;
 import com.hazelcast.internal.serialization.impl.ConstantSerializers.BooleanSerializer;
 import com.hazelcast.internal.serialization.impl.ConstantSerializers.ByteSerializer;
 import com.hazelcast.internal.serialization.impl.ConstantSerializers.StringArraySerializer;
-import com.hazelcast.internal.serialization.impl.bufferpool.BufferPoolFactory;
 import com.hazelcast.nio.BufferObjectDataInput;
+import com.hazelcast.nio.ClassNameFilter;
 import com.hazelcast.nio.ObjectDataInput;
 import com.hazelcast.nio.serialization.ClassDefinition;
 import com.hazelcast.nio.serialization.Data;
@@ -37,7 +35,6 @@
 import com.hazelcast.nio.serialization.Portable;
 import com.hazelcast.nio.serialization.PortableFactory;
 import com.hazelcast.nio.serialization.PortableReader;
-import com.hazelcast.util.function.Supplier;
 
 import java.io.Externalizable;
 import java.io.IOException;
@@ -46,6 +43,7 @@
 import java.math.BigInteger;
 import java.util.ArrayList;
 import java.util.Collection;
+import java.util.Collections;
 import java.util.Date;
 import java.util.LinkedList;
 import java.util.Map;
@@ -86,28 +84,23 @@ public class SerializationServiceV1 extends AbstractSerializationService {
     private final PortableContextImpl portableContext;
     private final PortableSerializer portableSerializer;
 
-     SerializationServiceV1(InputOutputFactory inputOutputFactory, byte version, int portableVersion, ClassLoader classLoader,
-            Map<Integer, ? extends DataSerializableFactory> dataSerializableFactories,
-            Map<Integer, ? extends PortableFactory> portableFactories, ManagedContext managedContext,
-            PartitioningStrategy globalPartitionStrategy, int initialOutputBufferSize, BufferPoolFactory bufferPoolFactory,
-            boolean enableCompression, boolean enableSharedObject, Supplier<RuntimeException> notActiveExceptionSupplier) {
-        super(inputOutputFactory, version, classLoader, managedContext, globalPartitionStrategy, initialOutputBufferSize,
-                bufferPoolFactory, notActiveExceptionSupplier);
-
-        PortableHookLoader loader = new PortableHookLoader(portableFactories, classLoader);
-        portableContext = new PortableContextImpl(this, portableVersion);
+    SerializationServiceV1(AbstractBuilder<?> builder) {
+        super(builder);
+        PortableHookLoader loader = new PortableHookLoader(builder.portableFactories, builder.getClassLoader());
+        portableContext = new PortableContextImpl(this, builder.portableVersion);
         for (ClassDefinition cd : loader.getDefinitions()) {
             portableContext.registerClassDefinition(cd);
         }
 
         dataSerializerAdapter = createSerializerAdapter(
-                new DataSerializableSerializer(dataSerializableFactories, classLoader), this);
+                new DataSerializableSerializer(builder.dataSerializableFactories, builder.getClassLoader()), this);
         portableSerializer = new PortableSerializer(portableContext, loader.getFactories());
         portableSerializerAdapter = createSerializerAdapter(portableSerializer, this);
 
-        javaSerializerAdapter = createSerializerAdapter(new JavaSerializer(enableSharedObject, enableCompression), this);
+        javaSerializerAdapter = createSerializerAdapter(
+                new JavaSerializer(builder.enableSharedObject, builder.enableCompression, builder.classNameFilter), this);
         javaExternalizableAdapter = createSerializerAdapter(
-                new JavaDefaultSerializers.ExternalizableSerializer(enableCompression), this);
+                new JavaDefaultSerializers.ExternalizableSerializer(builder.enableCompression, builder.classNameFilter), this);
         registerConstantSerializers();
         registerJavaTypeSerializers();
     }
@@ -248,10 +241,77 @@ public ObjectDataInput initDataSerializableInputAndSkipTheHeader(Data data) thro
         return input;
     }
 
+    public static Builder builder() {
+        return new Builder();
+    }
+
     private void skipBytesSafely(ObjectDataInput input, int count) throws IOException {
         if (input.skipBytes(count) != count) {
             throw new HazelcastSerializationException("Malformed serialization format");
         }
     }
 
+    public abstract static class AbstractBuilder<T extends AbstractBuilder<T>> extends AbstractSerializationService.Builder<T> {
+
+        private int portableVersion;
+        private Map<Integer, ? extends DataSerializableFactory> dataSerializableFactories = Collections.emptyMap();
+        private Map<Integer, ? extends PortableFactory> portableFactories = Collections.emptyMap();
+        private boolean enableCompression;
+        private boolean enableSharedObject;
+        private ClassNameFilter classNameFilter;
+
+        protected AbstractBuilder() {
+        }
+
+        public final T withPortableVersion(int portableVersion) {
+            this.portableVersion = portableVersion;
+            return self();
+        }
+
+        public final T withDataSerializableFactories(
+                Map<Integer, ? extends DataSerializableFactory> dataSerializableFactories) {
+            this.dataSerializableFactories = dataSerializableFactories;
+            return self();
+        }
+
+        public Map<Integer, ? extends DataSerializableFactory> getDataSerializableFactories() {
+            return dataSerializableFactories;
+        }
+
+        public final T withPortableFactories(Map<Integer, ? extends PortableFactory> portableFactories) {
+            this.portableFactories = portableFactories;
+            return self();
+        }
+
+        public final T withEnableCompression(boolean enableCompression) {
+            this.enableCompression = enableCompression;
+            return self();
+        }
+
+        public final T withEnableSharedObject(boolean enableSharedObject) {
+            this.enableSharedObject = enableSharedObject;
+            return self();
+        }
+
+        public final T withClassNameFilter(ClassNameFilter classNameFilter) {
+            this.classNameFilter = classNameFilter;
+            return self();
+        }
+    }
+
+    public static final class Builder extends AbstractBuilder<Builder> {
+
+        protected Builder() {
+        }
+
+        @Override
+        protected Builder self() {
+            return this;
+        }
+
+        public SerializationServiceV1 build() {
+            return new SerializationServiceV1(this);
+        }
+
+    }
 }
diff --git a/hazelcast/src/main/java/com/hazelcast/nio/ClassNameFilter.java b/hazelcast/src/main/java/com/hazelcast/nio/ClassNameFilter.java
new file mode 100644
index 000000000000..46fcbee38eeb
--- /dev/null
+++ b/hazelcast/src/main/java/com/hazelcast/nio/ClassNameFilter.java
@@ -0,0 +1,32 @@
+/*
+ * Copyright (c) 2008-2018, Hazelcast, Inc. All Rights Reserved.
+ *
+ * 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 com.hazelcast.nio;
+
+/**
+ * Allows to intercept class resolution during deserialization based on classname. It's used as a validation mechanism in
+ * Look-ahead ObjectInputStream.
+ */
+public interface ClassNameFilter {
+
+    /**
+     * Called from {@link java.io.ObjectInputStream#resolveClass}. When the deserialization should not be allowed the method
+     * throws a {@link RuntimeException}.
+     *
+     * @param className name of class to be deserialized
+     */
+    void filter(String className);
+}
diff --git a/hazelcast/src/main/java/com/hazelcast/nio/IOUtil.java b/hazelcast/src/main/java/com/hazelcast/nio/IOUtil.java
index bf1674c6655d..4e4d987fd55c 100644
--- a/hazelcast/src/main/java/com/hazelcast/nio/IOUtil.java
+++ b/hazelcast/src/main/java/com/hazelcast/nio/IOUtil.java
@@ -157,8 +157,9 @@ public static void readFully(InputStream in, byte[] buffer) throws IOException {
         }
     }
 
-    public static ObjectInputStream newObjectInputStream(final ClassLoader classLoader, InputStream in) throws IOException {
-        return new ClassLoaderAwareObjectInputStream(classLoader, in);
+    public static ObjectInputStream newObjectInputStream(final ClassLoader classLoader, ClassNameFilter classFilter,
+            InputStream in) throws IOException {
+        return new ClassLoaderAwareObjectInputStream(classLoader, classFilter, in);
     }
 
     public static OutputStream newOutputStream(final ByteBuffer dst) {
@@ -596,15 +597,22 @@ public static String toDebugString(String name, ByteBuffer byteBuffer) {
     private static final class ClassLoaderAwareObjectInputStream extends ObjectInputStream {
 
         private final ClassLoader classLoader;
+        private final ClassNameFilter classFilter;
 
-        private ClassLoaderAwareObjectInputStream(final ClassLoader classLoader, final InputStream in) throws IOException {
+        private ClassLoaderAwareObjectInputStream(final ClassLoader classLoader, ClassNameFilter classFilter,
+                final InputStream in) throws IOException {
             super(in);
             this.classLoader = classLoader;
+            this.classFilter = classFilter;
         }
 
         @Override
         protected Class<?> resolveClass(ObjectStreamClass desc) throws ClassNotFoundException {
-            return ClassLoaderUtil.loadClass(classLoader, desc.getName());
+            String name = desc.getName();
+            if (classFilter != null) {
+                classFilter.filter(name);
+            }
+            return ClassLoaderUtil.loadClass(classLoader, name);
         }
 
         @Override
diff --git a/hazelcast/src/main/java/com/hazelcast/nio/SerializationClassNameFilter.java b/hazelcast/src/main/java/com/hazelcast/nio/SerializationClassNameFilter.java
new file mode 100644
index 000000000000..4a8f01ba685e
--- /dev/null
+++ b/hazelcast/src/main/java/com/hazelcast/nio/SerializationClassNameFilter.java
@@ -0,0 +1,56 @@
+/*
+ * Copyright (c) 2008-2018, Hazelcast, Inc. All Rights Reserved.
+ *
+ * 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 com.hazelcast.nio;
+
+import static java.lang.String.format;
+
+import com.hazelcast.config.ClassFilter;
+import com.hazelcast.config.JavaSerializationFilterConfig;
+import com.hazelcast.util.Preconditions;
+
+/**
+ * Implementation of basic protection against untrusted deserialization. It holds blacklist and whitelist with classnames and
+ * package names.
+ *
+ * @see #filter(String)
+ */
+public final class SerializationClassNameFilter implements ClassNameFilter {
+
+    private static final String DESERIALIZATION_ERROR = "Resolving class %s is not allowed.";
+
+    private final ClassFilter blacklist;
+    private final ClassFilter whitelist;
+
+    public SerializationClassNameFilter(JavaSerializationFilterConfig config) {
+        Preconditions.checkNotNull(config, "JavaSerializationFilterConfig has to be provided");
+        blacklist = config.getBlacklist();
+        whitelist = config.getWhitelist();
+    }
+
+    /**
+     * Throws {@link SecurityException} if the given class name appears on the blacklist or does not appear on a non-empty
+     * whitelist.
+     *
+     * @param className class name to check
+     * @throws SecurityException if the classname is not allowed for deserialization
+     */
+    public void filter(String className) throws SecurityException {
+        if (blacklist.isListed(className) || (!whitelist.isEmpty() && !whitelist.isListed(className))) {
+            throw new SecurityException(format(DESERIALIZATION_ERROR, className));
+        }
+    }
+}
diff --git a/hazelcast/src/main/java/com/hazelcast/spi/properties/GroupProperty.java b/hazelcast/src/main/java/com/hazelcast/spi/properties/GroupProperty.java
index e157b022fa03..ba7f8cc30de6 100644
--- a/hazelcast/src/main/java/com/hazelcast/spi/properties/GroupProperty.java
+++ b/hazelcast/src/main/java/com/hazelcast/spi/properties/GroupProperty.java
@@ -928,6 +928,36 @@ public final class GroupProperty {
             = new HazelcastProperty("hazelcast.serialization.version",
             BuildInfoProvider.getBuildInfo().getSerializationVersion());
 
+    /**
+     * Enables Java deserialization protection - filtering based on blacklist and whitelist.
+     */
+    public static final HazelcastProperty SERIALIZATION_FILTER_ENABLED =
+            new HazelcastProperty("hazelcast.serialization.filter.enabled", false);
+
+    /**
+     * Holds comma separated list of blacklisted class names in Java deserialization protection feature.
+     */
+    public static final HazelcastProperty SERIALIZATION_FILTER_BLACKLIST_CLASSES =
+            new HazelcastProperty("hazelcast.serialization.filter.blacklist.classes", "");
+
+    /**
+     * Holds comma separated list of blacklisted package names in Java deserialization protection feature.
+     */
+    public static final HazelcastProperty SERIALIZATION_FILTER_BLACKLIST_PACKAGES =
+            new HazelcastProperty("hazelcast.serialization.filter.blacklist.packages", "");
+
+    /**
+     * Holds comma separated list of whitelisted class names in Java deserialization protection feature.
+     */
+    public static final HazelcastProperty SERIALIZATION_FILTER_WHITELIST_CLASSES =
+            new HazelcastProperty("hazelcast.serialization.filter.whitelist.classes", "");
+
+    /**
+     * Holds comma separated list of whitelisted package names in Java deserialization protection feature.
+     */
+    public static final HazelcastProperty SERIALIZATION_FILTER_WHITELIST_PACKAGES =
+            new HazelcastProperty("hazelcast.serialization.filter.whitelist.packages", "");
+
     /**
      * Override cluster version to use while node is not yet member of a cluster. The cluster version assumed before joining
      * a cluster may affect the serialization format of cluster discovery & join operations and its compatibility with members
diff --git a/hazelcast/src/test/java/com/hazelcast/cluster/MulticastDeserializationTest.java b/hazelcast/src/test/java/com/hazelcast/cluster/MulticastDeserializationTest.java
new file mode 100644
index 000000000000..33a3f10fecc9
--- /dev/null
+++ b/hazelcast/src/test/java/com/hazelcast/cluster/MulticastDeserializationTest.java
@@ -0,0 +1,135 @@
+/*
+ * Copyright (c) 2008-2018, Hazelcast, Inc. All Rights Reserved.
+ *
+ * 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 com.hazelcast.cluster;
+
+import static org.junit.Assert.assertFalse;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.ObjectOutputStream;
+import java.io.Serializable;
+import java.net.DatagramPacket;
+import java.net.InetAddress;
+import java.net.MulticastSocket;
+import java.nio.ByteBuffer;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+
+import com.hazelcast.config.Config;
+import com.hazelcast.config.JoinConfig;
+import com.hazelcast.config.MulticastConfig;
+import com.hazelcast.config.NetworkConfig;
+import com.hazelcast.core.Hazelcast;
+import com.hazelcast.instance.HazelcastInstanceFactory;
+import com.hazelcast.internal.serialization.impl.SerializationConstants;
+import com.hazelcast.nio.IOUtil;
+import com.hazelcast.nio.Packet;
+import com.hazelcast.spi.properties.GroupProperty;
+import com.hazelcast.test.HazelcastSerialClassRunner;
+import com.hazelcast.test.annotation.QuickTest;
+
+/**
+ * Tests if deserialization blacklisting works for MutlicastService.
+ */
+@RunWith(HazelcastSerialClassRunner.class)
+@Category(QuickTest.class)
+public class MulticastDeserializationTest {
+
+    private static final int MULTICAST_PORT = 53535;
+    private static final String MULTICAST_GROUP = "224.0.0.219";
+    // TTL==0 : Restricted to the same host. Won't be output by any interface.
+    private static final int MULTICAST_TTL = 0;
+
+    @Before
+    @After
+    public void killAllHazelcastInstances() throws IOException {
+        HazelcastInstanceFactory.terminateAll();
+    }
+
+    /**
+     * <pre>
+     * Given: Multicast is configured.
+     * When: DatagramPacket with a correct Packet comes. The Packet references Java serializer and the serialized object is not a Join message.
+     * Then: The object from the Packet is not deserialized.
+     * </pre>
+     */
+    @Test
+    public void test() throws Exception {
+        Config config = new Config();
+        config.setProperty(GroupProperty.SERIALIZATION_FILTER_ENABLED.getName(), "true");
+        config.setProperty(GroupProperty.SERIALIZATION_FILTER_BLACKLIST_CLASSES.getName(), TestDeserialized.class.getName());
+        NetworkConfig networkConfig = config.getNetworkConfig();
+        JoinConfig join = networkConfig.getJoin();
+        join.getTcpIpConfig().setEnabled(false);
+        MulticastConfig multicastConfig = join.getMulticastConfig();
+        multicastConfig.setMulticastPort(MULTICAST_PORT);
+        multicastConfig.setMulticastGroup(MULTICAST_GROUP);
+        multicastConfig.setMulticastTimeToLive(MULTICAST_TTL);
+        multicastConfig.setEnabled(true);
+
+        Hazelcast.newHazelcastInstance(config);
+        sendJoinDatagram(new TestDeserialized());
+        Thread.sleep(500L);
+        assertFalse("Untrusted deserialization is possible", TestDeserialized.IS_DESERIALIZED);
+    }
+
+    private void sendJoinDatagram(Object object) throws IOException {
+        ByteArrayOutputStream bos = new ByteArrayOutputStream();
+        ObjectOutputStream oos = new ObjectOutputStream(bos);
+        try {
+            oos.writeObject(object);
+        } finally {
+            IOUtil.closeResource(oos);
+        }
+        byte[] data = bos.toByteArray();
+        MulticastSocket multicastSocket = null;
+        try {
+            multicastSocket = new MulticastSocket(MULTICAST_PORT);
+            multicastSocket.setTimeToLive(MULTICAST_TTL);
+            InetAddress group = InetAddress.getByName(MULTICAST_GROUP);
+            multicastSocket.joinGroup(group);
+            int msgSize = data.length;
+
+            ByteBuffer bbuf = ByteBuffer.allocate(1 + 4 + msgSize);
+            bbuf.put(Packet.VERSION);
+            bbuf.putInt(SerializationConstants.JAVA_DEFAULT_TYPE_SERIALIZABLE);
+            bbuf.put(data);
+            byte[] packetData = bbuf.array();
+            DatagramPacket packet = new DatagramPacket(packetData, packetData.length, group, MULTICAST_PORT);
+            multicastSocket.send(packet);
+            multicastSocket.leaveGroup(group);
+        } finally {
+            IOUtil.closeResource(multicastSocket);
+        }
+    }
+
+    public static class TestDeserialized implements Serializable {
+        private static final long serialVersionUID = 1L;
+        public static volatile boolean IS_DESERIALIZED = false;
+
+        private void writeObject(java.io.ObjectOutputStream out) throws IOException {
+        }
+
+        private void readObject(java.io.ObjectInputStream in) throws IOException, ClassNotFoundException {
+            IS_DESERIALIZED = true;
+        }
+    }
+}
diff --git a/hazelcast/src/test/java/com/hazelcast/nio/SerializationClassNameFilterTest.java b/hazelcast/src/test/java/com/hazelcast/nio/SerializationClassNameFilterTest.java
new file mode 100644
index 000000000000..8f2e2c18304c
--- /dev/null
+++ b/hazelcast/src/test/java/com/hazelcast/nio/SerializationClassNameFilterTest.java
@@ -0,0 +1,137 @@
+/*
+ * Copyright (c) 2008-2018, Hazelcast, Inc. All Rights Reserved.
+ *
+ * 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 com.hazelcast.nio;
+
+import java.util.Properties;
+
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+
+import com.hazelcast.config.JavaSerializationFilterConfig;
+import com.hazelcast.spi.properties.GroupProperty;
+import com.hazelcast.spi.properties.HazelcastProperties;
+import com.hazelcast.test.HazelcastParallelClassRunner;
+import com.hazelcast.test.annotation.QuickTest;
+
+/**
+ * Unit tests for {@link SerializationClassNameFilter}.
+ */
+@RunWith(HazelcastParallelClassRunner.class)
+@Category(QuickTest.class)
+public class SerializationClassNameFilterTest {
+
+    /**
+     * <pre>
+     * Given: Neither whitelist nor blacklist is configured.
+     * When: {@link SerializationClassNameFilter#filter(String)} is called.
+     * Then: no exception is thrown
+     * </pre>
+     */
+    @Test
+    public void testNoList() {
+        JavaSerializationFilterConfig config = createFilterConfig();
+        new SerializationClassNameFilter(config).filter("java.lang.Object");
+    }
+
+    private JavaSerializationFilterConfig createFilterConfig() {
+        Properties properties = new Properties();
+        properties.setProperty(GroupProperty.SERIALIZATION_FILTER_ENABLED.getName(), "true");
+        return JavaSerializationFilterConfig.getInstance(new HazelcastProperties(properties));
+    }
+
+    /**
+     * <pre>
+     * Given: Default blacklist is used.
+     * When: {@link SerializationClassNameFilter#filter(String)} is called for a class name which is included in the default blacklist.
+     * Then: {@link SecurityException} is thrown
+     * </pre>
+     */
+    @Test(expected = SecurityException.class)
+    public void testDefaultBlacklist() throws ClassNotFoundException {
+        new SerializationClassNameFilter(createFilterConfig()).filter("bsh.XThis");
+    }
+
+    /**
+     * <pre>
+     * Given: Default blacklist is used.
+     * When: {@link SerializationClassNameFilter#filter(String)} is called for a class in package which is included in the default blacklist.
+     * Then: {@link SecurityException} is thrown
+     * </pre>
+     */
+    @Test(expected = SecurityException.class)
+    public void testPackageDefaultBlacklisted() throws ClassNotFoundException {
+        new SerializationClassNameFilter(createFilterConfig()).filter("org.apache.commons.collections.functors.Test");
+    }
+
+    /**
+     * <pre>
+     * Given: Whitelist is set.
+     * When: {@link SerializationClassNameFilter#filter(String)} is called for a whitelisted class.
+     * Then: no exception is thrown
+     * </pre>
+     */
+    @Test
+    public void testClassInWhitelist() throws ClassNotFoundException {
+        JavaSerializationFilterConfig config = createFilterConfig();
+        config.getWhitelist().addClasses("java.lang.Test1", "java.lang.Test2", "java.lang.Test3");
+        new SerializationClassNameFilter(config).filter("java.lang.Test2");
+    }
+
+    /**
+     * <pre>
+     * Given: Whitelist is set.
+     * When: {@link SerializationClassNameFilter#filter(String)} is called for a class which has whitelisted package.
+     * Then: no exception is thrown
+     * </pre>
+     */
+    @Test
+    public void testPackageInWhitelist() throws ClassNotFoundException {
+        JavaSerializationFilterConfig config = createFilterConfig();
+        config.getWhitelist().addPackages("com.whitelisted");
+        new SerializationClassNameFilter(config).filter("com.whitelisted.Test2");
+    }
+
+    /**
+     * <pre>
+     * Given: Whitelist is set.
+     * When: {@link SerializationClassNameFilter#filter(String)} is called for a not whitelisted class.
+     * Then: {@link SecurityException} is thrown
+     * </pre>
+     */
+    @Test(expected = SecurityException.class)
+    public void testClassNotInWhitelist() throws ClassNotFoundException {
+        JavaSerializationFilterConfig config = createFilterConfig();
+        config.getWhitelist().addClasses("java.lang.Test1", "java.lang.Test2", "java.lang.Test3");
+        new SerializationClassNameFilter(config).filter("java.lang.Test4");
+    }
+
+    /**
+     * <pre>
+     * Given: Blacklist and Whitelist are set.
+     * When: {@link SerializationClassNameFilter#filter(String)} is called for a class which is whitelisted and blacklisted together.
+     * Then: {@link SecurityException} is thrown
+     * </pre>
+     */
+    @Test(expected = SecurityException.class)
+    public void testWhitelistedAndBlacklisted() throws ClassNotFoundException {
+        JavaSerializationFilterConfig config = createFilterConfig();
+        config.getWhitelist().addClasses("java.lang.Test1", "java.lang.Test2", "java.lang.Test3");
+        config.getBlacklist().addClasses("java.lang.Test3", "java.lang.Test2", "java.lang.Test1");
+        new SerializationClassNameFilter(config).filter("java.lang.Test1");
+    }
+}
\ No newline at end of file
diff --git a/hazelcast/src/test/java/com/hazelcast/nio/serialization/SerializationTest.java b/hazelcast/src/test/java/com/hazelcast/nio/serialization/SerializationTest.java
index 87c168cd16c1..39a267e125db 100644
--- a/hazelcast/src/test/java/com/hazelcast/nio/serialization/SerializationTest.java
+++ b/hazelcast/src/test/java/com/hazelcast/nio/serialization/SerializationTest.java
@@ -85,7 +85,7 @@ public void testGlobalSerializer_withOverrideJavaSerializable() {
         globalSerializerConfig.setOverrideJavaSerialization(true);
         final AtomicInteger writeCounter = new AtomicInteger();
         final AtomicInteger readCounter = new AtomicInteger();
-        final JavaSerializer javaSerializer = new JavaSerializer(true, false);
+        final JavaSerializer javaSerializer = new JavaSerializer(true, false, null);
         SerializationConfig serializationConfig = new SerializationConfig().setGlobalSerializerConfig(
                 globalSerializerConfig.setImplementation(new StreamSerializer<Object>() {
                     @Override