You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by da...@apache.org on 2015/02/12 07:24:19 UTC
[2/2] camel git commit: CAMEL-8268: Various camel-hbases fixes and
improvements. Thanks to Emilien for the patch.
CAMEL-8268: Various camel-hbases fixes and improvements. Thanks to Emilien for the patch.
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/9edcfbc1
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/9edcfbc1
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/9edcfbc1
Branch: refs/heads/master
Commit: 9edcfbc1bfeeeb3625fb91815f2a8a117679fd16
Parents: e7c6529
Author: Claus Ibsen <da...@apache.org>
Authored: Thu Feb 12 07:15:20 2015 +0100
Committer: Claus Ibsen <da...@apache.org>
Committed: Thu Feb 12 07:15:20 2015 +0100
----------------------------------------------------------------------
.../camel/component/hbase/HBaseConstants.java | 3 +-
.../camel/component/hbase/HBaseEndpoint.java | 4 +-
.../camel/component/hbase/HBaseProducer.java | 69 +++++++++++++-------
.../ModelAwareRowPrefixMatchingFilter.java | 50 ++++++++++++++
.../hbase/mapping/HeaderMappingStrategy.java | 2 +
.../camel/component/hbase/model/HBaseCell.java | 27 +++++++-
.../camel/component/hbase/model/HBaseRow.java | 10 +++
.../component/hbase/HBaseProducerTest.java | 25 +++++++
8 files changed, 160 insertions(+), 30 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/9edcfbc1/components/camel-hbase/src/main/java/org/apache/camel/component/hbase/HBaseConstants.java
----------------------------------------------------------------------
diff --git a/components/camel-hbase/src/main/java/org/apache/camel/component/hbase/HBaseConstants.java b/components/camel-hbase/src/main/java/org/apache/camel/component/hbase/HBaseConstants.java
index c91a4d6..42ae4fc 100644
--- a/components/camel-hbase/src/main/java/org/apache/camel/component/hbase/HBaseConstants.java
+++ b/components/camel-hbase/src/main/java/org/apache/camel/component/hbase/HBaseConstants.java
@@ -26,5 +26,6 @@ public interface HBaseConstants {
String DELETE = "CamelHBaseDelete";
String HBASE_MAX_SCAN_RESULTS = "CamelHBaseMaxScanResults";
-
+
+ String FROM_ROW = "CamelHBaseStartRow";
}
http://git-wip-us.apache.org/repos/asf/camel/blob/9edcfbc1/components/camel-hbase/src/main/java/org/apache/camel/component/hbase/HBaseEndpoint.java
----------------------------------------------------------------------
diff --git a/components/camel-hbase/src/main/java/org/apache/camel/component/hbase/HBaseEndpoint.java b/components/camel-hbase/src/main/java/org/apache/camel/component/hbase/HBaseEndpoint.java
index ff547ef..11587fb 100644
--- a/components/camel-hbase/src/main/java/org/apache/camel/component/hbase/HBaseEndpoint.java
+++ b/components/camel-hbase/src/main/java/org/apache/camel/component/hbase/HBaseEndpoint.java
@@ -39,11 +39,11 @@ import org.apache.hadoop.hbase.filter.Filter;
public class HBaseEndpoint extends DefaultEndpoint {
private Configuration configuration;
- @UriPath
- private final String tableName;
private final HTablePool tablePool;
private HBaseAdmin admin;
+ @UriPath
+ private final String tableName;
//Operation properties.
@UriParam(defaultValue = "100")
private int maxResults = 100;
http://git-wip-us.apache.org/repos/asf/camel/blob/9edcfbc1/components/camel-hbase/src/main/java/org/apache/camel/component/hbase/HBaseProducer.java
----------------------------------------------------------------------
diff --git a/components/camel-hbase/src/main/java/org/apache/camel/component/hbase/HBaseProducer.java b/components/camel-hbase/src/main/java/org/apache/camel/component/hbase/HBaseProducer.java
index b9c5ebb..053309f 100644
--- a/components/camel-hbase/src/main/java/org/apache/camel/component/hbase/HBaseProducer.java
+++ b/components/camel-hbase/src/main/java/org/apache/camel/component/hbase/HBaseProducer.java
@@ -19,6 +19,7 @@ package org.apache.camel.component.hbase;
import java.util.LinkedList;
import java.util.List;
import java.util.Set;
+
import org.apache.camel.Exchange;
import org.apache.camel.ServicePoolAware;
import org.apache.camel.component.hbase.filters.ModelAwareFilter;
@@ -29,7 +30,6 @@ import org.apache.camel.component.hbase.model.HBaseData;
import org.apache.camel.component.hbase.model.HBaseRow;
import org.apache.camel.impl.DefaultProducer;
import org.apache.camel.util.ObjectHelper;
-
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Get;
@@ -41,6 +41,7 @@ import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.filter.FilterList;
+import org.apache.hadoop.hbase.util.Bytes;
/**
* The HBase producer.
@@ -66,6 +67,9 @@ public class HBaseProducer extends DefaultProducer implements ServicePoolAware {
updateHeaders(exchange);
String operation = (String) exchange.getIn().getHeader(HBaseConstants.OPERATION);
+
+ Integer maxScanResult = exchange.getIn().getHeader(HBaseConstants.HBASE_MAX_SCAN_RESULTS, Integer.class);
+ String fromRowId = (String) exchange.getIn().getHeader(HBaseConstants.FROM_ROW);
CellMappingStrategy mappingStrategy = endpoint.getCellMappingStrategyFactory().getStrategy(exchange.getIn());
HBaseData data = mappingStrategy.resolveModel(exchange.getIn());
@@ -85,7 +89,7 @@ public class HBaseProducer extends DefaultProducer implements ServicePoolAware {
} else if (HBaseConstants.DELETE.equals(operation)) {
deleteOperations.add(createDeleteRow(hRow));
} else if (HBaseConstants.SCAN.equals(operation)) {
- scanOperationResult = scanCells(table, hRow, endpoint.getFilters());
+ scanOperationResult = scanCells(table, hRow, fromRowId, maxScanResult, endpoint.getFilters());
}
}
@@ -131,13 +135,8 @@ public class HBaseProducer extends DefaultProducer implements ServicePoolAware {
}
/**
- * Perfoms an HBase {@link Get} on a specific row, using a collection of values (family/column/value pairs).
+ * Performs an HBase {@link Get} on a specific row, using a collection of values (family/column/value pairs).
* The result is <p>the most recent entry</p> for each column.
- *
- * @param table
- * @param hRow
- * @return
- * @throws Exception
*/
private HBaseRow getCells(HTableInterface table, HBaseRow hRow) throws Exception {
HBaseRow resultRow = new HBaseRow();
@@ -160,6 +159,10 @@ public class HBaseProducer extends DefaultProducer implements ServicePoolAware {
Result result = table.get(get);
+ if (!result.isEmpty()) {
+ resultRow.setTimestamp(result.raw()[0].getTimestamp());
+ }
+
for (HBaseCell cellModel : cellModels) {
HBaseCell resultCell = new HBaseCell();
String family = cellModel.getFamily();
@@ -171,6 +174,7 @@ public class HBaseProducer extends DefaultProducer implements ServicePoolAware {
if (kvs != null && !kvs.isEmpty()) {
//Return the most recent entry.
resultCell.setValue(endpoint.getCamelContext().getTypeConverter().convertTo(cellModel.getValueType(), kvs.get(0).getValue()));
+ resultCell.setTimestamp(kvs.get(0).getTimestamp());
}
resultCells.add(resultCell);
resultRow.getCells().add(resultCell);
@@ -180,9 +184,6 @@ public class HBaseProducer extends DefaultProducer implements ServicePoolAware {
/**
* Creates an HBase {@link Delete} on a specific row, using a collection of values (family/column/value pairs).
- *
- * @param hRow
- * @throws Exception
*/
private Delete createDeleteRow(HBaseRow hRow) throws Exception {
ObjectHelper.notNull(hRow, "HBase row");
@@ -191,25 +192,37 @@ public class HBaseProducer extends DefaultProducer implements ServicePoolAware {
}
/**
- * Perfoms an HBase {@link Get} on a specific row, using a collection of values (family/column/value pairs).
+ * Performs an HBase {@link Get} on a specific row, using a collection of values (family/column/value pairs).
* The result is <p>the most recent entry</p> for each column.
- *
- * @param table
- * @param model
- * @return
- * @throws Exception
*/
- private List<HBaseRow> scanCells(HTableInterface table, HBaseRow model, List<Filter> filters) throws Exception {
+ private List<HBaseRow> scanCells(HTableInterface table, HBaseRow model, String start, Integer maxRowScan, List<Filter> filters) throws Exception {
List<HBaseRow> rowSet = new LinkedList<HBaseRow>();
- Scan scan = new Scan();
+
+ HBaseRow startRow = new HBaseRow(model.getCells());
+ startRow.setId(start);
+
+ Scan scan;
+ if (start != null) {
+ scan = new Scan(Bytes.toBytes(start));
+ } else {
+ scan = new Scan();
+ }
+
+ // need to clone the filters as they are not thread safe to use
if (filters != null && !filters.isEmpty()) {
+ List<Filter> clonedFilters = new LinkedList<Filter>();
for (Filter filter : filters) {
if (ModelAwareFilter.class.isAssignableFrom(filter.getClass())) {
- ((ModelAwareFilter<?>) filter).apply(endpoint.getCamelContext(), model);
+ Object clone = endpoint.getCamelContext().getInjector().newInstance(filter.getClass());
+ if (clone instanceof ModelAwareFilter) {
+ ((ModelAwareFilter<?>) clone).apply(endpoint.getCamelContext(), model);
+ clonedFilters.add((ModelAwareFilter<?>) clone);
+ }
}
}
- scan.setFilter(new FilterList(FilterList.Operator.MUST_PASS_ALL, filters));
+ scan.setFilter(new FilterList(FilterList.Operator.MUST_PASS_ALL, clonedFilters));
}
+
Set<HBaseCell> cellModels = model.getCells();
for (HBaseCell cellModel : cellModels) {
String family = cellModel.getFamily();
@@ -221,10 +234,14 @@ public class HBaseProducer extends DefaultProducer implements ServicePoolAware {
}
ResultScanner resultScanner = table.getScanner(scan);
+ int count = 0;
Result result = resultScanner.next();
- while (result != null) {
+
+ while (result != null && count < maxRowScan) {
HBaseRow resultRow = new HBaseRow();
resultRow.setId(endpoint.getCamelContext().getTypeConverter().convertTo(model.getRowType(), result.getRow()));
+
+ resultRow.setTimestamp(result.raw()[0].getTimestamp());
cellModels = model.getCells();
for (HBaseCell modelCell : cellModels) {
HBaseCell resultCell = new HBaseCell();
@@ -235,10 +252,14 @@ public class HBaseProducer extends DefaultProducer implements ServicePoolAware {
result.getValue(HBaseHelper.getHBaseFieldAsBytes(family), HBaseHelper.getHBaseFieldAsBytes(column))));
resultCell.setFamily(modelCell.getFamily());
resultCell.setQualifier(modelCell.getQualifier());
+
+ if (result.getColumnLatest(HBaseHelper.getHBaseFieldAsBytes(family), HBaseHelper.getHBaseFieldAsBytes(column)) != null) {
+ resultCell.setTimestamp(result.getColumnLatest(HBaseHelper.getHBaseFieldAsBytes(family), HBaseHelper.getHBaseFieldAsBytes(column)).getTimestamp());
+ }
resultRow.getCells().add(resultCell);
- rowSet.add(resultRow);
}
-
+ rowSet.add(resultRow);
+ count++;
result = resultScanner.next();
}
return rowSet;
http://git-wip-us.apache.org/repos/asf/camel/blob/9edcfbc1/components/camel-hbase/src/main/java/org/apache/camel/component/hbase/filters/ModelAwareRowPrefixMatchingFilter.java
----------------------------------------------------------------------
diff --git a/components/camel-hbase/src/main/java/org/apache/camel/component/hbase/filters/ModelAwareRowPrefixMatchingFilter.java b/components/camel-hbase/src/main/java/org/apache/camel/component/hbase/filters/ModelAwareRowPrefixMatchingFilter.java
new file mode 100644
index 0000000..84be756
--- /dev/null
+++ b/components/camel-hbase/src/main/java/org/apache/camel/component/hbase/filters/ModelAwareRowPrefixMatchingFilter.java
@@ -0,0 +1,50 @@
+/**
+ * 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.camel.component.hbase.filters;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.component.hbase.model.HBaseRow;
+import org.apache.hadoop.hbase.filter.FilterList;
+import org.apache.hadoop.hbase.filter.PrefixFilter;
+
+/**
+ * A {@link FilterList} that contains multiple {@link PrefixFilter}s one per column that is part of the model.
+ */
+public class ModelAwareRowPrefixMatchingFilter extends FilterList implements ModelAwareFilter<FilterList> {
+
+ /**
+ * Writable constructor, do not use.
+ */
+ public ModelAwareRowPrefixMatchingFilter() {
+ }
+
+ /**
+ * Applies the message to {@link org.apache.hadoop.hbase.filter.Filter} to
+ * context.
+ */
+ @Override
+ public void apply(CamelContext context, HBaseRow rowModel) {
+ getFilters().clear();
+ if (rowModel != null) {
+ if (rowModel.getId() != null) {
+ byte[] value = context.getTypeConverter().convertTo(byte[].class, rowModel.getId());
+ PrefixFilter rowPrefixFilter = new PrefixFilter(value);
+ addFilter(rowPrefixFilter);
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/9edcfbc1/components/camel-hbase/src/main/java/org/apache/camel/component/hbase/mapping/HeaderMappingStrategy.java
----------------------------------------------------------------------
diff --git a/components/camel-hbase/src/main/java/org/apache/camel/component/hbase/mapping/HeaderMappingStrategy.java b/components/camel-hbase/src/main/java/org/apache/camel/component/hbase/mapping/HeaderMappingStrategy.java
index 7781de8..d8c54bd 100644
--- a/components/camel-hbase/src/main/java/org/apache/camel/component/hbase/mapping/HeaderMappingStrategy.java
+++ b/components/camel-hbase/src/main/java/org/apache/camel/component/hbase/mapping/HeaderMappingStrategy.java
@@ -106,6 +106,7 @@ public class HeaderMappingStrategy implements CellMappingStrategy {
* Applies the cells to the {@link org.apache.camel.Exchange}.
*/
public void applyGetResults(Message message, HBaseData data) {
+ message.setHeaders(message.getExchange().getIn().getHeaders());
int index = 1;
if (data == null || data.getRows() == null) {
return;
@@ -126,6 +127,7 @@ public class HeaderMappingStrategy implements CellMappingStrategy {
* Applies the cells to the {@link org.apache.camel.Exchange}.
*/
public void applyScanResults(Message message, HBaseData data) {
+ message.setHeaders(message.getExchange().getIn().getHeaders());
int index = 1;
if (data == null || data.getRows() == null) {
return;
http://git-wip-us.apache.org/repos/asf/camel/blob/9edcfbc1/components/camel-hbase/src/main/java/org/apache/camel/component/hbase/model/HBaseCell.java
----------------------------------------------------------------------
diff --git a/components/camel-hbase/src/main/java/org/apache/camel/component/hbase/model/HBaseCell.java b/components/camel-hbase/src/main/java/org/apache/camel/component/hbase/model/HBaseCell.java
index 54e5a1b..14f4104 100644
--- a/components/camel-hbase/src/main/java/org/apache/camel/component/hbase/model/HBaseCell.java
+++ b/components/camel-hbase/src/main/java/org/apache/camel/component/hbase/model/HBaseCell.java
@@ -28,6 +28,7 @@ public class HBaseCell {
private String family;
private String qualifier;
private Object value;
+ private Long timestamp;
//The value type can be optionally specified for Gets and Scan, to specify how the byte[] read will be converted.
private Class<?> valueType = String.class;
@@ -73,6 +74,14 @@ public class HBaseCell {
this.valueType = valueType;
}
+ public Long getTimestamp() {
+ return timestamp;
+ }
+
+ public void setTimestamp(Long timestamp) {
+ this.timestamp = timestamp;
+ }
+
@Override
public boolean equals(Object o) {
if (this == o) {
@@ -82,12 +91,21 @@ public class HBaseCell {
return false;
}
- HBaseCell cell = (HBaseCell) o;
+ HBaseCell hBaseCell = (HBaseCell) o;
- if (family != null ? !family.equals(cell.family) : cell.family != null) {
+ if (family != null ? !family.equals(hBaseCell.family) : hBaseCell.family != null) {
+ return false;
+ }
+ if (qualifier != null ? !qualifier.equals(hBaseCell.qualifier) : hBaseCell.qualifier != null) {
+ return false;
+ }
+ if (timestamp != null ? !timestamp.equals(hBaseCell.timestamp) : hBaseCell.timestamp != null) {
+ return false;
+ }
+ if (value != null ? !value.equals(hBaseCell.value) : hBaseCell.value != null) {
return false;
}
- if (qualifier != null ? !qualifier.equals(cell.qualifier) : cell.qualifier != null) {
+ if (valueType != null ? !valueType.equals(hBaseCell.valueType) : hBaseCell.valueType != null) {
return false;
}
@@ -98,6 +116,9 @@ public class HBaseCell {
public int hashCode() {
int result = family != null ? family.hashCode() : 0;
result = 31 * result + (qualifier != null ? qualifier.hashCode() : 0);
+ result = 31 * result + (value != null ? value.hashCode() : 0);
+ result = 31 * result + (timestamp != null ? timestamp.hashCode() : 0);
+ result = 31 * result + (valueType != null ? valueType.hashCode() : 0);
return result;
}
}
http://git-wip-us.apache.org/repos/asf/camel/blob/9edcfbc1/components/camel-hbase/src/main/java/org/apache/camel/component/hbase/model/HBaseRow.java
----------------------------------------------------------------------
diff --git a/components/camel-hbase/src/main/java/org/apache/camel/component/hbase/model/HBaseRow.java b/components/camel-hbase/src/main/java/org/apache/camel/component/hbase/model/HBaseRow.java
index f267bcf..daf6e5f 100644
--- a/components/camel-hbase/src/main/java/org/apache/camel/component/hbase/model/HBaseRow.java
+++ b/components/camel-hbase/src/main/java/org/apache/camel/component/hbase/model/HBaseRow.java
@@ -29,6 +29,8 @@ public class HBaseRow implements Cloneable {
private Class<?> rowType = String.class;
private Set<HBaseCell> cells;
+ private long timestamp;
+
public HBaseRow() {
this(new LinkedHashSet<HBaseCell>());
}
@@ -70,6 +72,14 @@ public class HBaseRow implements Cloneable {
return cells.size();
}
+ public long getTimestamp() {
+ return timestamp;
+ }
+
+ public void setTimestamp(long timestamp) {
+ this.timestamp = timestamp;
+ }
+
public void apply(HBaseRow modelRow) {
if (modelRow != null) {
if (rowType == null && modelRow.getRowType() != null) {
http://git-wip-us.apache.org/repos/asf/camel/blob/9edcfbc1/components/camel-hbase/src/test/java/org/apache/camel/component/hbase/HBaseProducerTest.java
----------------------------------------------------------------------
diff --git a/components/camel-hbase/src/test/java/org/apache/camel/component/hbase/HBaseProducerTest.java b/components/camel-hbase/src/test/java/org/apache/camel/component/hbase/HBaseProducerTest.java
index 2efd3b7..9e71c65 100644
--- a/components/camel-hbase/src/test/java/org/apache/camel/component/hbase/HBaseProducerTest.java
+++ b/components/camel-hbase/src/test/java/org/apache/camel/component/hbase/HBaseProducerTest.java
@@ -233,6 +233,28 @@ public class HBaseProducerTest extends CamelHBaseTestSupport {
}
@Test
+ public void testPutMultiRowsAndMaxScan() throws Exception {
+ testPutMultiRows();
+ if (systemReady) {
+ Exchange resp = template.request("direct:maxScan", new Processor() {
+ public void process(Exchange exchange) throws Exception {
+ exchange.getIn().setHeader(HbaseAttribute.HBASE_FAMILY.asHeader(), family[0]);
+ exchange.getIn().setHeader(HbaseAttribute.HBASE_QUALIFIER.asHeader(), column[0][0]);
+ }
+ });
+
+ Object result1 = resp.getOut().getHeader(HbaseAttribute.HBASE_VALUE.asHeader(1));
+ Object result2 = resp.getOut().getHeader(HbaseAttribute.HBASE_VALUE.asHeader(2));
+ // as we use maxResults=2 we only get 2 results back
+ Object result3 = resp.getOut().getHeader(HbaseAttribute.HBASE_VALUE.asHeader(3));
+ assertNull("Should only get 2 results back", result3);
+
+ List<?> bodies = Arrays.asList(body[0][0][0], body[1][0][0]);
+ assertTrue(bodies.contains(result1) && bodies.contains(result2));
+ }
+ }
+
+ @Test
public void testPutMultiRowsAndScan() throws Exception {
testPutMultiRows();
if (systemReady) {
@@ -304,6 +326,9 @@ public class HBaseProducerTest extends CamelHBaseTestSupport {
.to("hbase://" + PERSON_TABLE + "?family=info&qualifier=firstName&family2=birthdate&qualifier2=year");
from("direct:scan")
+ .to("hbase://" + PERSON_TABLE + "?operation=" + HBaseConstants.SCAN);
+
+ from("direct:maxScan")
.to("hbase://" + PERSON_TABLE + "?operation=" + HBaseConstants.SCAN + "&maxResults=2");
}
};