@Override
	public void fromTwisterMessage(TwisterMessage message)
			throws SerializationException {
		// First byte is the message type, has already been in message
		// Read the refId if any and set the boolean flag.
		readRefIdIfAny(message);
		this.jobId = message.readString();
		this.mapTaskNo = message.readInt();
		this.iteration = message.readInt();
		int numKeys = message.readInt();
		if (numKeys > 0) {
			keyClass = message.readString();
			valClass = message.readString();
			CustomClassLoader classLoader = DaemonWorker.getClassLoader(jobId);
			if (classLoader == null) {
				throw new SerializationException(
						"Could not find a class loader for this job id.");
			}
			Class<?> kClass = null;
			try {
				kClass = Class.forName(keyClass, true, classLoader);
			} catch (ClassNotFoundException e) {
				e.printStackTrace();
				throw new SerializationException(e);
			}
			Class<?> vClass = null;
			try {
				vClass = Class.forName(valClass, true, classLoader);
			} catch (ClassNotFoundException e) {
				e.printStackTrace();
				throw new SerializationException(e);
			}
			Key key = null;
			Value val = null;
			for (int i = 0; i < numKeys; i++) {
				try {
					key = (Key) kClass.newInstance();
				} catch (Exception e) {
					e.printStackTrace();
					throw new SerializationException(e);
				}
				key.fromTwisterMessage(message);
				try {
					val = (Value) vClass.newInstance();
				} catch (Exception e) {
					e.printStackTrace();
					throw new SerializationException(e);
				}
				val.fromTwisterMessage(message);
				// Add the key value pair.
				addKeyValue(key, val);
			}
		}
		responseTopic = message.readString();
	}