<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
	   xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
	   xmlns:context="http://www.springframework.org/schema/context"
	   xmlns:int-kafka="http://www.springframework.org/schema/integration/kafka"
	   xsi:schemaLocation="http://www.springframework.org/schema/integration/kafka
	   http://www.springframework.org/schema/integration/kafka/spring-integration-kafka.xsd
        http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd
		http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">

	<context:property-placeholder/>

	<int-kafka:zookeeper-connect id="zookeeperConnect" zk-connect="localhost:2181" zk-connection-timeout="6000"
								 zk-session-timeout="6000"
								 zk-sync-time="2000"/>

	<bean id="valueDecoder"
		  class="org.springframework.integration.kafka.serializer.avro.AvroReflectDatumBackedKafkaDecoder">
		<constructor-arg type="java.lang.Class" value="java.lang.String"/>
	</bean>

	<bean id="consumerProperties" class="org.springframework.beans.factory.config.PropertiesFactoryBean">
		<property name="properties">
			<props>
				<prop key="auto.offset.reset">largest</prop>
				<prop key="socket.receive.buffer.bytes">10485760</prop>
				<!-- 10M -->
				<prop key="fetch.message.max.bytes">5242880</prop>
				<prop key="auto.commit.interval.ms">1000</prop>
			</props>
		</property>
	</bean>

	<int-kafka:consumer-context id="consumerContext1"
								consumer-timeout="4000"
								zookeeper-connect="zookeeperConnect"
								consumer-properties="consumerProperties">
		<int-kafka:consumer-configurations>
			<int-kafka:consumer-configuration group-id="${groupId:default1}"
											  value-decoder="valueDecoder"
											  key-decoder="valueDecoder"
											  max-messages="5000">
				<int-kafka:topic id="test1" streams="${streams:3}"/>
				<int-kafka:topic id="test2" streams="4"/>
			</int-kafka:consumer-configuration>
			<int-kafka:consumer-configuration group-id="${groupId:default2}"
											  value-decoder="valueDecoder"
											  key-decoder="valueDecoder"
											  max-messages="5000">
				<int-kafka:topic id="test3" streams="${streams:1}"/>
			</int-kafka:consumer-configuration>
		</int-kafka:consumer-configurations>
	</int-kafka:consumer-context>
	<int-kafka:consumer-context id="consumerContext2"
								consumer-timeout="4000"
								zookeeper-connect="zookeeperConnect"
								consumer-properties="consumerProperties">
		<int-kafka:consumer-configurations>
			<int-kafka:consumer-configuration group-id="${groupId:default1}"
											  value-decoder="valueDecoder"
											  key-decoder="valueDecoder"
											  max-messages="5000">
				<int-kafka:topic id="test4" streams="${streams:3}"/>
				<int-kafka:topic id="test5" streams="4"/>
			</int-kafka:consumer-configuration>
		</int-kafka:consumer-configurations>
	</int-kafka:consumer-context>

</beans>
