Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

INTEXT-139: Deprecate High Level Consumer Adapter

Browse Source 
JIRA: https://jira.spring.io/browse/INTEXT-139

Revert removal for test and deprecate them
This commit is contained in:
Artem Bilan
2015-10-29 18:47:21 -04:00
committed by Artem Bilan
parent 28aca5332e
commit 0f756cc1ec
18 changed files with 104 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaConsumerContextParser.java
View File

@@ -48,6 +48,8 @@ import org.springframework.util.xml.DomUtils;
* @author Gary Russell
* @since 0.5
*/
@Deprecated
@SuppressWarnings("deprecation")
public class KafkaConsumerContextParser extends AbstractSingleBeanDefinitionParser {
@Override

9
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaInboundChannelAdapterParser.java
View File

@@ -15,14 +15,14 @@
*/
package org.springframework.integration.kafka.config.xml;
import org.w3c.dom.Element;
import org.springframework.beans.BeanMetadataElement;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.config.xml.AbstractPollingInboundChannelAdapterParser;
import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
import org.springframework.integration.kafka.inbound.KafkaHighLevelConsumerMessageSource;
import org.springframework.util.StringUtils;
import org.w3c.dom.Element;
/**
* The Kafka Inbound Channel adapter parser
@@ -31,11 +31,14 @@ import org.w3c.dom.Element;
* @since 0.5
*
*/
@Deprecated
public class KafkaInboundChannelAdapterParser extends AbstractPollingInboundChannelAdapterParser {
@Override
@SuppressWarnings("deprecation")
protected BeanMetadataElement parseSource(final Element element, final ParserContext parserContext) {
final BeanDefinitionBuilder highLevelConsumerMessageSourceBuilder =
BeanDefinitionBuilder.genericBeanDefinition(KafkaHighLevelConsumerMessageSource.class);
BeanDefinitionBuilder.genericBeanDefinition(org.springframework.integration.kafka.inbound.KafkaHighLevelConsumerMessageSource.class);
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(highLevelConsumerMessageSourceBuilder, element, "kafka-decoder");

1
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaNamespaceHandler.java
View File

@@ -28,6 +28,7 @@ import org.springframework.integration.config.xml.AbstractIntegrationNamespaceHa
public class KafkaNamespaceHandler extends AbstractIntegrationNamespaceHandler {
@Override
@SuppressWarnings("deprecation")
public void init() {
registerBeanDefinitionParser("zookeeper-connect", new ZookeeperConnectParser());
registerBeanDefinitionParser("inbound-channel-adapter", new KafkaInboundChannelAdapterParser());

9
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaHighLevelConsumerMessageSource.java
View File

@@ -15,19 +15,20 @@
*/
package org.springframework.integration.kafka.inbound;
import java.util.List;
import java.util.Map;
import org.springframework.integration.context.IntegrationObjectSupport;
import org.springframework.integration.core.MessageSource;
import org.springframework.integration.kafka.support.KafkaConsumerContext;
import org.springframework.messaging.Message;
import java.util.List;
import java.util.Map;
/**
* @author Soby Chacko
* @since 0.5
*
* @deprecated since 1.3 in favor of {@link KafkaMessageDrivenChannelAdapter}
*/
@Deprecated
public class KafkaHighLevelConsumerMessageSource<K,V> extends IntegrationObjectSupport implements MessageSource<Map<String, Map<Integer, List<Object>>>> {
private final KafkaConsumerContext<K,V> kafkaConsumerContext;

6
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/ConsumerConfigFactoryBean.java
View File

@@ -17,19 +17,23 @@ package org.springframework.integration.kafka.support;
import java.util.Properties;
import kafka.consumer.ConsumerConfig;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.FactoryBean;
import kafka.consumer.ConsumerConfig;
/**
* @author Soby Chacko
* @since 0.5
*/
@Deprecated
@SuppressWarnings("deprecation")
public class ConsumerConfigFactoryBean<K,V> implements FactoryBean<ConsumerConfig> {
private static final Log LOGGER = LogFactory.getLog(ConsumerConfigFactoryBean.class);
private final ConsumerMetadata<K,V> consumerMetadata;
private final ZookeeperConnect zookeeperConnect;
private Properties consumerProperties = new Properties();

3
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/ConsumerConfiguration.java
View File

@@ -39,7 +39,10 @@ import kafka.message.MessageAndMetadata;
* @author Artem Bilan
* @since 0.5
*/
@Deprecated
@SuppressWarnings("deprecation")
public class ConsumerConfiguration<K, V> {
private static final Log LOGGER = LogFactory.getLog(ConsumerConfiguration.class);
private final ConsumerMetadata<K, V> consumerMetadata;

1
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/ConsumerConnectionProvider.java
View File

@@ -22,6 +22,7 @@ import kafka.javaapi.consumer.ConsumerConnector;
* @author Soby Chacko
* @since 0.5
*/
@Deprecated
public class ConsumerConnectionProvider {
private final ConsumerConfig consumerConfig;

6
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/ConsumerMetadata.java
View File

@@ -17,16 +17,18 @@ package org.springframework.integration.kafka.support;
import java.util.Map;
import kafka.serializer.Decoder;
import kafka.serializer.DefaultDecoder;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.integration.kafka.core.KafkaConsumerDefaults;
import kafka.serializer.Decoder;
import kafka.serializer.DefaultDecoder;
/**
* @author Soby Chacko
* @author Rajasekar Elango
* @since 0.5
*/
@Deprecated
public class ConsumerMetadata<K,V> implements InitializingBean {
//High level consumer defaults

3
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/KafkaConsumerContext.java
View File

@@ -31,7 +31,10 @@ import org.springframework.util.CollectionUtils;
* @author Ilayaperumal Gopinathan
* @since 0.5
*/
@Deprecated
@SuppressWarnings("deprecation")
public class KafkaConsumerContext<K, V> implements DisposableBean {
private Map<String, ConsumerConfiguration<K, V>> consumerConfigurations;
private String consumerTimeout = KafkaConsumerDefaults.CONSUMER_TIMEOUT;

5
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/MessageLeftOverTracker.java
View File

@@ -15,15 +15,16 @@
*/
package org.springframework.integration.kafka.support;
import kafka.message.MessageAndMetadata;
import java.util.ArrayList;
import java.util.List;
import kafka.message.MessageAndMetadata;
/**
* @author Soby Chacko
* @since 0.5
*/
@Deprecated
public class MessageLeftOverTracker<K,V> {
private final List<MessageAndMetadata<K,V>> messageLeftOverFromPreviousPoll = new ArrayList<MessageAndMetadata<K,V>>();

1
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/TopicFilterConfiguration.java
View File

@@ -25,6 +25,7 @@ import kafka.consumer.Whitelist;
* @author Artem Bilan
* @since 0.5
*/
@Deprecated
public class TopicFilterConfiguration {
private final int numberOfStreams;

10
spring-integration-kafka/src/main/resources/org/springframework/integration/kafka/config/spring-integration-kafka-1.3.xsd
View File

@@ -133,7 +133,9 @@
<xsd:annotation>
<xsd:appinfo>
<xsd:documentation>
Custom implementation of a Kafka Encoder for encoding message values. This option is deprecated, 'value-serializer' is the recommended option.
[DEPRECATED]
Custom implementation of a Kafka Encoder for encoding message values.
This option is deprecated, 'value-serializer' is the recommended option.
</xsd:documentation>
<tool:annotation kind="ref">
<tool:expected-type type="kafka.serializer.Encoder"/>
@@ -157,7 +159,9 @@
<xsd:annotation>
<xsd:appinfo>
<xsd:documentation>
[DEPRECATED]
Custom implementation of a Kafka Encoder for encoding message keys.
This option is deprecated, 'key-serializer' is the recommended option.
</xsd:documentation>
<tool:annotation kind="ref">
<tool:expected-type type="kafka.serializer.Encoder"/>
@@ -267,6 +271,7 @@
<xsd:element name="consumer-context">
<xsd:annotation>
<xsd:documentation><![CDATA[
[DEPRECATED]
Defines a producer context.
]]></xsd:documentation>
</xsd:annotation>
@@ -423,8 +428,11 @@
<xsd:element name="inbound-channel-adapter">
<xsd:annotation>
<xsd:documentation>
[DEPRECATED]
The definition for the Spring Integration Kafka
Inbound Channel Adapter.
This High Level Consumer Adapter is deprecated in favor of
message-driven-channel-adapter - based on the Simple Consumer API.
</xsd:documentation>
</xsd:annotation>
<xsd:complexType>

22
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaConsumerContextParserTests.java
View File

@@ -19,7 +19,6 @@ package org.springframework.integration.kafka.config.xml;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertThat;
import kafka.consumer.Blacklist;
import org.hamcrest.Matchers;
import org.junit.Assert;
@@ -30,13 +29,11 @@ import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.integration.kafka.rule.KafkaRunning;
import org.springframework.integration.kafka.support.ConsumerConfiguration;
import org.springframework.integration.kafka.support.ConsumerMetadata;
import org.springframework.integration.kafka.support.KafkaConsumerContext;
import org.springframework.integration.kafka.support.TopicFilterConfiguration;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import kafka.consumer.Blacklist;
/**
* @author Soby Chacko
* @author Artem Bilan
@@ -45,6 +42,8 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
*/
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration
@Deprecated
@SuppressWarnings("deprecation")
public class KafkaConsumerContextParserTests<K, V> {
@ClassRule
@@ -56,13 +55,16 @@ public class KafkaConsumerContextParserTests<K, V> {
@Test
@SuppressWarnings("unchecked")
public void testConsumerContextConfiguration() {
final KafkaConsumerContext<K, V> consumerContext = appContext.getBean("consumerContext",
KafkaConsumerContext.class);
final org.springframework.integration.kafka.support.KafkaConsumerContext<K, V> consumerContext =
appContext.getBean("consumerContext",
org.springframework.integration.kafka.support.KafkaConsumerContext.class);
Assert.assertNotNull(consumerContext);
ConsumerConfiguration<K, V> cc = consumerContext.getConsumerConfiguration("default1");
ConsumerMetadata<K, V> cm = cc.getConsumerMetadata();
org.springframework.integration.kafka.support.ConsumerConfiguration<K, V> cc
= consumerContext.getConsumerConfiguration("default1");
org.springframework.integration.kafka.support.ConsumerMetadata<K, V> cm = cc.getConsumerMetadata();
assertNotNull(cm);
TopicFilterConfiguration topicFilterConfiguration = cm.getTopicFilterConfiguration();
org.springframework.integration.kafka.support.TopicFilterConfiguration topicFilterConfiguration =
cm.getTopicFilterConfiguration();
assertEquals("foo : 10", topicFilterConfiguration.toString());
assertThat(topicFilterConfiguration.getTopicFilter(), Matchers.instanceOf(Blacklist.class));
}

2
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaInboundAdapterParserTests.java
View File

@@ -34,6 +34,8 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
*/
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration
@Deprecated
@SuppressWarnings("deprecation")
public class KafkaInboundAdapterParserTests {
@ClassRule

39
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaMultiConsumerContextParserTests.java
View File

@@ -19,13 +19,9 @@ import org.junit.Assert;
import org.junit.ClassRule;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.integration.kafka.rule.KafkaRunning;
import org.springframework.integration.kafka.support.ConsumerConfiguration;
import org.springframework.integration.kafka.support.ConsumerMetadata;
import org.springframework.integration.kafka.support.KafkaConsumerContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@@ -35,6 +31,8 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
*/
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration
@Deprecated
@SuppressWarnings("deprecation")
public class KafkaMultiConsumerContextParserTests<K,V> {
@ClassRule
@@ -46,30 +44,43 @@ public class KafkaMultiConsumerContextParserTests<K,V> {
@SuppressWarnings("unchecked")
@Test
public void testMultiConsumerContexts() {
final KafkaConsumerContext<K,V> consumerContext1 = appContext.getBean("consumerContext1", KafkaConsumerContext.class);
final org.springframework.integration.kafka.support.KafkaConsumerContext<K,V> consumerContext1 =
appContext.getBean("consumerContext1",
org.springframework.integration.kafka.support.KafkaConsumerContext.class);
Assert.assertNotNull(consumerContext1);
final KafkaConsumerContext<K,V> consumerContext2 = appContext.getBean("consumerContext2", KafkaConsumerContext.class);
final org.springframework.integration.kafka.support.KafkaConsumerContext<K,V> consumerContext2 =
appContext.getBean("consumerContext2",
org.springframework.integration.kafka.support.KafkaConsumerContext.class);
Assert.assertNotNull(consumerContext2);
}
@SuppressWarnings("unchecked")
@Test
public void testConsumerContextConfigurations() {
final KafkaConsumerContext<K,V> consumerContext = appContext.getBean("consumerContext1", KafkaConsumerContext.class);
final org.springframework.integration.kafka.support.KafkaConsumerContext<K,V> consumerContext =
appContext.getBean("consumerContext1",
org.springframework.integration.kafka.support.KafkaConsumerContext.class);
Assert.assertNotNull(consumerContext);
final ConsumerConfiguration<K,V> cc = consumerContext.getConsumerConfiguration("default1");
final ConsumerMetadata<K,V> cm = cc.getConsumerMetadata();
final org.springframework.integration.kafka.support.ConsumerConfiguration<K,V> cc =
consumerContext.getConsumerConfiguration("default1");
final org.springframework.integration.kafka.support.ConsumerMetadata<K,V> cm = cc.getConsumerMetadata();
Assert.assertTrue(cm.getTopicStreamMap().get("test1") == 3);
Assert.assertTrue(cm.getTopicStreamMap().get("test2") == 4);
Assert.assertNotNull(cm);
final ConsumerConfiguration<K,V> cc2 = consumerContext.getConsumerConfiguration("default2");
final ConsumerMetadata<K,V> cm2 = cc2.getConsumerMetadata();
final org.springframework.integration.kafka.support.ConsumerConfiguration<K,V> cc2 =
consumerContext.getConsumerConfiguration("default2");
final org.springframework.integration.kafka.support.ConsumerMetadata<K,V> cm2 =
cc2.getConsumerMetadata();
Assert.assertTrue(cm2.getTopicStreamMap().get("test3") == 1);
Assert.assertNotNull(cm2);
final KafkaConsumerContext<K,V> consumerContext2 = appContext.getBean("consumerContext2", KafkaConsumerContext.class);
final org.springframework.integration.kafka.support.KafkaConsumerContext<K,V> consumerContext2 =
appContext.getBean("consumerContext2",
org.springframework.integration.kafka.support.KafkaConsumerContext.class);
Assert.assertNotNull(consumerContext2);
final ConsumerConfiguration<K,V> otherCC = consumerContext2.getConsumerConfiguration("default1");
final ConsumerMetadata<K,V> otherCM = otherCC.getConsumerMetadata();
final org.springframework.integration.kafka.support.ConsumerConfiguration<K,V> otherCC =
consumerContext2.getConsumerConfiguration("default1");
final org.springframework.integration.kafka.support.ConsumerMetadata<K,V> otherCM =
otherCC.getConsumerMetadata();
Assert.assertTrue(otherCM.getTopicStreamMap().get("test4") == 3);
}
}

25
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/support/ConsumerConfigurationTests.java
View File

@@ -16,9 +16,23 @@
package org.springframework.integration.kafka.support;
import static org.junit.Assert.assertNull;
import static org.mockito.Mockito.*;
import static org.mockito.Mockito.atLeast;
import static org.mockito.Mockito.atMost;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import java.util.*;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.junit.Assert;
import org.junit.Test;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;
@@ -26,16 +40,13 @@ import kafka.javaapi.consumer.ConsumerConnector;
import kafka.message.MessageAndMetadata;
import kafka.serializer.Decoder;
import org.junit.Assert;
import org.junit.Test;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
/**
* @author Soby Chacko
* @author Rajasekar Elango
* @since 0.5
*/
@Deprecated
@SuppressWarnings("deprecation")
public class ConsumerConfigurationTests<K,V> {
@Test

2
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/support/KafkaConsumerContextIntegrationTests.java
View File

@@ -49,6 +49,8 @@ import kafka.consumer.ConsumerConfig;
* @author Gary Russell
* @since 1.2.2
*/
@Deprecated
@SuppressWarnings("deprecation")
public class KafkaConsumerContextIntegrationTests {
private static final String TOPIC = "springIntegrationTestInbound";

2
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/support/KafkaConsumerContextTest.java
View File

@@ -32,6 +32,8 @@ import org.springframework.messaging.Message;
* @author Soby Chacko
* @since 0.5
*/
@Deprecated
@SuppressWarnings("deprecation")
public class KafkaConsumerContextTest<K, V> {
@Test