You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@usergrid.apache.org by to...@apache.org on 2015/04/23 18:24:38 UTC
[1/2] incubator-usergrid git commit: Updated RX java to 1.0.9 Updates
flush logic. Buffers of size > 1 are not getting flushed correctly. Still has
bugs
Repository: incubator-usergrid
Updated Branches:
refs/heads/two-dot-o-dev 014162335 -> 852c9747b
Updated RX java to 1.0.9
Updates flush logic. Buffers of size > 1 are not getting flushed correctly. Still has bugs
Project: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/commit/e91930ff
Tree: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/tree/e91930ff
Diff: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/diff/e91930ff
Branch: refs/heads/two-dot-o-dev
Commit: e91930ff0679ff3709490aacfdaec67e1ba83301
Parents: 34fd9cf
Author: Todd Nine <tn...@apigee.com>
Authored: Thu Apr 23 10:24:20 2015 -0600
Committer: Todd Nine <tn...@apigee.com>
Committed: Thu Apr 23 10:24:20 2015 -0600
----------------------------------------------------------------------
.../corepersistence/index/IndexServiceTest.java | 44 +++++---
stack/core/src/test/resources/log4j.properties | 1 +
stack/corepersistence/pom.xml | 2 +-
.../usergrid/persistence/index/IndexFig.java | 18 ++-
.../persistence/index/impl/BatchOperation.java | 41 +++++++
.../persistence/index/impl/BatchRequest.java | 41 -------
.../index/impl/DeIndexOperation.java | 104 +++++++++++++++++
.../persistence/index/impl/DeIndexRequest.java | 104 -----------------
.../index/impl/EsEntityIndexBatchImpl.java | 4 +-
.../index/impl/EsIndexBufferConsumerImpl.java | 107 ++++++++++--------
.../persistence/index/impl/IndexOperation.java | 107 ++++++++++++++++++
.../index/impl/IndexOperationMessage.java | 19 ++--
.../index/impl/IndexRefreshCommandImpl.java | 6 +-
.../persistence/index/impl/IndexRequest.java | 111 -------------------
14 files changed, 371 insertions(+), 338 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/e91930ff/stack/core/src/test/java/org/apache/usergrid/corepersistence/index/IndexServiceTest.java
----------------------------------------------------------------------
diff --git a/stack/core/src/test/java/org/apache/usergrid/corepersistence/index/IndexServiceTest.java b/stack/core/src/test/java/org/apache/usergrid/corepersistence/index/IndexServiceTest.java
index 8be93f6..92eb2e0 100644
--- a/stack/core/src/test/java/org/apache/usergrid/corepersistence/index/IndexServiceTest.java
+++ b/stack/core/src/test/java/org/apache/usergrid/corepersistence/index/IndexServiceTest.java
@@ -20,9 +20,9 @@
package org.apache.usergrid.corepersistence.index;
+import java.util.List;
import java.util.Set;
import java.util.UUID;
-import java.util.concurrent.CountDownLatch;
import org.junit.Before;
import org.junit.Test;
@@ -46,7 +46,7 @@ import org.apache.usergrid.persistence.index.SearchEdge;
import org.apache.usergrid.persistence.index.SearchTypes;
import org.apache.usergrid.persistence.index.impl.EsRunner;
import org.apache.usergrid.persistence.index.impl.IndexOperationMessage;
-import org.apache.usergrid.persistence.index.impl.IndexRequest;
+import org.apache.usergrid.persistence.index.impl.IndexOperation;
import org.apache.usergrid.persistence.model.entity.Entity;
import org.apache.usergrid.persistence.model.entity.Id;
import org.apache.usergrid.persistence.model.entity.SimpleId;
@@ -117,10 +117,10 @@ public class IndexServiceTest {
//real users should never call to blocking, we're not sure what we'll get
final IndexOperationMessage results = indexed.toBlocking().last();
- final Set<IndexRequest> indexRequests = results.getIndexRequests();
+ final Set<IndexOperation> indexRequests = results.getIndexRequests();
//ensure our value made it to the index request
- final IndexRequest indexRequest = indexRequests.iterator().next();
+ final IndexOperation indexRequest = indexRequests.iterator().next();
assertNotNull( indexRequest );
}
@@ -175,9 +175,6 @@ public class IndexServiceTest {
final SearchEdge collectionSearchEdge = CpNamingUtils.createSearchEdgeFromSource( collectionEdge );
- final CountDownLatch latch = new CountDownLatch( 1 );
-
-
//query until it's available
final CandidateResults collectionResults = getResults( applicationEntityIndex, collectionSearchEdge,
SearchTypes.fromTypes( testEntity.getId().getType() ), "select *", 100, 1, 100 );
@@ -186,7 +183,6 @@ public class IndexServiceTest {
assertEquals( testEntity.getId(), collectionResults.get( 0 ).getId() );
- latch.await();
final SearchEdge connectionSearchEdge = CpNamingUtils.createSearchEdgeFromSource( connectionSearch );
@@ -235,14 +231,23 @@ public class IndexServiceTest {
* Write 10k edges 10 at a time in parallel
*/
- final int edgeCount = 2000;
+// final int edgeCount = indexFig.getIndexBatchSize()*2;
+ final int edgeCount = 2;
- final Edge connectionSearch = Observable.range( 0, edgeCount ).flatMap( integer -> {
+ final List<Edge> connectionSearchEdges = Observable.range( 0, edgeCount ).flatMap( integer -> {
final Id connectingId = createId( "connecting" );
final Edge edge = CpNamingUtils.createConnectionEdge( connectingId, "likes", testEntity.getId() );
return graphManager.writeEdge( edge ).subscribeOn( Schedulers.io() );
- }, 10 ).toBlocking().last();
+ }, 20).toList().toBlocking().last();
+
+
+ assertEquals( "All edges saved", edgeCount, connectionSearchEdges.size() );
+
+ //get the first and last edge
+ final Edge connectionSearch = connectionSearchEdges.get( 0 );
+
+ final Edge lastSearch = connectionSearchEdges.get( edgeCount-1 );
//now index
@@ -258,8 +263,6 @@ public class IndexServiceTest {
final SearchEdge collectionSearchEdge = CpNamingUtils.createSearchEdgeFromSource( collectionEdge );
- final CountDownLatch latch = new CountDownLatch( 1 );
-
//query until it's available
final CandidateResults collectionResults = getResults( applicationEntityIndex, collectionSearchEdge,
@@ -269,7 +272,6 @@ public class IndexServiceTest {
assertEquals( testEntity.getId(), collectionResults.get( 0 ).getId() );
- latch.await();
final SearchEdge connectionSearchEdge = CpNamingUtils.createSearchEdgeFromSource( connectionSearch );
@@ -281,6 +283,18 @@ public class IndexServiceTest {
assertEquals( 1, connectionResults.size() );
assertEquals( testEntity.getId(), connectionResults.get( 0 ).getId() );
+
+
+ final SearchEdge lastConnectionSearchEdge = CpNamingUtils.createSearchEdgeFromSource( lastSearch );
+
+
+ //query until it's available
+ final CandidateResults lastConnectionResults = getResults( applicationEntityIndex, lastConnectionSearchEdge,
+ SearchTypes.fromTypes( testEntity.getId().getType() ), "select *", 100, 1, 100 );
+
+ assertEquals( 1, lastConnectionResults.size() );
+
+ assertEquals( testEntity.getId(), lastConnectionResults.get( 0 ).getId() );
}
@@ -291,7 +305,7 @@ public class IndexServiceTest {
for ( int i = 0; i < attempts; i++ ) {
final CandidateResults candidateResults =
- applicationEntityIndex.search( searchEdge, searchTypes, "select *", 100 );
+ applicationEntityIndex.search( searchEdge, searchTypes, ql , count );
if ( candidateResults.size() == expectedSize ) {
return candidateResults;
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/e91930ff/stack/core/src/test/resources/log4j.properties
----------------------------------------------------------------------
diff --git a/stack/core/src/test/resources/log4j.properties b/stack/core/src/test/resources/log4j.properties
index acab352..34bd708 100644
--- a/stack/core/src/test/resources/log4j.properties
+++ b/stack/core/src/test/resources/log4j.properties
@@ -64,4 +64,5 @@ log4j.logger.org.apache.usergrid.persistence.index.impl.EsApplicationEntityIndex
#log4j.logger.org.elasticsearch=DEBUG
#log4j.logger.org.apache.cassandra.service.StorageProxy=DEBUG, stdout
+log4j.logger.org.apache.usergrid.corepersistence.index.IndexServiceImpl=DEBUG
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/e91930ff/stack/corepersistence/pom.xml
----------------------------------------------------------------------
diff --git a/stack/corepersistence/pom.xml b/stack/corepersistence/pom.xml
index 8494996..8bd8613 100644
--- a/stack/corepersistence/pom.xml
+++ b/stack/corepersistence/pom.xml
@@ -71,7 +71,7 @@ limitations under the License.
<junit.version>4.11</junit.version>
<kryo-serializers.version>0.26</kryo-serializers.version>
<log4j.version>1.2.17</log4j.version>
- <rx.version>1.0.8</rx.version>
+ <rx.version>1.0.9</rx.version>
<slf4j.version>1.7.2</slf4j.version>
<surefire.version>2.16</surefire.version>
<aws.version>1.9.0</aws.version>
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/e91930ff/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/IndexFig.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/IndexFig.java b/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/IndexFig.java
index 47b25ae..eb59625 100644
--- a/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/IndexFig.java
+++ b/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/IndexFig.java
@@ -61,23 +61,27 @@ public interface IndexFig extends GuicyFig {
String INDEX_WRITE_CONSISTENCY_LEVEL = "elasticsearch.write_consistency_level";
/**
+ * The number of worker threads to flush collapsed batches
+ */
+ String INDEX_FLUSH_WORKER_COUNT = "index.flush.workers";
+
+ /**
* the number of times we can fail before we refresh the client
*/
String ELASTICSEARCH_FAIL_REFRESH = "elasticsearch.fail_refresh";
+ String QUERY_LIMIT_DEFAULT = "index.query.limit.default";
- String QUERY_LIMIT_DEFAULT = "index.query.limit.default";
-
/**
* Timeout calls to elasticsearch.
* @return
*/
- public static final String ELASTICSEARCH_QUERY_TIMEOUT = "elasticsearch.query.timeout";
+ String ELASTICSEARCH_QUERY_TIMEOUT = "elasticsearch.query.timeout";
- public static final String ELASTICSEARCH_WRITE_TIMEOUT= "elasticsearch.write.timeout";
+ String ELASTICSEARCH_WRITE_TIMEOUT= "elasticsearch.write.timeout";
/**
* The client type to use. Valid values are NODE or TRANSPORT
@@ -156,6 +160,10 @@ public interface IndexFig extends GuicyFig {
@Key( INDEX_BUFFER_SIZE )
int getIndexBufferSize();
+ @Default("10")
+ @Key(INDEX_FLUSH_WORKER_COUNT)
+ int getIndexFlushWorkerCount();
+
/**
@@ -184,7 +192,7 @@ public interface IndexFig extends GuicyFig {
@Key("elasticsearch.refresh_sleep_ms")
@Default("200")
long refreshSleep();
-
+
@Default( "5000" )
@Key( ELASTICSEARCH_QUERY_TIMEOUT )
long getQueryTimeout();
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/e91930ff/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/BatchOperation.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/BatchOperation.java b/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/BatchOperation.java
new file mode 100644
index 0000000..117532a
--- /dev/null
+++ b/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/BatchOperation.java
@@ -0,0 +1,41 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.usergrid.persistence.index.impl;
+
+
+import java.io.Serializable;
+
+import org.elasticsearch.action.bulk.BulkRequestBuilder;
+import org.elasticsearch.client.Client;
+
+
+/**
+ * A batch request we can serialize and construct on receive
+ */
+public interface BatchOperation extends Serializable {
+
+
+ /**
+ * Passing the client and the bulk request, add ourselves to the bulk request
+ * @param client
+ * @param bulkRequest
+ */
+ void doOperation( final Client client, final BulkRequestBuilder bulkRequest );
+}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/e91930ff/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/BatchRequest.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/BatchRequest.java b/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/BatchRequest.java
deleted file mode 100644
index df6c5b0..0000000
--- a/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/BatchRequest.java
+++ /dev/null
@@ -1,41 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-package org.apache.usergrid.persistence.index.impl;
-
-
-import java.io.Serializable;
-
-import org.elasticsearch.action.bulk.BulkRequestBuilder;
-import org.elasticsearch.client.Client;
-
-
-/**
- * A batch request we can serialize and construct on receive
- */
-public interface BatchRequest extends Serializable {
-
-
- /**
- * Passing the client and the bulk request, add ourselves to the bulk request
- * @param client
- * @param bulkRequest
- */
- public void doOperation(final Client client, final BulkRequestBuilder bulkRequest );
-}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/e91930ff/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/DeIndexOperation.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/DeIndexOperation.java b/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/DeIndexOperation.java
new file mode 100644
index 0000000..4f47749
--- /dev/null
+++ b/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/DeIndexOperation.java
@@ -0,0 +1,104 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.usergrid.persistence.index.impl;
+
+
+import java.util.Arrays;
+import java.util.UUID;
+
+import org.elasticsearch.action.bulk.BulkRequestBuilder;
+import org.elasticsearch.action.delete.DeleteRequestBuilder;
+import org.elasticsearch.client.Client;
+
+import org.apache.usergrid.persistence.core.scope.ApplicationScope;
+import org.apache.usergrid.persistence.index.SearchEdge;
+import org.apache.usergrid.persistence.model.entity.Id;
+
+import com.fasterxml.jackson.annotation.JsonTypeInfo;
+
+import static org.apache.usergrid.persistence.index.impl.IndexingUtils.createIndexDocId;
+
+
+/**
+ * Represent the properties required to build an index request
+ */
+@JsonTypeInfo( use = JsonTypeInfo.Id.CLASS, include = JsonTypeInfo.As.PROPERTY, property = "@class" )
+public class DeIndexOperation implements BatchOperation {
+
+ private String[] indexes;
+ private String documentId;
+
+
+ public DeIndexOperation() {
+
+ }
+
+
+ public DeIndexOperation( String[] indexes, ApplicationScope applicationScope, SearchEdge searchEdge, Id id,
+ UUID version ) {
+ this.indexes = indexes;
+ this.documentId = createIndexDocId( applicationScope, id, version, searchEdge );
+ }
+
+
+ @Override
+ public void doOperation( final Client client, final BulkRequestBuilder bulkRequest ) {
+
+
+ for ( final String index : indexes ) {
+ final DeleteRequestBuilder builder =
+ client.prepareDelete( index, IndexingUtils.ES_ENTITY_TYPE, documentId );
+ bulkRequest.add( builder );
+ }
+ }
+
+
+
+
+ @Override
+ public boolean equals( final Object o ) {
+ if ( this == o ) {
+ return true;
+ }
+ if ( o == null || getClass() != o.getClass() ) {
+ return false;
+ }
+
+ final DeIndexOperation that = ( DeIndexOperation ) o;
+
+ if ( !documentId.equals( that.documentId ) ) {
+ return false;
+ }
+
+ if ( !Arrays.equals( indexes, that.indexes ) ) {
+ return false;
+ }
+
+ return true;
+ }
+
+
+ @Override
+ public int hashCode() {
+ int result = Arrays.hashCode( indexes );
+ result = 31 * result + documentId.hashCode();
+ return result;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/e91930ff/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/DeIndexRequest.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/DeIndexRequest.java b/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/DeIndexRequest.java
deleted file mode 100644
index 3f0dc23..0000000
--- a/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/DeIndexRequest.java
+++ /dev/null
@@ -1,104 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-package org.apache.usergrid.persistence.index.impl;
-
-
-import java.util.Arrays;
-import java.util.UUID;
-
-import org.elasticsearch.action.bulk.BulkRequestBuilder;
-import org.elasticsearch.action.delete.DeleteRequestBuilder;
-import org.elasticsearch.client.Client;
-
-import org.apache.usergrid.persistence.core.scope.ApplicationScope;
-import org.apache.usergrid.persistence.index.SearchEdge;
-import org.apache.usergrid.persistence.model.entity.Id;
-
-import com.fasterxml.jackson.annotation.JsonTypeInfo;
-
-import static org.apache.usergrid.persistence.index.impl.IndexingUtils.createIndexDocId;
-
-
-/**
- * Represent the properties required to build an index request
- */
-@JsonTypeInfo( use = JsonTypeInfo.Id.CLASS, include = JsonTypeInfo.As.PROPERTY, property = "@class" )
-public class DeIndexRequest implements BatchRequest {
-
- private String[] indexes;
- private String documentId;
-
-
- public DeIndexRequest() {
-
- }
-
-
- public DeIndexRequest( String[] indexes, ApplicationScope applicationScope, SearchEdge searchEdge, Id id,
- UUID version ) {
- this.indexes = indexes;
- this.documentId = createIndexDocId( applicationScope, id, version, searchEdge );
- }
-
-
- @Override
- public void doOperation( final Client client, final BulkRequestBuilder bulkRequest ) {
-
-
- for ( final String index : indexes ) {
- final DeleteRequestBuilder builder =
- client.prepareDelete( index, IndexingUtils.ES_ENTITY_TYPE, documentId );
- bulkRequest.add( builder );
- }
- }
-
-
-
-
- @Override
- public boolean equals( final Object o ) {
- if ( this == o ) {
- return true;
- }
- if ( o == null || getClass() != o.getClass() ) {
- return false;
- }
-
- final DeIndexRequest that = ( DeIndexRequest ) o;
-
- if ( !documentId.equals( that.documentId ) ) {
- return false;
- }
-
- if ( !Arrays.equals( indexes, that.indexes ) ) {
- return false;
- }
-
- return true;
- }
-
-
- @Override
- public int hashCode() {
- int result = Arrays.hashCode( indexes );
- result = 31 * result + documentId.hashCode();
- return result;
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/e91930ff/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/EsEntityIndexBatchImpl.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/EsEntityIndexBatchImpl.java b/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/EsEntityIndexBatchImpl.java
index 32a0f02..66849ee 100644
--- a/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/EsEntityIndexBatchImpl.java
+++ b/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/EsEntityIndexBatchImpl.java
@@ -81,7 +81,7 @@ public class EsEntityIndexBatchImpl implements EntityIndexBatch {
}
//add app id for indexing
- container.addIndexRequest( new IndexRequest( writeAlias, applicationScope, indexEdge, entity ) );
+ container.addIndexRequest( new IndexOperation( writeAlias, applicationScope, indexEdge, entity ) );
return this;
}
@@ -105,7 +105,7 @@ public class EsEntityIndexBatchImpl implements EntityIndexBatch {
}
- container.addDeIndexRequest( new DeIndexRequest( indexes, applicationScope, searchEdge, id, version ) );
+ container.addDeIndexRequest( new DeIndexOperation( indexes, applicationScope, searchEdge, id, version ) );
return this;
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/e91930ff/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/EsIndexBufferConsumerImpl.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/EsIndexBufferConsumerImpl.java b/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/EsIndexBufferConsumerImpl.java
index 7cc7dcf..0577f28 100644
--- a/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/EsIndexBufferConsumerImpl.java
+++ b/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/EsIndexBufferConsumerImpl.java
@@ -21,6 +21,7 @@ package org.apache.usergrid.persistence.index.impl;
import java.util.List;
+import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
@@ -124,67 +125,80 @@ public class EsIndexBufferConsumerImpl implements IndexBufferConsumer {
private void startSubscription() {
- Observable.create( bufferProducer )
- .buffer( indexFig.getIndexBufferSize(), indexFig.getIndexBufferTimeout(), TimeUnit.MILLISECONDS,
- Schedulers.newThread() )
+ final Observable<IndexOperationMessage> observable = Observable.create( bufferProducer );
- .doOnNext( containerList -> {
- if ( containerList.size() == 0 ) {
- return;
- }
+ //buffer on our new thread with a timeout
+ observable.buffer( indexFig.getIndexBufferSize(), indexFig.getIndexBufferTimeout(), TimeUnit.MILLISECONDS,
+ Schedulers.newThread() ).flatMap( indexOpBuffer -> {
- flushMeter.mark( containerList.size() );
- Timer.Context time = flushTimer.time();
+ //hand off to processor in new observable thread so we can continue to buffer faster
+ return Observable.just( indexOpBuffer ).flatMap(
+ indexOpBufferObservable -> processBatch( indexOpBufferObservable ) )
+ //use the I/O scheduler for thread re-use and efficiency in context switching then use our concurrent
+ // flatmap count or higher throughput of batches once buffered
+ .subscribeOn( Schedulers.io() );
+ }, indexFig.getIndexFlushWorkerCount() )
+ //start in the background
+ .subscribe();
+ }
- execute( containerList );
- time.stop();
- } )
- //ack after we process
- .doOnNext( indexOperationMessages -> {
+ /**
+ * Process the buffer of batches
+ * @param batches
+ * @return
+ */
+ private Observable<IndexOperationMessage> processBatch( final List<IndexOperationMessage> batches ) {
- //release so we know we've done processing
- inFlight.addAndGet( -1 * indexOperationMessages.size() );
- } ).subscribe();
- //start in the background
+ final Observable<IndexOperationMessage> indexOps = Observable.from( batches );
- }
+ //take our stream of batches, then stream then into individual ops for consumption on ES
+ final Observable<BatchOperation> batchOps = indexOps.flatMap( batch -> {
+ final Set<IndexOperation> indexOperationSet = batch.getIndexRequests();
+ final Set<DeIndexOperation> deIndexOperationSet = batch.getDeIndexRequests();
- /**
- * Execute the request, check for errors, then re-init the batch for future use
- */
- private void execute( final List<IndexOperationMessage> operationMessages ) {
+ final int indexOperationSetSize = indexOperationSet.size();
+ final int deIndexOperationSetSize = deIndexOperationSet.size();
- if ( operationMessages == null || operationMessages.size() == 0 ) {
- return;
- }
+ log.debug( "Emitting {} add and {} remove operations", indexOperationSetSize, deIndexOperationSetSize );
- //process and flatten all the messages to builder requests
- //batch shard operations into a bulk request
- Observable.from( operationMessages ).flatMap( indexOperationMessage -> {
- final Observable<IndexRequest> index = Observable.from( indexOperationMessage.getIndexRequests() );
- final Observable<DeIndexRequest> deIndex = Observable.from( indexOperationMessage.getDeIndexRequests() );
+ indexSizeCounter.dec( indexOperationSetSize );
+ indexSizeCounter.dec( deIndexOperationSetSize );
- indexSizeCounter.dec( indexOperationMessage.getDeIndexRequests().size() );
- indexSizeCounter.dec( indexOperationMessage.getIndexRequests().size() );
+ final Observable<IndexOperation> index = Observable.from( batch.getIndexRequests() );
+ final Observable<DeIndexOperation> deIndex = Observable.from( batch.getDeIndexRequests() );
return Observable.merge( index, deIndex );
- } )
- //collection all the operations into a single stream
- .collect( () -> initRequest(), ( bulkRequestBuilder, batchRequest ) -> {
- batchRequest.doOperation( client, bulkRequestBuilder );
- } ) //send the request off to ES
- .doOnNext( bulkRequestBuilder -> sendRequest( bulkRequestBuilder ) ).toBlocking().lastOrDefault( null );
-
- //call back all futures
- Observable.from( operationMessages ).doOnNext( operationMessage -> operationMessage.getFuture().done() )
- .toBlocking().lastOrDefault( null );
+ } );
+
+ //buffer into the max size we can send ES and fire them all off until we're completed
+ final Observable<BulkRequestBuilder> requests = batchOps.buffer( indexFig.getIndexBatchSize() )
+ //flatten the buffer into a single batch execution
+ .flatMap( individualOps -> Observable.from( individualOps )
+ //collect them
+ .collect( () -> initRequest(), ( bulkRequestBuilder, batchOperation ) -> {
+ log.debug( "adding operation {} to bulkRequestBuilder {}", batchOperation, bulkRequestBuilder );
+ batchOperation.doOperation( client, bulkRequestBuilder );
+ } ) )
+ //write them
+ .doOnNext( bulkRequestBuilder -> sendRequest( bulkRequestBuilder ) ).doOnError( t -> log.error( "Unable to process batches", t ) );
+
+
+ //now that we've processed them all, ack the futures after our last batch comes through
+ final Observable<IndexOperationMessage> processedIndexOperations =
+ requests.last().flatMap( lastRequest -> Observable.from( batches ) );
+
+ //subscribe to the operations that generate requests on a new thread so that we can execute them quickly
+ //mark this as done
+ return processedIndexOperations.doOnNext( processedIndexOp -> processedIndexOp.done() ).doOnError( t -> log.error( "Unable to ack futures", t ) );
}
+ /*
+
/**
* initialize request
*/
@@ -209,7 +223,7 @@ public class EsIndexBufferConsumerImpl implements IndexBufferConsumer {
try {
- responses = bulkRequest.execute().actionGet(indexFig.getWriteTimeout());
+ responses = bulkRequest.execute().actionGet( indexFig.getWriteTimeout() );
}
catch ( Throwable t ) {
log.error( "Unable to communicate with elasticsearch" );
@@ -249,6 +263,11 @@ public class EsIndexBufferConsumerImpl implements IndexBufferConsumer {
* Send the data through the buffer
*/
public void send( final IndexOperationMessage indexOp ) {
+ //no-op
+ if ( indexOp.isEmpty() ) {
+ indexOp.done();
+ return;
+ }
subscriber.onNext( indexOp );
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/e91930ff/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/IndexOperation.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/IndexOperation.java b/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/IndexOperation.java
new file mode 100644
index 0000000..2e2f188
--- /dev/null
+++ b/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/IndexOperation.java
@@ -0,0 +1,107 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.usergrid.persistence.index.impl;
+
+
+import java.util.Map;
+
+import org.apache.usergrid.persistence.core.scope.ApplicationScope;
+import org.apache.usergrid.persistence.index.IndexEdge;
+import org.apache.usergrid.persistence.model.entity.Entity;
+
+import org.elasticsearch.action.bulk.BulkRequestBuilder;
+import org.elasticsearch.action.index.IndexRequestBuilder;
+import org.elasticsearch.client.Client;
+
+import static org.apache.usergrid.persistence.index.impl.IndexingUtils.idString;
+
+
+/**
+ * Represent the properties required to build an index request
+ */
+public class IndexOperation implements BatchOperation {
+
+ public String writeAlias;
+ public String documentId;
+
+ public Map<String, Object> data;
+
+ public IndexOperation( final String writeAlias, final ApplicationScope applicationScope, IndexEdge indexEdge,
+ Entity entity ) {
+ this(writeAlias,IndexingUtils.createIndexDocId(applicationScope, entity,indexEdge), EntityToMapConverter.convert(applicationScope,indexEdge, entity));
+
+ }
+
+ public IndexOperation( final String writeAlias, String documentId, Map<String, Object> data ) {
+ this.writeAlias = writeAlias;
+ this.data = data;
+ this.documentId = documentId;
+ }
+
+ /**
+ * DO NOT DELETE! Required for Jackson
+ */
+ public IndexOperation() {
+ }
+
+
+ public void doOperation( final Client client, final BulkRequestBuilder bulkRequest ) {
+ IndexRequestBuilder builder = client.prepareIndex( writeAlias, IndexingUtils.ES_ENTITY_TYPE, documentId ).setSource( data );
+
+
+ bulkRequest.add( builder );
+ }
+
+
+
+ @Override
+ public boolean equals( final Object o ) {
+ if ( this == o ) {
+ return true;
+ }
+ if ( o == null || getClass() != o.getClass() ) {
+ return false;
+ }
+
+ final IndexOperation that = ( IndexOperation ) o;
+
+ if ( !data.equals( that.data ) ) {
+ return false;
+ }
+ if ( !documentId.equals( that.documentId ) ) {
+ return false;
+ }
+
+ if ( !writeAlias.equals( that.writeAlias ) ) {
+ return false;
+ }
+
+ return true;
+ }
+
+
+ @Override
+ public int hashCode() {
+ int result = writeAlias.hashCode();
+ result = 31 * result + documentId.hashCode();
+ result = 31 * result + data.hashCode();
+ return result;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/e91930ff/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/IndexOperationMessage.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/IndexOperationMessage.java b/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/IndexOperationMessage.java
index ad9b91f..e830476 100644
--- a/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/IndexOperationMessage.java
+++ b/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/IndexOperationMessage.java
@@ -34,8 +34,8 @@ import com.fasterxml.jackson.annotation.JsonIgnore;
* Container for index operations.
*/
public class IndexOperationMessage implements Serializable {
- private final Set<IndexRequest> indexRequests;
- private final Set<DeIndexRequest> deIndexRequests;
+ private final Set<IndexOperation> indexRequests;
+ private final Set<DeIndexOperation> deIndexRequests;
@@ -46,33 +46,28 @@ public class IndexOperationMessage implements Serializable {
final IndexOperationMessage parent = this;
this.indexRequests = new HashSet<>();
this.deIndexRequests = new HashSet<>();
- this.containerFuture = new BetterFuture<>( new Callable<IndexOperationMessage>() {
- @Override
- public IndexOperationMessage call() throws Exception {
- return parent;
- }
- } );
+ this.containerFuture = new BetterFuture<>( () -> parent );
}
- public void addIndexRequest( final IndexRequest indexRequest ) {
+ public void addIndexRequest( final IndexOperation indexRequest ) {
indexRequests.add( indexRequest );
}
- public void addDeIndexRequest( final DeIndexRequest deIndexRequest ) {
+ public void addDeIndexRequest( final DeIndexOperation deIndexRequest ) {
this.deIndexRequests.add( deIndexRequest );
}
- public Set<IndexRequest> getIndexRequests() {
+ public Set<IndexOperation> getIndexRequests() {
return indexRequests;
}
- public Set<DeIndexRequest> getDeIndexRequests() {
+ public Set<DeIndexOperation> getDeIndexRequests() {
return deIndexRequests;
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/e91930ff/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/IndexRefreshCommandImpl.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/IndexRefreshCommandImpl.java b/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/IndexRefreshCommandImpl.java
index c997e5a..3f2a56f 100644
--- a/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/IndexRefreshCommandImpl.java
+++ b/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/IndexRefreshCommandImpl.java
@@ -104,7 +104,7 @@ public class IndexRefreshCommandImpl implements IndexRefreshCommand {
final String entityId = entityData.get( IndexingUtils.ENTITY_ID_FIELDNAME ).toString();
//add a tracer record
- IndexRequest indexRequest = new IndexRequest( alias.getWriteAlias(), docId, entityData );
+ IndexOperation indexRequest = new IndexOperation( alias.getWriteAlias(), docId, entityData );
//save the item
IndexOperationMessage message = new IndexOperationMessage();
@@ -155,8 +155,8 @@ public class IndexRefreshCommandImpl implements IndexRefreshCommand {
} ).doOnCompleted(() -> {
//clean up our data
String[] aliases = indexCache.getIndexes(alias, AliasedEntityIndex.AliasType.Read);
- DeIndexRequest deIndexRequest =
- new DeIndexRequest(aliases, appScope, edge, entity.getId(), entity.getVersion());
+ DeIndexOperation deIndexRequest =
+ new DeIndexOperation(aliases, appScope, edge, entity.getId(), entity.getVersion());
//delete the item
IndexOperationMessage indexOperationMessage =
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/e91930ff/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/IndexRequest.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/IndexRequest.java b/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/IndexRequest.java
deleted file mode 100644
index 5852394..0000000
--- a/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/impl/IndexRequest.java
+++ /dev/null
@@ -1,111 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-package org.apache.usergrid.persistence.index.impl;
-
-
-import java.util.Map;
-
-import org.apache.usergrid.persistence.core.scope.ApplicationScope;
-import org.apache.usergrid.persistence.index.IndexEdge;
-import org.apache.usergrid.persistence.index.SearchEdge;
-import org.apache.usergrid.persistence.index.SearchType;
-import org.apache.usergrid.persistence.model.entity.Entity;
-
-import org.elasticsearch.action.bulk.BulkRequestBuilder;
-import org.elasticsearch.action.index.IndexRequestBuilder;
-import org.elasticsearch.client.Client;
-
-import com.fasterxml.jackson.annotation.JsonTypeInfo;
-
-import static org.apache.usergrid.persistence.index.impl.IndexingUtils.createContextName;
-import static org.apache.usergrid.persistence.index.impl.IndexingUtils.idString;
-
-
-/**
- * Represent the properties required to build an index request
- */
-public class IndexRequest implements BatchRequest {
-
- public String writeAlias;
- public String documentId;
-
- public Map<String, Object> data;
-
- public IndexRequest( final String writeAlias, final ApplicationScope applicationScope, IndexEdge indexEdge, Entity entity) {
- this(writeAlias,IndexingUtils.createIndexDocId(applicationScope, entity,indexEdge), EntityToMapConverter.convert(applicationScope,indexEdge, entity));
-
- }
-
- public IndexRequest( final String writeAlias, String documentId, Map<String, Object> data) {
- this.writeAlias = writeAlias;
- this.data = data;
- this.documentId = documentId;
- }
-
- /**
- * DO NOT DELETE! Required for Jackson
- */
- public IndexRequest() {
- }
-
-
- public void doOperation( final Client client, final BulkRequestBuilder bulkRequest ) {
- IndexRequestBuilder builder = client.prepareIndex( writeAlias, IndexingUtils.ES_ENTITY_TYPE, documentId ).setSource( data );
-
-
- bulkRequest.add( builder );
- }
-
-
-
- @Override
- public boolean equals( final Object o ) {
- if ( this == o ) {
- return true;
- }
- if ( o == null || getClass() != o.getClass() ) {
- return false;
- }
-
- final IndexRequest that = ( IndexRequest ) o;
-
- if ( !data.equals( that.data ) ) {
- return false;
- }
- if ( !documentId.equals( that.documentId ) ) {
- return false;
- }
-
- if ( !writeAlias.equals( that.writeAlias ) ) {
- return false;
- }
-
- return true;
- }
-
-
- @Override
- public int hashCode() {
- int result = writeAlias.hashCode();
- result = 31 * result + documentId.hashCode();
- result = 31 * result + data.hashCode();
- return result;
- }
-}
[2/2] incubator-usergrid git commit: Merge branch 'two-dot-o-dev' of
https://git-wip-us.apache.org/repos/asf/incubator-usergrid into two-dot-o-dev
Posted by to...@apache.org.
Merge branch 'two-dot-o-dev' of https://git-wip-us.apache.org/repos/asf/incubator-usergrid into two-dot-o-dev
Project: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/commit/852c9747
Tree: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/tree/852c9747
Diff: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/diff/852c9747
Branch: refs/heads/two-dot-o-dev
Commit: 852c9747bb8c88401a343d971a68203ee19459c5
Parents: e91930f 0141623
Author: Todd Nine <tn...@apigee.com>
Authored: Thu Apr 23 10:24:28 2015 -0600
Committer: Todd Nine <tn...@apigee.com>
Committed: Thu Apr 23 10:24:28 2015 -0600
----------------------------------------------------------------------
.../Usergrid.Notifications/BackgroundTask.cs | 57 +++++++++++++++++++
.../Usergrid.Notifications/Client/PushClient.cs | 12 +++-
.../Usergrid.Notifications/MainPage.xaml.cs | 12 ++--
.../Usergrid.Notifications/MyBackgroundTask.cs | 18 ++++++
.../Usergrid.Notifications/Package.appxmanifest | 8 +++
.../Usergrid.Notifications.csproj | 1 +
.../Usergrid.Notifications.sln | 40 +++++++++++++
.../persistence/query/IteratingQueryIT.java | 54 +++++++++---------
.../services/notifications/wns/WNSAdapter.java | 60 +++++++-------------
9 files changed, 189 insertions(+), 73 deletions(-)
----------------------------------------------------------------------