lundi 13 avril 2015

Spring Integration Kafka

I'm trying to parse messages to kafka from my spring integration application. I'm pretty sure i should set a KafkaProducerMessageHandler as my endpoint, like this



@Bean
@ServiceActivator(inputChannel = "correctChannel")
public KafkaProducerMessageHandler<String, String> handler() throws Exception{
return new KafkaProducerMessageHandler<String, String>(kafkaContext());
}


Then i'm trying to build all the arguments for the constructor, like this



@Bean
public KafkaProducerContext<String, String> kafkaContext () throws Exception{
KafkaProducerContext<String, String> ctx = new KafkaProducerContext<String, String>();
ctx.setProducerConfigurations(Collections.singletonMap("test", config()));
return ctx;
}

@Bean
public ProducerConfiguration<String, String> config() throws Exception{
return new ProducerConfiguration<String, String>(metaData(), producer().getObject());
}

@Bean
public Encoder<String> encoder(){
return new StringEncoder<String>();
}

@Bean
public ProducerMetadata<String, String> metaData(){
ProducerMetadata<String, String> meta = new ProducerMetadata<String,String>("test");
meta.setValueClassType(String.class);
meta.setKeyClassType(String.class);
meta.setValueEncoder(encoder());
meta.setKeyEncoder(encoder());
return meta;
}

@Bean
public ProducerFactoryBean<String, String> producer(){
return new ProducerFactoryBean<String,String>(metaData(), "localhost:9092");
}


But the ProducerConfiguration part throws this exception:



Exception in thread "main" org.springframework.beans.factory.BeanCreationException: Error creating bean with name 'config' defined in class path resource [demo/IntegrationConfiguration.class]: Bean instantiation via factory method failed; nested exception is org.springframework.beans.BeanInstantiationException: Failed to instantiate [org.springframework.integration.kafka.support.ProducerConfiguration]: Factory method 'config' threw exception; nested exception is java.lang.IllegalArgumentException: Superclass has no null constructors but no arguments were given
at org.springframework.beans.factory.support.ConstructorResolver.instantiateUsingFactoryMethod(ConstructorResolver.java:599)
at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.instantiateUsingFactoryMethod(AbstractAutowireCapableBeanFactory.java:1111)
at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.createBeanInstance(AbstractAutowireCapableBeanFactory.java:1006)
at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.doCreateBean(AbstractAutowireCapableBeanFactory.java:504)
at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.createBean(AbstractAutowireCapableBeanFactory.java:476)
at org.springframework.beans.factory.support.AbstractBeanFactory$1.getObject(AbstractBeanFactory.java:303)
at org.springframework.beans.factory.support.DefaultSingletonBeanRegistry.getSingleton(DefaultSingletonBeanRegistry.java:230)
at org.springframework.beans.factory.support.AbstractBeanFactory.doGetBean(AbstractBeanFactory.java:299)
at org.springframework.beans.factory.support.AbstractBeanFactory.getBean(AbstractBeanFactory.java:194)
at org.springframework.beans.factory.support.DefaultListableBeanFactory.preInstantiateSingletons(DefaultListableBeanFactory.java:755)
at org.springframework.context.support.AbstractApplicationContext.finishBeanFactoryInitialization(AbstractApplicationContext.java:757)
at org.springframework.context.support.AbstractApplicationContext.refresh(AbstractApplicationContext.java:480)
at org.springframework.boot.SpringApplication.refresh(SpringApplication.java:686)
at org.springframework.boot.SpringApplication.run(SpringApplication.java:320)
at org.springframework.boot.SpringApplication.run(SpringApplication.java:957)
at org.springframework.boot.SpringApplication.run(SpringApplication.java:946)
at demo.IntegrationnnnApplication.main(IntegrationnnnApplication.java:12)
Caused by: org.springframework.beans.BeanInstantiationException: Failed to instantiate [org.springframework.integration.kafka.support.ProducerConfiguration]: Factory method 'config' threw exception; nested exception is java.lang.IllegalArgumentException: Superclass has no null constructors but no arguments were given
at org.springframework.beans.factory.support.SimpleInstantiationStrategy.instantiate(SimpleInstantiationStrategy.java:189)
at org.springframework.beans.factory.support.ConstructorResolver.instantiateUsingFactoryMethod(ConstructorResolver.java:588)
... 16 more
Caused by: java.lang.IllegalArgumentException: Superclass has no null constructors but no arguments were given
at org.springframework.cglib.proxy.Enhancer.emitConstructors(Enhancer.java:721)
at org.springframework.cglib.proxy.Enhancer.generateClass(Enhancer.java:499)
at org.springframework.cglib.core.DefaultGeneratorStrategy.generate(DefaultGeneratorStrategy.java:25)
at org.springframework.cglib.core.AbstractClassGenerator.create(AbstractClassGenerator.java:216)
at org.springframework.cglib.proxy.Enhancer.createHelper(Enhancer.java:377)
at org.springframework.cglib.proxy.Enhancer.create(Enhancer.java:285)
at org.springframework.context.annotation.ConfigurationClassEnhancer$BeanMethodInterceptor.enhanceFactoryBean(ConfigurationClassEnhancer.java:384)
at org.springframework.context.annotation.ConfigurationClassEnhancer$BeanMethodInterceptor.intercept(ConfigurationClassEnhancer.java:292)
at demo.IntegrationConfiguration$$EnhancerBySpringCGLIB$$8445d892.producer(<generated>)
at demo.IntegrationConfiguration.config(IntegrationConfiguration.java:178)
at demo.IntegrationConfiguration$$EnhancerBySpringCGLIB$$8445d892.CGLIB$config$0(<generated>)
at demo.IntegrationConfiguration$$EnhancerBySpringCGLIB$$8445d892$$FastClassBySpringCGLIB$$59e71af6.invoke(<generated>)
at org.springframework.cglib.proxy.MethodProxy.invokeSuper(MethodProxy.java:228)
at org.springframework.context.annotation.ConfigurationClassEnhancer$BeanMethodInterceptor.intercept(ConfigurationClassEnhancer.java:309)
at demo.IntegrationConfiguration$$EnhancerBySpringCGLIB$$8445d892.config(<generated>)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:483)
at org.springframework.beans.factory.support.SimpleInstantiationStrategy.instantiate(SimpleInstantiationStrategy.java:162)
... 17 more


I do not know what that's supposed to mean.


Another question: Is it possible to send a message to kafka with multiple payloads?


Aucun commentaire:

Enregistrer un commentaire