<?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:int="http://www.springframework.org/schema/integration"
	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/integration http://www.springframework.org/schema/integration/spring-integration.xsd
		http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">

	<import resource="kafkaInboundAdapterCommon-context.xml"/>

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

	<int-kafka:inbound-channel-adapter id="kafkaInboundChannelAdapter"
                                       kafka-consumer-context-ref="consumerContext"
                                       auto-startup="false"
                                       channel="inputFromKafka">
        <int:poller fixed-delay="10" time-unit="MILLISECONDS" max-messages-per-poll="5"/>
    </int-kafka:inbound-channel-adapter>

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

    <int-kafka:consumer-context id="consumerContext"
                                consumer-timeout="4000"
                                zookeeper-connect="zookeeperConnect">
        <int-kafka:consumer-configurations>
            <int-kafka:consumer-configuration group-id="default"
                    value-decoder="valueDecoder"
                    key-decoder="valueDecoder"
                    max-messages="5000">
                <int-kafka:topic id="test1" streams="4"/>
                <int-kafka:topic id="test2" streams="4"/>
            </int-kafka:consumer-configuration>
        </int-kafka:consumer-configurations>
    </int-kafka:consumer-context>
</beans>
