@Override
	public void bindProducer(final String name, MessageChannel moduleOutputChannel, Properties properties) {

		Assert.isInstanceOf(SubscribableChannel.class, moduleOutputChannel);
		KafkaPropertiesAccessor accessor = new KafkaPropertiesAccessor(properties);
		if (name.startsWith(P2P_NAMED_CHANNEL_TYPE_PREFIX) || name.startsWith(PUBSUB_NAMED_CHANNEL_TYPE_PREFIX)) {
			validateProducerProperties(name, properties, SUPPORTED_NAMED_PRODUCER_PROPERTIES);
		}
		else {
			validateProducerProperties(name, properties, SUPPORTED_PRODUCER_PROPERTIES);
		}

		Collection<Partition> partitions;
		if (!bindNewProducerDirectlyIfPossible(name, (SubscribableChannel) moduleOutputChannel, accessor)) {
			if (logger.isInfoEnabled()) {
				logger.info("Using Kafka topic for outbound: " + name);
			}

			final String topicName = escapeTopicName(name);
			int defaultPartitionCount = accessor.getNumberOfKafkaPartitionsForProducer();

			partitions = initializePartitions(name, topicName, defaultPartitionCount);

			ProducerMetadata<byte[], byte[]> producerMetadata = new ProducerMetadata<>(topicName, byte[].class,
					byte[].class, BYTE_ARRAY_SERIALIZER, BYTE_ARRAY_SERIALIZER);
			producerMetadata.setCompressionType(ProducerMetadata.CompressionType
					.valueOf(accessor.getCompressionCodec(this.defaultCompressionCodec)));
			producerMetadata.setBatchBytes(accessor.getBatchSize(this.defaultBatchSize));
			producerMetadata.setSync(accessor.getSyncProducer(this.defaultSyncProducer));
			producerMetadata.setSendTimeout(accessor.getSyncProducerTimeout(this.defaultSyncProducerTimeout));
			Properties additionalProps = new Properties();
			additionalProps.put(ProducerConfig.ACKS_CONFIG,
					String.valueOf(accessor.getRequiredAcks(this.defaultRequiredAcks)));
			additionalProps.put(ProducerConfig.LINGER_MS_CONFIG,
					String.valueOf(accessor.getBatchTimeout(this.defaultBatchTimeout)));
			ProducerFactoryBean<byte[], byte[]> producerFB = new ProducerFactoryBean<>(producerMetadata, brokers,
					additionalProps);

			try {
				final ProducerConfiguration<byte[], byte[]> producerConfiguration = new ProducerConfiguration<>(
						producerMetadata, producerFB.getObject());
				MessageHandler handler = new SendingHandler(topicName, accessor, partitions.size(),
						producerConfiguration);
				EventDrivenConsumer consumer = new EventDrivenConsumer((SubscribableChannel) moduleOutputChannel,
						handler);
				consumer.setBeanFactory(this.getBeanFactory());
				consumer.setBeanName("outbound." + name);
				consumer.afterPropertiesSet();
				Binding producerBinding = Binding.forProducer(name, moduleOutputChannel, consumer, accessor);
				addBinding(producerBinding);
				producerBinding.start();
			}
			catch (Exception e) {
				throw new RuntimeException(e);
			}

		}

	}
Set the {@link OffsetManagement} to use. Default:
{@link OffsetManagement#kafkaTopic}.
@param offsetManagement the offsetManagement.
spring-xd-messagebus-kafka/src/main/java/org/springframework/xd/dirt/integration/kafka/KafkaMessageBus.java