Class object used by StreamReceiver is not send to remote node

## issue

When I use IgniteDataStreamer with user defined class, like this:
```
cfg.setPeerClassLoadingEnabled(true);
Ignite ignite = Ignition.start(cfg);

IgniteDataStreamer mktStmr = ignite.dataStreamer(mktCache.getName());
```

I got the following error, it seems that the `MyDouble` class is not send to remote node:
```
SEVERE: DataStreamer operation failed.
class org.apache.ignite.IgniteCheckedException: Failed to finish operation (too many remaps): 32
at org.apache.ignite.internal.processors.datastreamer.DataStreamerImpl$5.apply(DataStreamerImpl.java:977)
at org.apache.ignite.internal.processors.datastreamer.DataStreamerImpl$5.apply(DataStreamerImpl.java:942)
at org.apache.ignite.internal.util.future.GridFutureAdapter.notifyListener(GridFutureAdapter.java:474)
at org.apache.ignite.internal.util.future.GridFutureAdapter.unblock(GridFutureAdapter.java:350)
at org.apache.ignite.internal.util.future.GridFutureAdapter.unblockAll(GridFutureAdapter.java:338)
at org.apache.ignite.internal.util.future.GridFutureAdapter.onDone(GridFutureAdapter.java:586)
at org.apache.ignite.internal.util.future.GridFutureAdapter.onDone(GridFutureAdapter.java:565)
at org.apache.ignite.internal.util.future.GridFutureAdapter.onDone(GridFutureAdapter.java:553)
at org.apache.ignite.internal.util.future.GridCompoundFuture.apply(GridCompoundFuture.java:132)
at org.apache.ignite.internal.util.future.GridCompoundFuture.apply(GridCompoundFuture.java:46)
at org.apache.ignite.internal.util.future.GridFutureAdapter.notifyListener(GridFutureAdapter.java:474)
at org.apache.ignite.internal.util.future.GridFutureAdapter.unblock(GridFutureAdapter.java:350)
at org.apache.ignite.internal.util.future.GridFutureAdapter.unblockAll(GridFutureAdapter.java:338)
at org.apache.ignite.internal.util.future.GridFutureAdapter.onDone(GridFutureAdapter.java:586)
at org.apache.ignite.internal.util.future.GridFutureAdapter.onDone(GridFutureAdapter.java:565)
at org.apache.ignite.internal.processors.datastreamer.DataStreamerImpl$Buffer.onResponse(DataStreamerImpl.java:2121)
at org.apache.ignite.internal.processors.datastreamer.DataStreamerImpl$3.onMessage(DataStreamerImpl.java:368)
at org.apache.ignite.internal.managers.communication.GridIoManager.invokeListener(GridIoManager.java:1906)
at org.apache.ignite.internal.managers.communication.GridIoManager.processRegularMessage0(GridIoManager.java:1527)
at org.apache.ignite.internal.managers.communication.GridIoManager.access$5300(GridIoManager.java:242)
at org.apache.ignite.internal.managers.communication.GridIoManager$9.execute(GridIoManager.java:1420)
at org.apache.ignite.internal.managers.communication.TraceRunnable.run(TraceRunnable.java:55)
at org.apache.ignite.internal.util.StripedExecutor$Stripe.body(StripedExecutor.java:637)
at org.apache.ignite.internal.util.worker.GridWorker.run(GridWorker.java:125)
at java.lang.Thread.run(Thread.java:750)
Caused by: class org.apache.ignite.IgniteCheckedException: DataStreamer request failed [node=4167fc2a-b0c4-45a7-bf8e-52a97457ad34]
at org.apache.ignite.internal.processors.datastreamer.DataStreamerImpl$Buffer.onResponse(DataStreamerImpl.java:2112)
... 9 more
Caused by: class org.apache.ignite.binary.BinaryInvalidTypeException: com.nimblex.StreamVisitorExample$MyDouble
at org.apache.ignite.internal.binary.BinaryContext.descriptorForTypeId(BinaryContext.java:741)
at org.apache.ignite.internal.binary.BinaryReaderExImpl.deserialize0(BinaryReaderExImpl.java:1772)
at org.apache.ignite.internal.binary.BinaryReaderExImpl.deserialize(BinaryReaderExImpl.java:1731)
at org.apache.ignite.internal.binary.BinaryObjectImpl.deserializeValue(BinaryObjectImpl.java:866)
at org.apache.ignite.internal.binary.BinaryObjectImpl.value(BinaryObjectImpl.java:198)
at org.apache.ignite.internal.processors.cache.CacheObjectUtils.unwrapBinary(CacheObjectUtils.java:199)
at org.apache.ignite.internal.processors.cache.CacheObjectUtils.unwrapBinaryIfNeeded(CacheObjectUtils.java:78)
at org.apache.ignite.internal.processors.cache.CacheObjectContext.unwrapBinaryIfNeeded(CacheObjectContext.java:138)
at org.apache.ignite.internal.processors.datastreamer.DataStreamerEntry$1.getValue(DataStreamerEntry.java:96)
at com.nimblex.StreamVisitorExample$DeviceStreamReceiver.receive(StreamVisitorExample.java:143)
at org.apache.ignite.internal.processors.datastreamer.DataStreamerUpdateJob.call(DataStreamerUpdateJob.java:141)
at org.apache.ignite.internal.processors.datastreamer.DataStreamProcessor.localUpdate(DataStreamProcessor.java:394)
at org.apache.ignite.internal.processors.datastreamer.DataStreamProcessor.processRequest(DataStreamProcessor.java:299)
at org.apache.ignite.internal.processors.datastreamer.DataStreamProcessor.access$000(DataStreamProcessor.java:60)
at org.apache.ignite.internal.processors.datastreamer.DataStreamProcessor$1.onMessage(DataStreamProcessor.java:90)
... 8 more
Caused by: java.lang.ClassNotFoundException: com.nimblex.StreamVisitorExample$MyDouble
at java.net.URLClassLoader.findClass(URLClassLoader.java:407)
at java.lang.ClassLoader.loadClass(ClassLoader.java:418)
at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:352)
at java.lang.ClassLoader.loadClass(ClassLoader.java:351)
at java.lang.Class.forName0(Native Method)
at java.lang.Class.forName(Class.java:348)
at org.apache.ignite.internal.util.IgniteUtils.forName(IgniteUtils.java:9373)
at org.apache.ignite.internal.util.IgniteUtils.forName(IgniteUtils.java:9311)
at org.apache.ignite.internal.MarshallerContextImpl.getClass(MarshallerContextImpl.java:384)
at org.apache.ignite.internal.binary.BinaryContext.descriptorForTypeId(BinaryContext.java:717)
... 22 more
```

## code
The code is modifed from https://github.com/apache/ignite/blob/master/examples/src/main/java/org/apache/ignite/examples/streaming/StreamVisitorExample.java

```java
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You 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.nimblex;

import java.io.Serializable;
import java.util.Arrays;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.concurrent.atomic.AtomicLong;

import org.apache.ignite.Ignite;
import org.apache.ignite.IgniteCache;
import org.apache.ignite.IgniteDataStreamer;
import org.apache.ignite.Ignition;
import org.apache.ignite.cache.query.SqlFieldsQuery;
import org.apache.ignite.cache.query.annotations.QuerySqlField;
import org.apache.ignite.configuration.CacheConfiguration;
import org.apache.ignite.configuration.DeploymentMode;
import org.apache.ignite.configuration.IgniteConfiguration;
import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi;
import org.apache.ignite.spi.discovery.tcp.ipfinder.multicast.TcpDiscoveryMulticastIpFinder;
import org.apache.ignite.stream.StreamReceiver;
import org.apache.ignite.stream.StreamVisitor;

public class StreamVisitorExample {
/** Random number generator. */
private static final Random RAND = new Random();

/** The list of instruments. */
private static final String[] INSTRUMENTS = {"IBM", "GOOG", "MSFT", "GE", "EBAY", "YHOO", "ORCL", "CSCO", "AMZN", "RHT"};

/** The list of initial instrument prices. */
private static final double[] INITIAL_PRICES = {194.9, 893.49, 34.21, 23.24, 57.93, 45.03, 44.41, 28.44, 378.49, 69.50};

public static void main(String[] args) throws Exception {

TcpDiscoverySpi spi = new TcpDiscoverySpi();
// create a new instance of tcp discovery multicast ip finder
TcpDiscoveryMulticastIpFinder tcMp = new TcpDiscoveryMulticastIpFinder();
tcMp.setAddresses(Arrays.asList("192.168.8.63")); // change your IP address here
// set the multi cast ip finder for spi
spi.setIpFinder(tcMp);
// create new ignite configuration
IgniteConfiguration cfg = new IgniteConfiguration();
cfg.setClientMode(true);
// set the discovery spi to ignite configuration
cfg.setDiscoverySpi(spi);

// fixbug： Remote node has deployment mode different from local
cfg.setPeerClassLoadingEnabled(true);
// cfg.setDeploymentMode(DeploymentMode.CONTINUOUS);

try (Ignite ignite = Ignition.start(cfg)) {

// Market data cache with default configuration.
CacheConfiguration mktDataCfg = new CacheConfiguration<>("marketTicks");

// Financial instrument cache configuration.
CacheConfiguration instCfg = new CacheConfiguration<>("instCache");

// Index key and value for querying financial instruments.
// Note that Instrument class has @QuerySqlField annotation for secondary field indexing.
instCfg.setIndexedTypes(String.class, Instrument.class);

// Auto-close caches at the end of the example.
try (
IgniteCache mktCache = ignite.getOrCreateCache(mktDataCfg);
IgniteCache instCache = ignite.getOrCreateCache(instCfg)
) {
try (IgniteDataStreamer mktStmr = ignite.dataStreamer(mktCache.getName())) {
// Note that we receive market data, but do not populate 'mktCache' (it remains empty).
// Instead we update the instruments in the 'instCache'.
// Since both, 'instCache' and 'mktCache' use the same key, updates are collocated.
System.out.println("register receiver: ");

mktStmr.receiver(new DeviceStreamReceiver(instCache));

// Stream 10 million market data ticks into the system.
System.out.println("begin write data: ");
for (int i = 1; i <= 1; i++) {
int idx = RAND.nextInt(INSTRUMENTS.length);

// Use gaussian distribution to ensure that
// numbers closer to 0 have higher probability.
double price = round2(INITIAL_PRICES[idx] + RAND.nextGaussian());

mktStmr.addData(INSTRUMENTS[idx], new MyDouble(price));

if (i % 500_000 == 0)
System.out.println("Number of tuples streamed into Ignite: " + i);
}
}

System.out.println("try to query data: ");

// Select top 3 best performing instruments.
SqlFieldsQuery top3qry = new SqlFieldsQuery(
"select symbol, (latest - open) from Instrument order by (latest - open) desc limit 3");

// Execute queries.
List> top3 = instCache.query(top3qry).getAll();

System.out.println("Top performing financial instruments: " + top3.toString());
}
finally {
// Distributed cache could be removed from cluster only by #destroyCache() call.
ignite.destroyCache(mktDataCfg.getName());
ignite.destroyCache(instCfg.getName());
}
}
}

public static class DeviceStreamReceiver implements StreamReceiver {
private IgniteCache instCache;
public DeviceStreamReceiver(IgniteCache cache) {
instCache = cache;
}

@Override
public void receive(IgniteCache cache, Collection> entries) {
//CommunicationTrackKafkaProducer producer = CommunicationTrackKafkaProducer.getInstance();

for (Map.Entry entry : entries) {
String symbol = entry.getKey();
MyDouble tick = entry.getValue();

Instrument inst = instCache.get(symbol);

if (inst == null)
inst = new Instrument(symbol);

// Don't populate market cache, as we don't use it for querying.
// Update cached instrument based on the latest market tick.
inst.update(tick.getValue());

instCache.put(symbol, inst);
}
}
}

/**
* Rounds double value to two significant signs.
*
* @param val value to be rounded.
* @return rounded double value.
*/
private static double round2(double val) {
return Math.floor(100 * val + 0.5) / 100;
}

/**
* Financial instrument.
*/
public static class Instrument implements Serializable {
/** Instrument symbol. */
@QuerySqlField(index = true)
private final String symbol;

/** Open price. */
@QuerySqlField(index = true)
private double open;

/** Close price. */
@QuerySqlField(index = true)
private double latest;

/**
* @param symbol Symbol.
*/
public Instrument(String symbol) {
this.symbol = symbol;
}

/**
* Updates this instrument based on the latest market tick price.
*
* @param price Latest price.
*/
public void update(double price) {
if (open == 0)
open = price;

this.latest = price;
}
}

public static class MyDouble implements Serializable {
private Double value;

public MyDouble(Double value) {
this.value = value;
}

public Double getValue() {
return value;
}

public void setValue(Double value) {
this.value = value;
}
}
}
```

Start with examples/src/main/java/org/apache/ignite/examples/streaming/StreamVisitorExample.java and the DataStreamerImpl and DataStreamProcessor paths named in the stack trace. Reproduce the shown custom MyDouble and DeviceStreamReceiver setup with peer class loading enabled, then inspect why the remote node cannot resolve the class. Done means the receiver processes the streamed value without BinaryInvalidTypeException or repeated remaps.

Written by the indexing model from the issue text.