ElementSerialisation { public long getTimestamp(final Element element) throws SerialisationException { return getTimestamp(element.getProperties()); } ElementSerialisation(final Schema schema); byte[] getValue(final Element element); byte[] getValue(final String group, final Properties properties); Properties getPropertiesFromValue(final String group, final byte[] value); Element getPartialElement(final String group, final byte[] rowId, final boolean includeMatchedVertex); Element getElement(final Cell cell, final boolean includeMatchedVertex); Properties getProperties(final String group, final Cell cell); byte[] getColumnVisibility(final Element element); byte[] getColumnVisibility(final String group, final Properties properties); byte[] getColumnQualifier(final Element element); byte[] getColumnQualifier(final String group, final Properties properties); Properties getPropertiesFromColumnQualifier(final String group, final byte[] bytes); byte[] getPropertiesAsBytesFromColumnQualifier(final String group, final byte[] bytes, final int numProps); long getTimestamp(final Element element); long getTimestamp(final Properties properties); Properties getPropertiesFromTimestamp(final String group, final long timestamp); byte[] serialiseVertex(final Object vertex); boolean isEntity(final Cell cell); String getGroup(final Cell cell); String getGroup(final byte[] columnQualifier); @SuppressFBWarnings(value = "BC_UNCONFIRMED_CAST", justification = "If an element is not an Entity it must be an Edge") Pair<byte[], byte[]> getRowKeys(final Element element); byte[] getRowKey(final Entity entity); Pair<byte[], byte[]> getRowKeys(final Edge edge); Pair<Cell, Cell> getCells(final Element element, final CellCreator kvCreator); Pair<Cell, Cell> getCells(final Element element, final Pair<byte[], byte[]> row, final byte[] cq, final CellCreator kvCreator); Pair<Put, Put> getPuts(final Element element); Pair<Put, Put> getPuts(final Element element, final Pair<byte[], byte[]> row, final byte[] cq); EdgeDirection getSourceAndDestination(final byte[] rowKey, final byte[][] sourceDestValues);  }
@Test public void shouldBuildTimestampFromProperty() throws Exception { final Schema schema = new Schema.Builder() .json(StreamUtil.schemas(getClass())) .build(); serialisation = new ElementSerialisation(new Schema.Builder(schema) .type("timestamp", Long.class) .edge(TestGroups.EDGE, new SchemaEdgeDefinition.Builder() .property(HBasePropertyNames.TIMESTAMP, "timestamp") .build()) .config(HBaseStoreConstants.TIMESTAMP_PROPERTY, TestPropertyNames.TIMESTAMP) .build()); final long propertyTimestamp = 10L; final Properties properties = new Properties(); properties.put(HBasePropertyNames.COLUMN_QUALIFIER, 1); properties.put(HBasePropertyNames.PROP_1, 2); properties.put(HBasePropertyNames.TIMESTAMP, propertyTimestamp); final long timestamp = serialisation.getTimestamp(properties); assertEquals(propertyTimestamp, timestamp); }
@Test public void shouldBuildRandomTimeBasedTimestampWhenPropertyIsNull() throws Exception { final Schema schema = new Schema.Builder().json(StreamUtil.schemas(getClass())).build(); serialisation = new ElementSerialisation(new Schema.Builder(schema) .config(HBaseStoreConstants.TIMESTAMP_PROPERTY, TestPropertyNames.TIMESTAMP) .build()); final Long propertyTimestamp = null; final Properties properties = new Properties(); properties.put(HBasePropertyNames.TIMESTAMP, propertyTimestamp); final int n = 100; final Set<Long> timestamps = new HashSet<>(n); for (int i = 0; i < n; i++) { timestamps.add(serialisation.getTimestamp(properties)); } assertEquals(n, timestamps.size()); }
@Test public void shouldBuildRandomTimeBasedTimestamp() throws Exception { final Properties properties = new Properties(); properties.put(HBasePropertyNames.COLUMN_QUALIFIER, 1); properties.put(HBasePropertyNames.PROP_1, 2); final int n = 100; final Set<Long> timestamps = new HashSet<>(n); for (int i = 0; i < n; i++) { timestamps.add(serialisation.getTimestamp(properties)); } assertEquals(n, timestamps.size()); }