@Override
		@SuppressWarnings("unchecked")
		protected Object handleRequestMessage(Message<?> requestMessage) {
			if (Mode.embeddedHeaders.equals(mode)) {
				MessageValues messageValues;
				try {
					messageValues = embeddedHeadersMessageConverter.extractHeaders((Message<byte[]>) requestMessage,
							true);
				}
				catch (Exception e) {
					logger.error(EmbeddedHeadersMessageConverter.decodeExceptionMessage(requestMessage), e);
					messageValues = new MessageValues(requestMessage);
				}
				messageValues = deserializePayloadIfNecessary(messageValues);
				return MessageBuilder.createMessage(messageValues.getPayload(), new KafkaBusMessageHeaders(
						messageValues));
			}
			else {
				return requestMessage;
			}
		}
Fetches partition data, throwing an exception if the expected number is not met after a number of
subsequent retries
@param topicName topic
@param expectedNumber expected number of partitions. If less or equal to 0, check is suppressed.
@return the partitions
spring-xd-messagebus-kafka/src/main/java/org/springframework/xd/dirt/integration/kafka/KafkaMessageBus.java