Skip to content

Commit

Permalink
Merged from john's dataProvider branch.
Browse files Browse the repository at this point in the history
  • Loading branch information
@wonlay
wonlay committed May 13, 2011
2 parents c182a9d + 461786d commit f628831
Show file tree
Hide file tree
Showing 16 changed files with 455 additions and 143 deletions.
149 changes: 74 additions & 75 deletions .classpath
View file
Original file line number Diff line number Diff line change
@@ -1,76 +1,75 @@
<?xml version="1.0" encoding="UTF-8"?>
<classpath>
<classpathentry including="**/*.java" kind="src" output="target/test-classes" path="src/test/java"/>
<classpathentry including="**/*.java" kind="src" path="src/main/java"/>
<classpathentry kind="var" path="M2_REPO/javax/servlet/servlet-api/2.5/servlet-api-2.5.jar"/>
<classpathentry kind="var" path="M2_REPO/org/apache/ant/ant/1.7.1/ant-1.7.1.jar"/>
<classpathentry kind="var" path="M2_REPO/org/apache/ant/ant-launcher/1.7.1/ant-launcher-1.7.1.jar"/>
<classpathentry kind="var" path="M2_REPO/aopalliance/aopalliance/1.0/aopalliance-1.0.jar"/>
<classpathentry kind="var" path="M2_REPO/asm/asm/3.2/asm-3.2.jar"/>
<classpathentry kind="var" path="M2_REPO/org/apache/avro/avro/1.4.0/avro-1.4.0.jar"/>
<classpathentry kind="var" path="M2_REPO/com/browseengine/bobo/bobo-browse/2.5.2-SNAPSHOT/bobo-browse-2.5.2-SNAPSHOT.jar"/>
<classpathentry kind="var" path="M2_REPO/cglib/cglib-nodep/2.2/cglib-nodep-2.2.jar"/>
<classpathentry kind="var" path="M2_REPO/commons-beanutils/commons-beanutils/1.7.0/commons-beanutils-1.7.0.jar"/>
<classpathentry kind="var" path="M2_REPO/commons-beanutils/commons-beanutils-core/1.8.0/commons-beanutils-core-1.8.0.jar"/>
<classpathentry kind="var" path="M2_REPO/commons-cli/commons-cli/1.2/commons-cli-1.2.jar"/>
<classpathentry kind="var" path="M2_REPO/commons-codec/commons-codec/1.4/commons-codec-1.4.jar"/>
<classpathentry kind="var" path="M2_REPO/commons-collections/commons-collections/3.2.1/commons-collections-3.2.1.jar"/>
<classpathentry kind="var" path="M2_REPO/commons-configuration/commons-configuration/1.6/commons-configuration-1.6.jar" sourcepath="M2_REPO/commons-configuration/commons-configuration/1.6/commons-configuration-1.6-sources.jar"/>
<classpathentry kind="var" path="M2_REPO/commons-digester/commons-digester/1.8/commons-digester-1.8.jar"/>
<classpathentry kind="var" path="M2_REPO/commons-io/commons-io/2.0.1/commons-io-2.0.1.jar"/>
<classpathentry kind="var" path="M2_REPO/commons-lang/commons-lang/2.5/commons-lang-2.5.jar"/>
<classpathentry kind="var" path="M2_REPO/commons-logging/commons-logging/1.1.1/commons-logging-1.1.1.jar"/>
<classpathentry kind="var" path="M2_REPO/org/easymock/easymock/3.0/easymock-3.0.jar"/>
<classpathentry kind="var" path="M2_REPO/fastutil/fastutil/5.0.5/fastutil-5.0.5.jar"/>
<classpathentry kind="var" path="M2_REPO/com/google/inject/guice/2.0/guice-2.0.jar"/>
<classpathentry kind="var" path="M2_REPO/org/apache/httpcomponents/httpclient/4.1/httpclient-4.1.jar"/>
<classpathentry kind="var" path="M2_REPO/org/apache/httpcomponents/httpcore/4.1/httpcore-4.1.jar"/>
<classpathentry kind="var" path="M2_REPO/org/codehaus/jackson/jackson-core-asl/1.4.2/jackson-core-asl-1.4.2.jar"/>
<classpathentry kind="var" path="M2_REPO/org/codehaus/jackson/jackson-mapper-asl/1.7.3/jackson-mapper-asl-1.7.3.jar"/>
<classpathentry kind="var" path="M2_REPO/org/mortbay/jetty/jetty/6.1.26/jetty-6.1.26.jar"/>
<classpathentry kind="var" path="M2_REPO/org/eclipse/jetty/jetty-continuation/7.3.1.v20110307/jetty-continuation-7.3.1.v20110307.jar"/>
<classpathentry kind="var" path="M2_REPO/org/eclipse/jetty/jetty-http/7.3.1.v20110307/jetty-http-7.3.1.v20110307.jar"/>
<classpathentry kind="var" path="M2_REPO/org/eclipse/jetty/jetty-io/7.3.1.v20110307/jetty-io-7.3.1.v20110307.jar"/>
<classpathentry kind="var" path="M2_REPO/org/eclipse/jetty/jetty-server/7.3.1.v20110307/jetty-server-7.3.1.v20110307.jar"/>
<classpathentry kind="var" path="M2_REPO/org/eclipse/jetty/jetty-util/7.3.1.v20110307/jetty-util-7.3.1.v20110307.jar"/>
<classpathentry kind="var" path="M2_REPO/org/mortbay/jetty/jetty-util/6.1.26/jetty-util-6.1.26.jar"/>
<classpathentry kind="var" path="M2_REPO/jline/jline/0.9.94/jline-0.9.94.jar"/>
<classpathentry kind="var" path="M2_REPO/net/sf/jopt-simple/jopt-simple/3.2/jopt-simple-3.2.jar"/>
<classpathentry kind="var" path="M2_REPO/org/json/json/20080701/json-20080701.jar"/>
<classpathentry kind="var" path="M2_REPO/junit/junit/4.5/junit-4.5.jar" sourcepath="M2_REPO/junit/junit/4.5/junit-4.5-sources.jar"/>
<classpathentry kind="var" path="M2_REPO/com/sna-projects/kafka/kafka/0.5-SNAPSHOT/kafka-0.5-SNAPSHOT.jar"/>
<classpathentry kind="var" path="M2_REPO/com/linkedin/kamikaze/kamikaze/3.0.4/kamikaze-3.0.4.jar"/>
<classpathentry kind="var" path="M2_REPO/com/sna-projects/kamikaze/kamikaze/3.0.3/kamikaze-3.0.3.jar"/>
<classpathentry kind="var" path="M2_REPO/log4j/log4j/1.2.14/log4j-1.2.14.jar"/>
<classpathentry kind="var" path="M2_REPO/org/apache/lucene/lucene-core/3.0.2/lucene-core-3.0.2.jar"/>
<classpathentry kind="var" path="M2_REPO/org/jboss/netty/netty/3.2.3.Final/netty-3.2.3.Final.jar"/>
<classpathentry kind="var" path="M2_REPO/com/sna-projects/norbert/norbert-cluster/1.0.0-SNAPSHOT/norbert-cluster-1.0.0-SNAPSHOT.jar"/>
<classpathentry kind="var" path="M2_REPO/com/sna-projects/norbert/norbert-java-cluster/1.0.0-SNAPSHOT/norbert-java-cluster-1.0.0-SNAPSHOT.jar"/>
<classpathentry kind="var" path="M2_REPO/com/sna-projects/norbert/norbert-java-network/1.0.0-SNAPSHOT/norbert-java-network-1.0.0-SNAPSHOT.jar"/>
<classpathentry kind="var" path="M2_REPO/com/sna-projects/norbert/norbert-network/1.0.0-SNAPSHOT/norbert-network-1.0.0-SNAPSHOT.jar"/>
<classpathentry kind="var" path="M2_REPO/org/objenesis/objenesis/1.2/objenesis-1.2.jar"/>
<classpathentry kind="var" path="M2_REPO/oro/oro/2.0.8/oro-2.0.8.jar"/>
<classpathentry kind="var" path="M2_REPO/com/thoughtworks/paranamer/paranamer/2.2/paranamer-2.2.jar"/>
<classpathentry kind="var" path="M2_REPO/com/thoughtworks/paranamer/paranamer-ant/2.2/paranamer-ant-2.2.jar"/>
<classpathentry kind="var" path="M2_REPO/com/thoughtworks/paranamer/paranamer-generator/2.2/paranamer-generator-2.2.jar"/>
<classpathentry kind="var" path="M2_REPO/com/google/protobuf/protobuf-java/2.3.0/protobuf-java-2.3.0.jar"/>
<classpathentry kind="var" path="M2_REPO/com/thoughtworks/qdox/qdox/1.10.1/qdox-1.10.1.jar"/>
<classpathentry kind="var" path="M2_REPO/org/scala-lang/scala-library/2.8.0/scala-library-2.8.0.jar"/>
<classpathentry kind="var" path="M2_REPO/org/scalatest/scalatest/1.3/scalatest-1.3.jar"/>
<classpathentry kind="var" path="M2_REPO/org/mortbay/jetty/servlet-api/2.5-20081211/servlet-api-2.5-20081211.jar"/>
<classpathentry kind="var" path="M2_REPO/org/slf4j/slf4j-api/1.6.1/slf4j-api-1.6.1.jar"/>
<classpathentry kind="var" path="M2_REPO/org/slf4j/slf4j-log4j12/1.6.1/slf4j-log4j12-1.6.1.jar"/>
<classpathentry kind="var" path="M2_REPO/org/springframework/spring/2.5.6/spring-2.5.6.jar" sourcepath="M2_REPO/org/springframework/spring/2.5.6/spring-2.5.6-sources.jar"/>
<classpathentry kind="var" path="M2_REPO/org/springframework/spring-beans/2.5.6/spring-beans-2.5.6.jar"/>
<classpathentry kind="var" path="M2_REPO/org/springframework/spring-context/2.5.6/spring-context-2.5.6.jar"/>
<classpathentry kind="var" path="M2_REPO/org/springframework/spring-core/2.5.6/spring-core-2.5.6.jar"/>
<classpathentry kind="var" path="M2_REPO/org/apache/velocity/velocity/1.6.4/velocity-1.6.4.jar"/>
<classpathentry kind="var" path="M2_REPO/com/linkedin/zoie/zoie-core/2.6.0-SNAPSHOT/zoie-core-2.6.0-SNAPSHOT.jar" sourcepath="M2_REPO/com/linkedin/zoie/zoie-core/2.6.0-SNAPSHOT/zoie-core-2.6.0-SNAPSHOT-sources.jar">
<attributes>
<attribute name="javadoc_location" value="jar:file:/Users/jwang/.m2/repository/com/linkedin/zoie/zoie-core/2.6.0-SNAPSHOT/zoie-core-2.6.0-SNAPSHOT-javadoc.jar!/"/>
</attributes>
</classpathentry>
<classpathentry kind="var" path="M2_REPO/org/apache/zookeeper/zookeeper/3.3.3/zookeeper-3.3.3.jar"/>
<classpathentry kind="con" path="org.eclipse.jdt.launching.JRE_CONTAINER"/>
<classpathentry kind="output" path="target/classes"/>
</classpath>
<classpathentry kind="src" path="src/test/java" output="target/test-classes" including="**/*.java"/>
<classpathentry kind="src" path="src/main/java" including="**/*.java"/>
<classpathentry kind="output" path="target/classes"/>
<classpathentry kind="var" path="M2_REPO/javax/jms/jms/1.1/jms-1.1.jar"/>
<classpathentry kind="var" path="M2_REPO/javax/servlet/servlet-api/2.5/servlet-api-2.5.jar"/>
<classpathentry kind="var" path="M2_REPO/org/apache/ant/ant/1.7.1/ant-1.7.1.jar"/>
<classpathentry kind="var" path="M2_REPO/org/apache/ant/ant-launcher/1.7.1/ant-launcher-1.7.1.jar"/>
<classpathentry kind="var" path="M2_REPO/aopalliance/aopalliance/1.0/aopalliance-1.0.jar"/>
<classpathentry kind="var" path="M2_REPO/asm/asm/3.2/asm-3.2.jar"/>
<classpathentry kind="var" path="M2_REPO/org/apache/avro/avro/1.4.0/avro-1.4.0.jar"/>
<classpathentry kind="var" path="M2_REPO/com/browseengine/bobo/bobo-browse/2.5.2-SNAPSHOT/bobo-browse-2.5.2-SNAPSHOT.jar"/>
<classpathentry kind="var" path="M2_REPO/cglib/cglib-nodep/2.2/cglib-nodep-2.2.jar"/>
<classpathentry kind="var" path="M2_REPO/commons-beanutils/commons-beanutils/1.7.0/commons-beanutils-1.7.0.jar"/>
<classpathentry kind="var" path="M2_REPO/commons-beanutils/commons-beanutils-core/1.8.0/commons-beanutils-core-1.8.0.jar"/>
<classpathentry kind="var" path="M2_REPO/commons-cli/commons-cli/1.2/commons-cli-1.2.jar"/>
<classpathentry kind="var" path="M2_REPO/commons-codec/commons-codec/1.4/commons-codec-1.4.jar"/>
<classpathentry kind="var" path="M2_REPO/commons-collections/commons-collections/3.2.1/commons-collections-3.2.1.jar"/>
<classpathentry kind="var" path="M2_REPO/commons-configuration/commons-configuration/1.6/commons-configuration-1.6.jar" sourcepath="M2_REPO/commons-configuration/commons-configuration/1.6/commons-configuration-1.6-sources.jar"/>
<classpathentry kind="var" path="M2_REPO/commons-digester/commons-digester/1.8/commons-digester-1.8.jar"/>
<classpathentry kind="var" path="M2_REPO/commons-io/commons-io/2.0.1/commons-io-2.0.1.jar"/>
<classpathentry kind="var" path="M2_REPO/commons-lang/commons-lang/2.5/commons-lang-2.5.jar"/>
<classpathentry kind="var" path="M2_REPO/commons-logging/commons-logging/1.1.1/commons-logging-1.1.1.jar"/>
<classpathentry kind="var" path="M2_REPO/org/easymock/easymock/3.0/easymock-3.0.jar"/>
<classpathentry kind="var" path="M2_REPO/fastutil/fastutil/5.0.5/fastutil-5.0.5.jar"/>
<classpathentry kind="var" path="M2_REPO/org/apache/httpcomponents/httpclient/4.1/httpclient-4.1.jar"/>
<classpathentry kind="var" path="M2_REPO/org/apache/httpcomponents/httpcore/4.1/httpcore-4.1.jar"/>
<classpathentry kind="var" path="M2_REPO/org/codehaus/jackson/jackson-core-asl/1.4.2/jackson-core-asl-1.4.2.jar"/>
<classpathentry kind="var" path="M2_REPO/org/codehaus/jackson/jackson-mapper-asl/1.4.2/jackson-mapper-asl-1.4.2.jar"/>
<classpathentry kind="var" path="M2_REPO/org/mortbay/jetty/jetty/6.1.26/jetty-6.1.26.jar"/>
<classpathentry kind="var" path="M2_REPO/org/mortbay/jetty/jetty-util/6.1.26/jetty-util-6.1.26.jar"/>
<classpathentry kind="var" path="M2_REPO/jline/jline/0.9.94/jline-0.9.94.jar"/>
<classpathentry kind="var" path="M2_REPO/net/sf/jopt-simple/jopt-simple/3.2/jopt-simple-3.2.jar"/>
<classpathentry kind="var" path="M2_REPO/org/json/json/20080701/json-20080701.jar"/>
<classpathentry kind="var" path="M2_REPO/junit/junit/4.5/junit-4.5.jar" sourcepath="M2_REPO/junit/junit/4.5/junit-4.5-sources.jar"/>
<classpathentry kind="var" path="M2_REPO/com/sna-projects/kafka/kafka/0.5-SNAPSHOT/kafka-0.5-SNAPSHOT.jar"/>
<classpathentry kind="var" path="M2_REPO/com/linkedin/kamikaze/kamikaze/3.0.4/kamikaze-3.0.4.jar"/>
<classpathentry kind="var" path="M2_REPO/com/sna-projects/kamikaze/kamikaze/3.0.3/kamikaze-3.0.3.jar"/>
<classpathentry kind="var" path="M2_REPO/log4j/log4j/1.2.14/log4j-1.2.14.jar"/>
<classpathentry kind="var" path="M2_REPO/org/apache/lucene/lucene-core/3.0.2/lucene-core-3.0.2.jar"/>
<classpathentry kind="var" path="M2_REPO/org/jboss/netty/netty/3.2.3.Final/netty-3.2.3.Final.jar"/>
<classpathentry kind="var" path="M2_REPO/com/sna-projects/norbert/norbert-cluster/1.0.0-SNAPSHOT/norbert-cluster-1.0.0-SNAPSHOT.jar"/>
<classpathentry kind="var" path="M2_REPO/com/sna-projects/norbert/norbert-java-cluster/1.0.0-SNAPSHOT/norbert-java-cluster-1.0.0-SNAPSHOT.jar"/>
<classpathentry kind="var" path="M2_REPO/com/sna-projects/norbert/norbert-java-network/1.0.0-SNAPSHOT/norbert-java-network-1.0.0-SNAPSHOT.jar"/>
<classpathentry kind="var" path="M2_REPO/com/sna-projects/norbert/norbert-network/1.0.0-SNAPSHOT/norbert-network-1.0.0-SNAPSHOT.jar"/>
<classpathentry kind="var" path="M2_REPO/org/objenesis/objenesis/1.2/objenesis-1.2.jar"/>
<classpathentry kind="var" path="M2_REPO/oro/oro/2.0.8/oro-2.0.8.jar"/>
<classpathentry kind="var" path="M2_REPO/com/thoughtworks/paranamer/paranamer/2.2/paranamer-2.2.jar"/>
<classpathentry kind="var" path="M2_REPO/com/thoughtworks/paranamer/paranamer-ant/2.2/paranamer-ant-2.2.jar"/>
<classpathentry kind="var" path="M2_REPO/com/thoughtworks/paranamer/paranamer-generator/2.2/paranamer-generator-2.2.jar"/>
<classpathentry kind="var" path="M2_REPO/com/google/protobuf/protobuf-java/2.3.0/protobuf-java-2.3.0.jar"/>
<classpathentry kind="var" path="M2_REPO/com/thoughtworks/qdox/qdox/1.10.1/qdox-1.10.1.jar"/>
<classpathentry kind="var" path="M2_REPO/org/scala-lang/scala-library/2.8.0/scala-library-2.8.0.jar"/>
<classpathentry kind="var" path="M2_REPO/org/scalatest/scalatest/1.3/scalatest-1.3.jar"/>
<classpathentry kind="var" path="M2_REPO/org/mortbay/jetty/servlet-api/2.5-20081211/servlet-api-2.5-20081211.jar"/>
<classpathentry kind="var" path="M2_REPO/org/slf4j/slf4j-api/1.6.1/slf4j-api-1.6.1.jar"/>
<classpathentry kind="var" path="M2_REPO/org/slf4j/slf4j-log4j12/1.6.1/slf4j-log4j12-1.6.1.jar"/>
<classpathentry kind="var" path="M2_REPO/org/springframework/spring/2.5.6/spring-2.5.6.jar" sourcepath="M2_REPO/org/springframework/spring/2.5.6/spring-2.5.6-sources.jar"/>
<classpathentry kind="var" path="M2_REPO/org/springframework/spring-beans/2.5.6/spring-beans-2.5.6.jar"/>
<classpathentry kind="var" path="M2_REPO/org/springframework/spring-context/2.5.6/spring-context-2.5.6.jar"/>
<classpathentry kind="var" path="M2_REPO/org/springframework/spring-core/2.5.6/spring-core-2.5.6.jar"/>
<classpathentry kind="var" path="M2_REPO/org/apache/velocity/velocity/1.6.4/velocity-1.6.4.jar"/>
<classpathentry kind="var" path="M2_REPO/com/linkedin/zoie/zoie-core/2.6.0-SNAPSHOT/zoie-core-2.6.0-SNAPSHOT.jar" sourcepath="M2_REPO/com/linkedin/zoie/zoie-core/2.6.0-SNAPSHOT/zoie-core-2.6.0-SNAPSHOT-sources.jar">
<attributes>
<attribute value="jar:file:/Users/jwang/.m2/repository/com/linkedin/zoie/zoie-core/2.6.0-SNAPSHOT/zoie-core-2.6.0-SNAPSHOT-javadoc.jar!/" name="javadoc_location"/>
</attributes>
</classpathentry>
<classpathentry kind="var" path="M2_REPO/com/linkedin/zoie/zoie-jms/2.6.0-SNAPSHOT/zoie-jms-2.6.0-SNAPSHOT.jar" sourcepath="M2_REPO/com/linkedin/zoie/zoie-jms/2.6.0-SNAPSHOT/zoie-jms-2.6.0-SNAPSHOT-sources.jar">
<attributes>
<attribute value="jar:file:/Users/jwang/.m2/repository/com/linkedin/zoie/zoie-jms/2.6.0-SNAPSHOT/zoie-jms-2.6.0-SNAPSHOT-javadoc.jar!/" name="javadoc_location"/>
</attributes>
</classpathentry>
<classpathentry kind="var" path="M2_REPO/org/apache/zookeeper/zookeeper/3.3.3/zookeeper-3.3.3.jar"/>
<classpathentry kind="con" path="org.eclipse.jdt.launching.JRE_CONTAINER"/>
</classpath>
2 changes: 1 addition & 1 deletion conf/schema.xml
View file
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,7 @@
lucene - use store in lucene's index
-->

<table uid="id" shard-by="id" src-data-store="none" src-data-field="src_data" compress-src-data="true" delete-field="" skip-field="">
<table uid="id" src-data-store="none" src-data-field="src_data" compress-src-data="true" delete-field="" skip-field="">
<column name="color" type="string" />
<column name="category" type="string" />
<column name="city" type="string" />
Expand Down
7 changes: 7 additions & 0 deletions pom.xml
View file
Original file line number Diff line number Diff line change
Expand Up @@ -141,6 +141,13 @@
<artifactId>zoie-core</artifactId>
<version>2.6.0-SNAPSHOT</version>
</dependency>

<dependency>
<groupId>com.linkedin.zoie</groupId>
<artifactId>zoie-jms</artifactId>
<version>2.6.0-SNAPSHOT</version>
</dependency>

<dependency>
<groupId>fastutil</groupId>
<artifactId>fastutil</artifactId>
Expand Down
7 changes: 0 additions & 7 deletions src/main/java/com/sensei/conf/SenseiSchema.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -27,7 +27,6 @@ public class SenseiSchema {
private static Logger logger = Logger.getLogger(SenseiSchema.class);

private String _uidField;
private String _shardByField;
private String _deleteField;
private String _skipField;
private String _srcDataStore;
Expand All @@ -52,9 +51,6 @@ public String getUidField(){
return _uidField;
}

public String getShardByField(){
return _shardByField;
}

public String getDeleteField(){
return _deleteField;
Expand Down Expand Up @@ -95,9 +91,6 @@ public static SenseiSchema build(Document schemaDoc) throws ConfigurationExcepti

Element tableElem = (Element) tables.item(0);
schema._uidField = tableElem.getAttribute("uid");
schema._shardByField = tableElem.getAttribute("shard-by");
if (schema._shardByField==null || schema._shardByField.length()==0)
schema._shardByField = schema._uidField;
schema._deleteField = tableElem.getAttribute("delete-field");
if (schema._deleteField==null) schema._deleteField="";
schema._skipField = tableElem.getAttribute("skip-field");
Expand Down
7 changes: 0 additions & 7 deletions src/main/java/com/sensei/conf/SenseiServerBuilder.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -397,13 +397,6 @@ else if (SENSEI_HOURGLASS_FREQUENCY_DAY.equals(frequencyString)){
if (idxMgrType.length()==0){
String uidField = _senseiSchema.getUidField();
indexingManager = new DefaultStreamingIndexingManager(_senseiSchema,_senseiConf, _pluginContext, _versionComparator);
String filterName = _senseiConf.getString(SENSEI_INDEX_MANAGER_FILTER, "");
if (filterName.length()>0){
DataSourceFilter filter = (DataSourceFilter)_pluginContext.getBean(filterName);
filter.setSrcDataStore(_senseiSchema.getSrcDataStore());
filter.setSrcDataField(_senseiSchema.getSrcDataField());
((DefaultStreamingIndexingManager)indexingManager).setFilter(filter);
}
}
else{
indexingManager = (SenseiIndexingManager)_pluginContext.getBean(idxMgrType);
Expand Down
4 changes: 2 additions & 2 deletions src/main/java/com/sensei/dataprovider/file/LinedJsonFileDataProvider.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,7 @@
import com.sensei.indexing.api.DataSourceFilter;
import com.sensei.indexing.api.DataSourceFilterable;

public class LinedJsonFileDataProvider extends LinedFileDataProvider<JSONObject> implements DataSourceFilterable {
public class LinedJsonFileDataProvider extends LinedFileDataProvider<JSONObject> implements DataSourceFilterable<String> {

private DataSourceFilter<String> _dataSourceFilter;

Expand All @@ -19,7 +19,7 @@ public LinedJsonFileDataProvider(Comparator<String> versionComparator, File file
}

@Override
public void setFilter(DataSourceFilter filter)
public void setFilter(DataSourceFilter<String> filter)
{
_dataSourceFilter = filter;
}
Expand Down
12 changes: 6 additions & 6 deletions src/main/java/com/sensei/dataprovider/kafka/KafkaJsonStreamDataProvider.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,7 @@
import com.sensei.indexing.api.DataSourceFilter;
import com.sensei.indexing.api.DataSourceFilterable;

public class KafkaJsonStreamDataProvider extends KafkaStreamDataProvider<JSONObject> implements DataSourceFilterable {
public class KafkaJsonStreamDataProvider extends KafkaStreamDataProvider<JSONObject> implements DataSourceFilterable<byte[]> {
private final static Charset UTF8 = Charset.forName("UTF-8");
public final static String KAFKA_MSG_OFFSET = "_KAFKA_MSG_OFFSET_";

Expand All @@ -22,11 +22,11 @@ public KafkaJsonStreamDataProvider(Comparator<String> versionComparator, String
super(versionComparator, kafkaHost, kafkaPort, soTimeout, batchSize, topic, startingOffset);
}

@Override
public void setFilter(DataSourceFilter filter)
{
_dataSourceFilter = filter;
}
@Override
public void setFilter(DataSourceFilter<byte[]> filter)
{
_dataSourceFilter = filter;
}

@Override
protected JSONObject convertMessageBytes(long msgStreamOffset, byte[] bytes,
Expand Down
Loading

0 comments on commit f628831

Please sign in to comment.