You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@samza.apache.org by we...@apache.org on 2018/11/07 22:46:27 UTC
samza git commit: SAMZA-1980: Rename LocalStoreBackedTable to
LocalTable
Repository: samza
Updated Branches:
refs/heads/master ad12c6ddd -> 2159171ee
SAMZA-1980: Rename LocalStoreBackedTable to LocalTable
As per subject, this is to keep naming of local tables consistent with other table types.
Author: Wei Song <ws...@linkedin.com>
Reviewers: Prateek Maheshwari <pm...@linkedin.com>
Closes #798 from weisong44/SAMZA-1980 and squashes the following commits:
51c17ff4 [Wei Song] Renamed LocalStoreBackedTable to LocalTable
9c121207 [Wei Song] Merge remote-tracking branch 'upstream/master'
89bfc14c [Wei Song] Merge remote-tracking branch 'upstream/master'
a53e5628 [Wei Song] SAMZA-1964 Make getTableSpec() in RemoteTableDescriptor reentrant
c9e8bf7c [Wei Song] Merge remote-tracking branch 'upstream/master'
7c777fec [Wei Song] Merge remote-tracking branch 'upstream/master'
a06e8ec2 [Wei Song] Merge remote-tracking branch 'upstream/master'
2c679c39 [Wei Song] Merge remote-tracking branch 'upstream/master'
a56c28dc [Wei Song] Merge remote-tracking branch 'upstream/master'
097958c8 [Wei Song] Merge remote-tracking branch 'upstream/master'
05822f0a [Wei Song] Merge remote-tracking branch 'upstream/master'
f7480505 [Wei Song] Merge remote-tracking branch 'upstream/master'
7706ab1f [Wei Song] Merge remote-tracking branch 'upstream/master'
f5731b10 [Wei Song] Merge remote-tracking branch 'upstream/master'
1e5de45a [Wei Song] Merge remote-tracking branch 'upstream/master'
c85604e0 [Wei Song] Merge remote-tracking branch 'upstream/master'
242d8442 [Wei Song] Merge remote-tracking branch 'upstream/master'
ec7d8409 [Wei Song] Merge remote-tracking branch 'upstream/master'
e19b4dc9 [Wei Song] Merge remote-tracking branch 'upstream/master'
8ee78441 [Wei Song] Merge remote-tracking branch 'upstream/master'
1c6a2eae [Wei Song] Merge remote-tracking branch 'upstream/master'
a6c94add [Wei Song] Merge remote-tracking branch 'upstream/master'
41299b5b [Wei Song] Merge remote-tracking branch 'upstream/master'
239a0950 [Wei Song] Merge remote-tracking branch 'upstream/master'
eca00204 [Wei Song] Merge remote-tracking branch 'upstream/master'
51562391 [Wei Song] Merge remote-tracking branch 'upstream/master'
de708f5e [Wei Song] Merge remote-tracking branch 'upstream/master'
df2f8d7b [Wei Song] Merge remote-tracking branch 'upstream/master'
f28b491d [Wei Song] Merge remote-tracking branch 'upstream/master'
4782c61d [Wei Song] Merge remote-tracking branch 'upstream/master'
0440f75f [Wei Song] Merge remote-tracking branch 'upstream/master'
aae0f380 [Wei Song] Merge remote-tracking branch 'upstream/master'
a15a7c9a [Wei Song] Merge remote-tracking branch 'upstream/master'
5cbf9af9 [Wei Song] Merge remote-tracking branch 'upstream/master'
3f7ed71f [Wei Song] Added self to committer list
Project: http://git-wip-us.apache.org/repos/asf/samza/repo
Commit: http://git-wip-us.apache.org/repos/asf/samza/commit/2159171e
Tree: http://git-wip-us.apache.org/repos/asf/samza/tree/2159171e
Diff: http://git-wip-us.apache.org/repos/asf/samza/diff/2159171e
Branch: refs/heads/master
Commit: 2159171ee8df3f7266fe853a72cea4acd6a19d8e
Parents: ad12c6d
Author: Wei Song <ws...@linkedin.com>
Authored: Wed Nov 7 14:46:17 2018 -0800
Committer: Wei Song <ws...@linkedin.com>
Committed: Wed Nov 7 14:46:17 2018 -0800
----------------------------------------------------------------------
.../descriptors/InMemoryTableDescriptor.java | 4 +-
.../descriptors/InMemoryTableProvider.java | 4 +-
.../kv/descriptors/RocksDbTableDescriptor.java | 2 +-
.../kv/descriptors/RocksDbTableProvider.java | 2 +-
.../descriptors/TestRocksDbTableDescriptor.java | 8 +-
.../samza/storage/kv/LocalReadWriteTable.java | 149 ++++++++++++++++
.../samza/storage/kv/LocalReadableTable.java | 108 ++++++++++++
.../kv/LocalStoreBackedReadWriteTable.java | 149 ----------------
.../kv/LocalStoreBackedReadableTable.java | 108 ------------
.../BaseLocalStoreBackedTableDescriptor.java | 168 -------------------
.../BaseLocalStoreBackedTableProvider.java | 149 ----------------
.../descriptors/BaseLocalTableDescriptor.java | 168 +++++++++++++++++++
.../kv/descriptors/BaseLocalTableProvider.java | 149 ++++++++++++++++
.../TestBaseLocalStoreBackedTableProvider.java | 150 -----------------
.../descriptors/TestBaseLocalTableProvider.java | 150 +++++++++++++++++
.../apache/samza/test/table/TestLocalTable.java | 4 +-
16 files changed, 736 insertions(+), 736 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/samza/blob/2159171e/samza-kv-inmemory/src/main/java/org/apache/samza/storage/kv/inmemory/descriptors/InMemoryTableDescriptor.java
----------------------------------------------------------------------
diff --git a/samza-kv-inmemory/src/main/java/org/apache/samza/storage/kv/inmemory/descriptors/InMemoryTableDescriptor.java b/samza-kv-inmemory/src/main/java/org/apache/samza/storage/kv/inmemory/descriptors/InMemoryTableDescriptor.java
index c2d08ca..f785cd7 100644
--- a/samza-kv-inmemory/src/main/java/org/apache/samza/storage/kv/inmemory/descriptors/InMemoryTableDescriptor.java
+++ b/samza-kv-inmemory/src/main/java/org/apache/samza/storage/kv/inmemory/descriptors/InMemoryTableDescriptor.java
@@ -22,7 +22,7 @@ import java.util.HashMap;
import java.util.Map;
import org.apache.samza.serializers.KVSerde;
-import org.apache.samza.storage.kv.descriptors.BaseLocalStoreBackedTableDescriptor;
+import org.apache.samza.storage.kv.descriptors.BaseLocalTableDescriptor;
import org.apache.samza.table.TableSpec;
@@ -32,7 +32,7 @@ import org.apache.samza.table.TableSpec;
* @param <K> the type of the key
* @param <V> the type of the value
*/
-public class InMemoryTableDescriptor<K, V> extends BaseLocalStoreBackedTableDescriptor<K, V, InMemoryTableDescriptor<K, V>> {
+public class InMemoryTableDescriptor<K, V> extends BaseLocalTableDescriptor<K, V, InMemoryTableDescriptor<K, V>> {
/**
* Constructs a table descriptor instance
http://git-wip-us.apache.org/repos/asf/samza/blob/2159171e/samza-kv-inmemory/src/main/java/org/apache/samza/storage/kv/inmemory/descriptors/InMemoryTableProvider.java
----------------------------------------------------------------------
diff --git a/samza-kv-inmemory/src/main/java/org/apache/samza/storage/kv/inmemory/descriptors/InMemoryTableProvider.java b/samza-kv-inmemory/src/main/java/org/apache/samza/storage/kv/inmemory/descriptors/InMemoryTableProvider.java
index 448a316..d93bf75 100644
--- a/samza-kv-inmemory/src/main/java/org/apache/samza/storage/kv/inmemory/descriptors/InMemoryTableProvider.java
+++ b/samza-kv-inmemory/src/main/java/org/apache/samza/storage/kv/inmemory/descriptors/InMemoryTableProvider.java
@@ -24,7 +24,7 @@ import java.util.Map;
import org.apache.samza.config.Config;
import org.apache.samza.config.JavaTableConfig;
import org.apache.samza.config.StorageConfig;
-import org.apache.samza.storage.kv.descriptors.BaseLocalStoreBackedTableProvider;
+import org.apache.samza.storage.kv.descriptors.BaseLocalTableProvider;
import org.apache.samza.storage.kv.inmemory.InMemoryKeyValueStorageEngineFactory;
import org.apache.samza.table.TableSpec;
@@ -32,7 +32,7 @@ import org.apache.samza.table.TableSpec;
/**
* Table provider of an in-memory table
*/
-public class InMemoryTableProvider extends BaseLocalStoreBackedTableProvider {
+public class InMemoryTableProvider extends BaseLocalTableProvider {
public InMemoryTableProvider(TableSpec tableSpec) {
super(tableSpec);
http://git-wip-us.apache.org/repos/asf/samza/blob/2159171e/samza-kv-rocksdb/src/main/java/org/apache/samza/storage/kv/descriptors/RocksDbTableDescriptor.java
----------------------------------------------------------------------
diff --git a/samza-kv-rocksdb/src/main/java/org/apache/samza/storage/kv/descriptors/RocksDbTableDescriptor.java b/samza-kv-rocksdb/src/main/java/org/apache/samza/storage/kv/descriptors/RocksDbTableDescriptor.java
index a8ede78..8fc410f 100644
--- a/samza-kv-rocksdb/src/main/java/org/apache/samza/storage/kv/descriptors/RocksDbTableDescriptor.java
+++ b/samza-kv-rocksdb/src/main/java/org/apache/samza/storage/kv/descriptors/RocksDbTableDescriptor.java
@@ -31,7 +31,7 @@ import org.apache.samza.table.TableSpec;
* @param <K> the type of the key
* @param <V> the type of the value
*/
-public class RocksDbTableDescriptor<K, V> extends BaseLocalStoreBackedTableDescriptor<K, V, RocksDbTableDescriptor<K, V>> {
+public class RocksDbTableDescriptor<K, V> extends BaseLocalTableDescriptor<K, V, RocksDbTableDescriptor<K, V>> {
static final public String WRITE_BATCH_SIZE = "write.batch.size";
static final public String OBJECT_CACHE_SIZE = "object.cache.size";
http://git-wip-us.apache.org/repos/asf/samza/blob/2159171e/samza-kv-rocksdb/src/main/java/org/apache/samza/storage/kv/descriptors/RocksDbTableProvider.java
----------------------------------------------------------------------
diff --git a/samza-kv-rocksdb/src/main/java/org/apache/samza/storage/kv/descriptors/RocksDbTableProvider.java b/samza-kv-rocksdb/src/main/java/org/apache/samza/storage/kv/descriptors/RocksDbTableProvider.java
index e0c3355..c1c12ea 100644
--- a/samza-kv-rocksdb/src/main/java/org/apache/samza/storage/kv/descriptors/RocksDbTableProvider.java
+++ b/samza-kv-rocksdb/src/main/java/org/apache/samza/storage/kv/descriptors/RocksDbTableProvider.java
@@ -32,7 +32,7 @@ import org.apache.samza.table.TableSpec;
/**
* Table provider for tables backed by RocksDb.
*/
-public class RocksDbTableProvider extends BaseLocalStoreBackedTableProvider {
+public class RocksDbTableProvider extends BaseLocalTableProvider {
public RocksDbTableProvider(TableSpec tableSpec) {
super(tableSpec);
http://git-wip-us.apache.org/repos/asf/samza/blob/2159171e/samza-kv-rocksdb/src/test/java/org/apache/samza/storage/kv/descriptors/TestRocksDbTableDescriptor.java
----------------------------------------------------------------------
diff --git a/samza-kv-rocksdb/src/test/java/org/apache/samza/storage/kv/descriptors/TestRocksDbTableDescriptor.java b/samza-kv-rocksdb/src/test/java/org/apache/samza/storage/kv/descriptors/TestRocksDbTableDescriptor.java
index 86f6e5c..22558dc 100644
--- a/samza-kv-rocksdb/src/test/java/org/apache/samza/storage/kv/descriptors/TestRocksDbTableDescriptor.java
+++ b/samza-kv-rocksdb/src/test/java/org/apache/samza/storage/kv/descriptors/TestRocksDbTableDescriptor.java
@@ -80,7 +80,7 @@ public class TestRocksDbTableDescriptor {
Assert.assertEquals("snappy", getConfig(tableSpec, RocksDbTableDescriptor.ROCKSDB_COMPRESSION));
Assert.assertEquals("fifo", getConfig(tableSpec, RocksDbTableDescriptor.ROCKSDB_COMPACTION_STYLE));
Assert.assertEquals("xyz", getConfig(tableSpec, "abc"));
- Assert.assertEquals("false", tableSpec.getConfig().get(BaseLocalStoreBackedTableDescriptor.INTERNAL_ENABLE_CHANGELOG));
+ Assert.assertEquals("false", tableSpec.getConfig().get(BaseLocalTableDescriptor.INTERNAL_ENABLE_CHANGELOG));
}
@Test
@@ -91,9 +91,9 @@ public class TestRocksDbTableDescriptor {
.withChangelogReplicationFactor(10)
.getTableSpec();
- Assert.assertEquals("10", tableSpec.getConfig().get(BaseLocalStoreBackedTableDescriptor.INTERNAL_CHANGELOG_REPLICATION_FACTOR));
- Assert.assertEquals("changelog-$tream", tableSpec.getConfig().get(BaseLocalStoreBackedTableDescriptor.INTERNAL_CHANGELOG_STREAM));
- Assert.assertEquals("true", tableSpec.getConfig().get(BaseLocalStoreBackedTableDescriptor.INTERNAL_ENABLE_CHANGELOG));
+ Assert.assertEquals("10", tableSpec.getConfig().get(BaseLocalTableDescriptor.INTERNAL_CHANGELOG_REPLICATION_FACTOR));
+ Assert.assertEquals("changelog-$tream", tableSpec.getConfig().get(BaseLocalTableDescriptor.INTERNAL_CHANGELOG_STREAM));
+ Assert.assertEquals("true", tableSpec.getConfig().get(BaseLocalTableDescriptor.INTERNAL_ENABLE_CHANGELOG));
}
private String getConfig(TableSpec tableSpec, String key) {
http://git-wip-us.apache.org/repos/asf/samza/blob/2159171e/samza-kv/src/main/java/org/apache/samza/storage/kv/LocalReadWriteTable.java
----------------------------------------------------------------------
diff --git a/samza-kv/src/main/java/org/apache/samza/storage/kv/LocalReadWriteTable.java b/samza-kv/src/main/java/org/apache/samza/storage/kv/LocalReadWriteTable.java
new file mode 100644
index 0000000..d4a4aca
--- /dev/null
+++ b/samza-kv/src/main/java/org/apache/samza/storage/kv/LocalReadWriteTable.java
@@ -0,0 +1,149 @@
+/*
+ * 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.samza.storage.kv;
+
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import org.apache.samza.context.Context;
+import org.apache.samza.table.ReadWriteTable;
+import org.apache.samza.table.utils.DefaultTableWriteMetrics;
+
+
+/**
+ * A store backed readable and writable table
+ *
+ * @param <K> the type of the key in this table
+ * @param <V> the type of the value in this table
+ */
+public class LocalReadWriteTable<K, V> extends LocalReadableTable<K, V>
+ implements ReadWriteTable<K, V> {
+
+ protected DefaultTableWriteMetrics writeMetrics;
+
+ /**
+ * Constructs an instance of {@link LocalReadWriteTable}
+ * @param tableId the table Id
+ * @param kvStore the backing store
+ */
+ public LocalReadWriteTable(String tableId, KeyValueStore kvStore) {
+ super(tableId, kvStore);
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public void init(Context context) {
+ super.init(context);
+ writeMetrics = new DefaultTableWriteMetrics(context, this, tableId);
+ }
+
+ @Override
+ public void put(K key, V value) {
+ if (value != null) {
+ writeMetrics.numPuts.inc();
+ long startNs = System.nanoTime();
+ kvStore.put(key, value);
+ writeMetrics.putNs.update(System.nanoTime() - startNs);
+ } else {
+ delete(key);
+ }
+ }
+
+ @Override
+ public CompletableFuture<Void> putAsync(K key, V value) {
+ CompletableFuture<Void> future = new CompletableFuture();
+ try {
+ put(key, value);
+ future.complete(null);
+ } catch (Exception e) {
+ future.completeExceptionally(e);
+ }
+ return future;
+ }
+
+ @Override
+ public void putAll(List<Entry<K, V>> entries) {
+ writeMetrics.numPutAlls.inc();
+ long startNs = System.nanoTime();
+ kvStore.putAll(entries);
+ writeMetrics.putAllNs.update(System.nanoTime() - startNs);
+ }
+
+ @Override
+ public CompletableFuture<Void> putAllAsync(List<Entry<K, V>> entries) {
+ CompletableFuture<Void> future = new CompletableFuture();
+ try {
+ putAll(entries);
+ future.complete(null);
+ } catch (Exception e) {
+ future.completeExceptionally(e);
+ }
+ return future;
+ }
+
+ @Override
+ public void delete(K key) {
+ writeMetrics.numDeletes.inc();
+ long startNs = System.nanoTime();
+ kvStore.delete(key);
+ writeMetrics.deleteNs.update(System.nanoTime() - startNs);
+ }
+
+ @Override
+ public CompletableFuture<Void> deleteAsync(K key) {
+ CompletableFuture<Void> future = new CompletableFuture();
+ try {
+ delete(key);
+ future.complete(null);
+ } catch (Exception e) {
+ future.completeExceptionally(e);
+ }
+ return future;
+ }
+
+ @Override
+ public void deleteAll(List<K> keys) {
+ writeMetrics.numDeleteAlls.inc();
+ long startNs = System.nanoTime();
+ kvStore.deleteAll(keys);
+ writeMetrics.deleteAllNs.update(System.nanoTime() - startNs);
+ }
+
+ @Override
+ public CompletableFuture<Void> deleteAllAsync(List<K> keys) {
+ CompletableFuture<Void> future = new CompletableFuture();
+ try {
+ deleteAll(keys);
+ future.complete(null);
+ } catch (Exception e) {
+ future.completeExceptionally(e);
+ }
+ return future;
+ }
+
+ @Override
+ public void flush() {
+ writeMetrics.numFlushes.inc();
+ long startNs = System.nanoTime();
+ kvStore.flush();
+ writeMetrics.flushNs.update(System.nanoTime() - startNs);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/samza/blob/2159171e/samza-kv/src/main/java/org/apache/samza/storage/kv/LocalReadableTable.java
----------------------------------------------------------------------
diff --git a/samza-kv/src/main/java/org/apache/samza/storage/kv/LocalReadableTable.java b/samza-kv/src/main/java/org/apache/samza/storage/kv/LocalReadableTable.java
new file mode 100644
index 0000000..4eed288
--- /dev/null
+++ b/samza-kv/src/main/java/org/apache/samza/storage/kv/LocalReadableTable.java
@@ -0,0 +1,108 @@
+/*
+ * 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.samza.storage.kv;
+
+import com.google.common.base.Preconditions;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import org.apache.samza.context.Context;
+import org.apache.samza.table.ReadableTable;
+import org.apache.samza.table.utils.DefaultTableReadMetrics;
+
+
+/**
+ * A store backed readable table
+ *
+ * @param <K> the type of the key in this table
+ * @param <V> the type of the value in this table
+ */
+public class LocalReadableTable<K, V> implements ReadableTable<K, V> {
+
+ protected final KeyValueStore<K, V> kvStore;
+ protected final String tableId;
+
+ protected DefaultTableReadMetrics readMetrics;
+
+ /**
+ * Constructs an instance of {@link LocalReadableTable}
+ * @param tableId the table Id
+ * @param kvStore the backing store
+ */
+ public LocalReadableTable(String tableId, KeyValueStore<K, V> kvStore) {
+ Preconditions.checkArgument(tableId != null & !tableId.isEmpty() , "invalid tableId");
+ Preconditions.checkNotNull(kvStore, "null KeyValueStore");
+ this.tableId = tableId;
+ this.kvStore = kvStore;
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public void init(Context context) {
+ readMetrics = new DefaultTableReadMetrics(context, this, tableId);
+ }
+
+ @Override
+ public V get(K key) {
+ readMetrics.numGets.inc();
+ long startNs = System.nanoTime();
+ V result = kvStore.get(key);
+ readMetrics.getNs.update(System.nanoTime() - startNs);
+ return result;
+ }
+
+ @Override
+ public CompletableFuture<V> getAsync(K key) {
+ CompletableFuture<V> future = new CompletableFuture();
+ try {
+ future.complete(get(key));
+ } catch (Exception e) {
+ future.completeExceptionally(e);
+ }
+ return future;
+ }
+
+ @Override
+ public Map<K, V> getAll(List<K> keys) {
+ readMetrics.numGetAlls.inc();
+ long startNs = System.nanoTime();
+ Map<K, V> result = kvStore.getAll(keys);
+ readMetrics.getAllNs.update(System.nanoTime() - startNs);
+ return result;
+ }
+
+ @Override
+ public CompletableFuture<Map<K, V>> getAllAsync(List<K> keys) {
+ CompletableFuture<Map<K, V>> future = new CompletableFuture();
+ try {
+ future.complete(getAll(keys));
+ } catch (Exception e) {
+ future.completeExceptionally(e);
+ }
+ return future;
+ }
+
+ @Override
+ public void close() {
+ // The KV store is not closed here as it may still be needed by downstream operators,
+ // it will be closed by the SamzaContainer
+ }
+}
http://git-wip-us.apache.org/repos/asf/samza/blob/2159171e/samza-kv/src/main/java/org/apache/samza/storage/kv/LocalStoreBackedReadWriteTable.java
----------------------------------------------------------------------
diff --git a/samza-kv/src/main/java/org/apache/samza/storage/kv/LocalStoreBackedReadWriteTable.java b/samza-kv/src/main/java/org/apache/samza/storage/kv/LocalStoreBackedReadWriteTable.java
deleted file mode 100644
index 804df43..0000000
--- a/samza-kv/src/main/java/org/apache/samza/storage/kv/LocalStoreBackedReadWriteTable.java
+++ /dev/null
@@ -1,149 +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.samza.storage.kv;
-
-import java.util.List;
-import java.util.concurrent.CompletableFuture;
-import org.apache.samza.context.Context;
-import org.apache.samza.table.ReadWriteTable;
-import org.apache.samza.table.utils.DefaultTableWriteMetrics;
-
-
-/**
- * A store backed readable and writable table
- *
- * @param <K> the type of the key in this table
- * @param <V> the type of the value in this table
- */
-public class LocalStoreBackedReadWriteTable<K, V> extends LocalStoreBackedReadableTable<K, V>
- implements ReadWriteTable<K, V> {
-
- protected DefaultTableWriteMetrics writeMetrics;
-
- /**
- * Constructs an instance of {@link LocalStoreBackedReadWriteTable}
- * @param tableId the table Id
- * @param kvStore the backing store
- */
- public LocalStoreBackedReadWriteTable(String tableId, KeyValueStore kvStore) {
- super(tableId, kvStore);
- }
-
- /**
- * {@inheritDoc}
- */
- @Override
- public void init(Context context) {
- super.init(context);
- writeMetrics = new DefaultTableWriteMetrics(context, this, tableId);
- }
-
- @Override
- public void put(K key, V value) {
- if (value != null) {
- writeMetrics.numPuts.inc();
- long startNs = System.nanoTime();
- kvStore.put(key, value);
- writeMetrics.putNs.update(System.nanoTime() - startNs);
- } else {
- delete(key);
- }
- }
-
- @Override
- public CompletableFuture<Void> putAsync(K key, V value) {
- CompletableFuture<Void> future = new CompletableFuture();
- try {
- put(key, value);
- future.complete(null);
- } catch (Exception e) {
- future.completeExceptionally(e);
- }
- return future;
- }
-
- @Override
- public void putAll(List<Entry<K, V>> entries) {
- writeMetrics.numPutAlls.inc();
- long startNs = System.nanoTime();
- kvStore.putAll(entries);
- writeMetrics.putAllNs.update(System.nanoTime() - startNs);
- }
-
- @Override
- public CompletableFuture<Void> putAllAsync(List<Entry<K, V>> entries) {
- CompletableFuture<Void> future = new CompletableFuture();
- try {
- putAll(entries);
- future.complete(null);
- } catch (Exception e) {
- future.completeExceptionally(e);
- }
- return future;
- }
-
- @Override
- public void delete(K key) {
- writeMetrics.numDeletes.inc();
- long startNs = System.nanoTime();
- kvStore.delete(key);
- writeMetrics.deleteNs.update(System.nanoTime() - startNs);
- }
-
- @Override
- public CompletableFuture<Void> deleteAsync(K key) {
- CompletableFuture<Void> future = new CompletableFuture();
- try {
- delete(key);
- future.complete(null);
- } catch (Exception e) {
- future.completeExceptionally(e);
- }
- return future;
- }
-
- @Override
- public void deleteAll(List<K> keys) {
- writeMetrics.numDeleteAlls.inc();
- long startNs = System.nanoTime();
- kvStore.deleteAll(keys);
- writeMetrics.deleteAllNs.update(System.nanoTime() - startNs);
- }
-
- @Override
- public CompletableFuture<Void> deleteAllAsync(List<K> keys) {
- CompletableFuture<Void> future = new CompletableFuture();
- try {
- deleteAll(keys);
- future.complete(null);
- } catch (Exception e) {
- future.completeExceptionally(e);
- }
- return future;
- }
-
- @Override
- public void flush() {
- writeMetrics.numFlushes.inc();
- long startNs = System.nanoTime();
- kvStore.flush();
- writeMetrics.flushNs.update(System.nanoTime() - startNs);
- }
-
-}
http://git-wip-us.apache.org/repos/asf/samza/blob/2159171e/samza-kv/src/main/java/org/apache/samza/storage/kv/LocalStoreBackedReadableTable.java
----------------------------------------------------------------------
diff --git a/samza-kv/src/main/java/org/apache/samza/storage/kv/LocalStoreBackedReadableTable.java b/samza-kv/src/main/java/org/apache/samza/storage/kv/LocalStoreBackedReadableTable.java
deleted file mode 100644
index d440d42..0000000
--- a/samza-kv/src/main/java/org/apache/samza/storage/kv/LocalStoreBackedReadableTable.java
+++ /dev/null
@@ -1,108 +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.samza.storage.kv;
-
-import com.google.common.base.Preconditions;
-import java.util.List;
-import java.util.Map;
-import java.util.concurrent.CompletableFuture;
-import org.apache.samza.context.Context;
-import org.apache.samza.table.ReadableTable;
-import org.apache.samza.table.utils.DefaultTableReadMetrics;
-
-
-/**
- * A store backed readable table
- *
- * @param <K> the type of the key in this table
- * @param <V> the type of the value in this table
- */
-public class LocalStoreBackedReadableTable<K, V> implements ReadableTable<K, V> {
-
- protected final KeyValueStore<K, V> kvStore;
- protected final String tableId;
-
- protected DefaultTableReadMetrics readMetrics;
-
- /**
- * Constructs an instance of {@link LocalStoreBackedReadableTable}
- * @param tableId the table Id
- * @param kvStore the backing store
- */
- public LocalStoreBackedReadableTable(String tableId, KeyValueStore<K, V> kvStore) {
- Preconditions.checkArgument(tableId != null & !tableId.isEmpty() , "invalid tableId");
- Preconditions.checkNotNull(kvStore, "null KeyValueStore");
- this.tableId = tableId;
- this.kvStore = kvStore;
- }
-
- /**
- * {@inheritDoc}
- */
- @Override
- public void init(Context context) {
- readMetrics = new DefaultTableReadMetrics(context, this, tableId);
- }
-
- @Override
- public V get(K key) {
- readMetrics.numGets.inc();
- long startNs = System.nanoTime();
- V result = kvStore.get(key);
- readMetrics.getNs.update(System.nanoTime() - startNs);
- return result;
- }
-
- @Override
- public CompletableFuture<V> getAsync(K key) {
- CompletableFuture<V> future = new CompletableFuture();
- try {
- future.complete(get(key));
- } catch (Exception e) {
- future.completeExceptionally(e);
- }
- return future;
- }
-
- @Override
- public Map<K, V> getAll(List<K> keys) {
- readMetrics.numGetAlls.inc();
- long startNs = System.nanoTime();
- Map<K, V> result = kvStore.getAll(keys);
- readMetrics.getAllNs.update(System.nanoTime() - startNs);
- return result;
- }
-
- @Override
- public CompletableFuture<Map<K, V>> getAllAsync(List<K> keys) {
- CompletableFuture<Map<K, V>> future = new CompletableFuture();
- try {
- future.complete(getAll(keys));
- } catch (Exception e) {
- future.completeExceptionally(e);
- }
- return future;
- }
-
- @Override
- public void close() {
- // The KV store is not closed here as it may still be needed by downstream operators,
- // it will be closed by the SamzaContainer
- }
-}
http://git-wip-us.apache.org/repos/asf/samza/blob/2159171e/samza-kv/src/main/java/org/apache/samza/storage/kv/descriptors/BaseLocalStoreBackedTableDescriptor.java
----------------------------------------------------------------------
diff --git a/samza-kv/src/main/java/org/apache/samza/storage/kv/descriptors/BaseLocalStoreBackedTableDescriptor.java b/samza-kv/src/main/java/org/apache/samza/storage/kv/descriptors/BaseLocalStoreBackedTableDescriptor.java
deleted file mode 100644
index 7957fd3..0000000
--- a/samza-kv/src/main/java/org/apache/samza/storage/kv/descriptors/BaseLocalStoreBackedTableDescriptor.java
+++ /dev/null
@@ -1,168 +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.samza.storage.kv.descriptors;
-
-import com.google.common.base.Preconditions;
-
-import java.util.List;
-import java.util.Map;
-
-import org.apache.samza.table.descriptors.BaseTableDescriptor;
-import org.apache.samza.serializers.KVSerde;
-import org.apache.samza.storage.SideInputsProcessor;
-
-
-/**
- * Table descriptor for store backed tables.
- *
- * @param <K> the type of the key in this table
- * @param <V> the type of the value in this table
- * @param <D> the type of the concrete table descriptor
- */
-abstract public class BaseLocalStoreBackedTableDescriptor<K, V, D extends BaseLocalStoreBackedTableDescriptor<K, V, D>>
- extends BaseTableDescriptor<K, V, D> {
-
- static final public String INTERNAL_ENABLE_CHANGELOG = "internal.enable.changelog";
- static final public String INTERNAL_CHANGELOG_STREAM = "internal.changelog.stream";
- static final public String INTERNAL_CHANGELOG_REPLICATION_FACTOR = "internal.changelog.replication.factor";
-
- protected List<String> sideInputs;
- protected SideInputsProcessor sideInputsProcessor;
- protected boolean enableChangelog;
- protected String changelogStream;
- protected Integer changelogReplicationFactor;
-
- /**
- * Constructs a table descriptor instance
- * @param tableId Id of the table, it must conform to pattern {@literal [\\d\\w-_]+}
- */
- public BaseLocalStoreBackedTableDescriptor(String tableId) {
- super(tableId);
- }
-
- /**
- * Constructs a table descriptor instance
- * @param tableId Id of the table, it must conform to pattern {@literal [\\d\\w-_]+}
- * @param serde the serde for key and value
- */
- public BaseLocalStoreBackedTableDescriptor(String tableId, KVSerde<K, V> serde) {
- super(tableId, serde);
- }
-
- public D withSideInputs(List<String> sideInputs) {
- this.sideInputs = sideInputs;
- // Disable changelog
- this.enableChangelog = false;
- this.changelogStream = null;
- this.changelogReplicationFactor = null;
- return (D) this;
- }
-
- public D withSideInputsProcessor(SideInputsProcessor sideInputsProcessor) {
- this.sideInputsProcessor = sideInputsProcessor;
- return (D) this;
- }
-
- /**
- * Enable changelog for this table, by default changelog is disabled. When the
- * changelog stream name is not specified, it is automatically generated in
- * the format {@literal [job-name]-[job-id]-table-[table-id]}.
- * Refer to <code>stores.store-name.changelog</code> in Samza configuration guide
- *
- * @return this table descriptor instance
- */
- public D withChangelogEnabled() {
- this.enableChangelog = true;
- return (D) this;
- }
-
- /**
- * Samza stores are local to a container. If the container fails, the contents of
- * the store are lost. To prevent loss of data, you need to set this property to
- * configure a changelog stream: Samza then ensures that writes to the store are
- * replicated to this stream, and the store is restored from this stream after a
- * failure. The value of this property is given in the form system-name.stream-name.
- * The "system-name" part is optional. If it is omitted you must specify the system
- * in <code>job.changelog.system</code> config. Any output stream can be used as
- * changelog, but you must ensure that only one job ever writes to a given changelog
- * stream (each instance of a job and each store needs its own changelog stream).
- * <p>
- * Refer to <code>stores.store-name.changelog</code> in Samza configuration guide
- *
- * @param changelogStream changelog stream name
- * @return this table descriptor instance
- */
- public D withChangelogStream(String changelogStream) {
- this.enableChangelog = true;
- this.changelogStream = changelogStream;
- return (D) this;
- }
-
- /**
- * The property defines the number of replicas to use for the change log stream.
- * <p>
- * Default value is <code>stores.default.changelog.replication.factor</code>.
- * <p>
- * Refer to <code>stores.store-name.changelog.replication.factor</code> in Samza configuration guide
- *
- * @param replicationFactor replication factor
- * @return this table descriptor instance
- */
- public D withChangelogReplicationFactor(int replicationFactor) {
- this.enableChangelog = true;
- this.changelogReplicationFactor = replicationFactor;
- return (D) this;
- }
-
- @Override
- protected void generateTableSpecConfig(Map<String, String> tableSpecConfig) {
- super.generateTableSpecConfig(tableSpecConfig);
-
- tableSpecConfig.put(INTERNAL_ENABLE_CHANGELOG, String.valueOf(enableChangelog));
- if (enableChangelog) {
- if (changelogStream != null) {
- tableSpecConfig.put(INTERNAL_CHANGELOG_STREAM, changelogStream);
- }
- if (changelogReplicationFactor != null) {
- tableSpecConfig.put(INTERNAL_CHANGELOG_REPLICATION_FACTOR, String.valueOf(changelogReplicationFactor));
- }
- }
- }
-
- /**
- * Validate that this table descriptor is constructed properly
- */
- protected void validate() {
- super.validate();
- if (sideInputs != null || sideInputsProcessor != null) {
- Preconditions.checkArgument(sideInputs != null && !sideInputs.isEmpty() && sideInputsProcessor != null,
- String.format("Invalid side input configuration for table: %s. " +
- "Both side inputs and the processor must be provided", tableId));
- }
- if (!enableChangelog) {
- Preconditions.checkState(changelogStream == null,
- String.format("Invalid changelog configuration for table: %s. Changelog " +
- "must be enabled, when changelog stream name is provided", tableId));
- Preconditions.checkState(changelogReplicationFactor == null,
- String.format("Invalid changelog configuration for table: %s. Changelog " +
- "must be enabled, when changelog replication factor is provided", tableId));
- }
- }
-
-}
http://git-wip-us.apache.org/repos/asf/samza/blob/2159171e/samza-kv/src/main/java/org/apache/samza/storage/kv/descriptors/BaseLocalStoreBackedTableProvider.java
----------------------------------------------------------------------
diff --git a/samza-kv/src/main/java/org/apache/samza/storage/kv/descriptors/BaseLocalStoreBackedTableProvider.java b/samza-kv/src/main/java/org/apache/samza/storage/kv/descriptors/BaseLocalStoreBackedTableProvider.java
deleted file mode 100644
index f5dd71b..0000000
--- a/samza-kv/src/main/java/org/apache/samza/storage/kv/descriptors/BaseLocalStoreBackedTableProvider.java
+++ /dev/null
@@ -1,149 +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.samza.storage.kv.descriptors;
-
-import com.google.common.base.Preconditions;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.regex.Pattern;
-import org.apache.commons.lang3.StringUtils;
-import org.apache.samza.SamzaException;
-import org.apache.samza.config.Config;
-import org.apache.samza.config.JavaStorageConfig;
-import org.apache.samza.config.JavaTableConfig;
-import org.apache.samza.config.JobConfig;
-import org.apache.samza.config.MapConfig;
-import org.apache.samza.config.StorageConfig;
-import org.apache.samza.context.Context;
-import org.apache.samza.storage.kv.KeyValueStore;
-import org.apache.samza.storage.kv.LocalStoreBackedReadWriteTable;
-import org.apache.samza.table.ReadableTable;
-import org.apache.samza.table.Table;
-import org.apache.samza.table.TableSpec;
-import org.apache.samza.table.utils.descriptors.BaseTableProvider;
-import org.apache.samza.table.utils.SerdeUtils;
-
-
-/**
- * Base class for tables backed by Samza local stores. The backing stores are
- * injected during initialization of the table. Since the lifecycle
- * of the underlying stores are already managed by Samza container,
- * the table provider will not manage the lifecycle of the backing
- * stores.
- */
-abstract public class BaseLocalStoreBackedTableProvider extends BaseTableProvider {
- public static final Pattern SYSTEM_STREAM_NAME_PATTERN = Pattern.compile("[\\d\\w-_.]+");
-
- protected KeyValueStore kvStore;
-
- public BaseLocalStoreBackedTableProvider(TableSpec tableSpec) {
- super(tableSpec);
- }
-
- @Override
- public void init(Context context) {
- super.init(context);
-
- Preconditions.checkNotNull(this.context, "Must specify context for local tables.");
-
- kvStore = (KeyValueStore) this.context.getTaskContext().getStore(tableSpec.getId());
-
- if (kvStore == null) {
- throw new SamzaException(String.format(
- "Backing store for table %s was not injected by SamzaContainer", tableSpec.getId()));
- }
-
- logger.info("Initialized backing store for table " + tableSpec.getId());
- }
-
- @Override
- public Table getTable() {
- if (kvStore == null) {
- throw new SamzaException("Store not initialized for table " + tableSpec.getId());
- }
- ReadableTable table = new LocalStoreBackedReadWriteTable(tableSpec.getId(), kvStore);
- table.init(this.context);
- return table;
- }
-
- protected Map<String, String> generateCommonStoreConfig(Config jobConfig, Map<String, String> generatedConfig) {
-
- Map<String, String> storeConfig = new HashMap<>();
-
- // serde configurations for tables are generated at top level by JobNodeConfigurationGenerator and are included
- // in the global jobConfig. generatedConfig has all table specific configuration generated from TableSpec, such
- // as TableProviderFactory, sideInputs, etc.
- // Merge the global jobConfig and generatedConfig to get full access to configuration needed to create local
- // store configuration
- Map<String, String> mergedConfigMap = new HashMap<>(jobConfig);
- mergedConfigMap.putAll(generatedConfig);
- JobConfig mergedJobConfig = new JobConfig(new MapConfig(mergedConfigMap));
- JavaTableConfig tableConfig = new JavaTableConfig(mergedJobConfig);
-
- String keySerde = tableConfig.getKeySerde(tableSpec.getId());
- storeConfig.put(String.format(StorageConfig.KEY_SERDE(), tableSpec.getId()), keySerde);
-
- String valueSerde = tableConfig.getValueSerde(tableSpec.getId());
- storeConfig.put(String.format(StorageConfig.MSG_SERDE(), tableSpec.getId()), valueSerde);
-
- List<String> sideInputs = tableSpec.getSideInputs();
- if (sideInputs != null && !sideInputs.isEmpty()) {
- sideInputs.forEach(si -> Preconditions.checkState(isValidSystemStreamName(si), String.format(
- "Side input stream %s doesn't confirm to pattern %s", si, SYSTEM_STREAM_NAME_PATTERN)));
- String formattedSideInputs = String.join(",", sideInputs);
- storeConfig.put(String.format(JavaStorageConfig.SIDE_INPUTS, tableSpec.getId()), formattedSideInputs);
- storeConfig.put(String.format(JavaStorageConfig.SIDE_INPUTS_PROCESSOR_SERIALIZED_INSTANCE, tableSpec.getId()),
- SerdeUtils.serialize("Side Inputs Processor", tableSpec.getSideInputsProcessor()));
- }
-
- // Changelog configuration
- Boolean enableChangelog = Boolean.valueOf(
- tableSpec.getConfig().get(BaseLocalStoreBackedTableDescriptor.INTERNAL_ENABLE_CHANGELOG));
- if (enableChangelog) {
- String changelogStream = tableSpec.getConfig().get(BaseLocalStoreBackedTableDescriptor.INTERNAL_CHANGELOG_STREAM);
- if (StringUtils.isEmpty(changelogStream)) {
- changelogStream = String.format("%s-%s-table-%s", mergedJobConfig.getName().get(), mergedJobConfig.getJobId(),
- tableSpec.getId());
- }
-
- Preconditions.checkState(isValidSystemStreamName(changelogStream), String.format(
- "Changelog stream %s doesn't confirm to pattern %s", changelogStream, SYSTEM_STREAM_NAME_PATTERN));
- storeConfig.put(String.format(StorageConfig.CHANGELOG_STREAM(), tableSpec.getId()), changelogStream);
-
- String changelogReplicationFactor = tableSpec.getConfig().get(
- BaseLocalStoreBackedTableDescriptor.INTERNAL_CHANGELOG_REPLICATION_FACTOR);
- if (changelogReplicationFactor != null) {
- storeConfig.put(String.format(StorageConfig.CHANGELOG_REPLICATION_FACTOR(), tableSpec.getId()),
- changelogReplicationFactor);
- }
- }
-
- return storeConfig;
- }
-
- @Override
- public void close() {
- logger.info("Shutting down table provider for table " + tableSpec.getId());
- }
-
- private boolean isValidSystemStreamName(String name) {
- return StringUtils.isNotBlank(name) && SYSTEM_STREAM_NAME_PATTERN.matcher(name).matches();
- }
-}
http://git-wip-us.apache.org/repos/asf/samza/blob/2159171e/samza-kv/src/main/java/org/apache/samza/storage/kv/descriptors/BaseLocalTableDescriptor.java
----------------------------------------------------------------------
diff --git a/samza-kv/src/main/java/org/apache/samza/storage/kv/descriptors/BaseLocalTableDescriptor.java b/samza-kv/src/main/java/org/apache/samza/storage/kv/descriptors/BaseLocalTableDescriptor.java
new file mode 100644
index 0000000..bd80691
--- /dev/null
+++ b/samza-kv/src/main/java/org/apache/samza/storage/kv/descriptors/BaseLocalTableDescriptor.java
@@ -0,0 +1,168 @@
+/*
+ * 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.samza.storage.kv.descriptors;
+
+import com.google.common.base.Preconditions;
+
+import java.util.List;
+import java.util.Map;
+
+import org.apache.samza.table.descriptors.BaseTableDescriptor;
+import org.apache.samza.serializers.KVSerde;
+import org.apache.samza.storage.SideInputsProcessor;
+
+
+/**
+ * Table descriptor for store backed tables.
+ *
+ * @param <K> the type of the key in this table
+ * @param <V> the type of the value in this table
+ * @param <D> the type of the concrete table descriptor
+ */
+abstract public class BaseLocalTableDescriptor<K, V, D extends BaseLocalTableDescriptor<K, V, D>>
+ extends BaseTableDescriptor<K, V, D> {
+
+ static final public String INTERNAL_ENABLE_CHANGELOG = "internal.enable.changelog";
+ static final public String INTERNAL_CHANGELOG_STREAM = "internal.changelog.stream";
+ static final public String INTERNAL_CHANGELOG_REPLICATION_FACTOR = "internal.changelog.replication.factor";
+
+ protected List<String> sideInputs;
+ protected SideInputsProcessor sideInputsProcessor;
+ protected boolean enableChangelog;
+ protected String changelogStream;
+ protected Integer changelogReplicationFactor;
+
+ /**
+ * Constructs a table descriptor instance
+ * @param tableId Id of the table, it must conform to pattern {@literal [\\d\\w-_]+}
+ */
+ public BaseLocalTableDescriptor(String tableId) {
+ super(tableId);
+ }
+
+ /**
+ * Constructs a table descriptor instance
+ * @param tableId Id of the table, it must conform to pattern {@literal [\\d\\w-_]+}
+ * @param serde the serde for key and value
+ */
+ public BaseLocalTableDescriptor(String tableId, KVSerde<K, V> serde) {
+ super(tableId, serde);
+ }
+
+ public D withSideInputs(List<String> sideInputs) {
+ this.sideInputs = sideInputs;
+ // Disable changelog
+ this.enableChangelog = false;
+ this.changelogStream = null;
+ this.changelogReplicationFactor = null;
+ return (D) this;
+ }
+
+ public D withSideInputsProcessor(SideInputsProcessor sideInputsProcessor) {
+ this.sideInputsProcessor = sideInputsProcessor;
+ return (D) this;
+ }
+
+ /**
+ * Enable changelog for this table, by default changelog is disabled. When the
+ * changelog stream name is not specified, it is automatically generated in
+ * the format {@literal [job-name]-[job-id]-table-[table-id]}.
+ * Refer to <code>stores.store-name.changelog</code> in Samza configuration guide
+ *
+ * @return this table descriptor instance
+ */
+ public D withChangelogEnabled() {
+ this.enableChangelog = true;
+ return (D) this;
+ }
+
+ /**
+ * Samza stores are local to a container. If the container fails, the contents of
+ * the store are lost. To prevent loss of data, you need to set this property to
+ * configure a changelog stream: Samza then ensures that writes to the store are
+ * replicated to this stream, and the store is restored from this stream after a
+ * failure. The value of this property is given in the form system-name.stream-name.
+ * The "system-name" part is optional. If it is omitted you must specify the system
+ * in <code>job.changelog.system</code> config. Any output stream can be used as
+ * changelog, but you must ensure that only one job ever writes to a given changelog
+ * stream (each instance of a job and each store needs its own changelog stream).
+ * <p>
+ * Refer to <code>stores.store-name.changelog</code> in Samza configuration guide
+ *
+ * @param changelogStream changelog stream name
+ * @return this table descriptor instance
+ */
+ public D withChangelogStream(String changelogStream) {
+ this.enableChangelog = true;
+ this.changelogStream = changelogStream;
+ return (D) this;
+ }
+
+ /**
+ * The property defines the number of replicas to use for the change log stream.
+ * <p>
+ * Default value is <code>stores.default.changelog.replication.factor</code>.
+ * <p>
+ * Refer to <code>stores.store-name.changelog.replication.factor</code> in Samza configuration guide
+ *
+ * @param replicationFactor replication factor
+ * @return this table descriptor instance
+ */
+ public D withChangelogReplicationFactor(int replicationFactor) {
+ this.enableChangelog = true;
+ this.changelogReplicationFactor = replicationFactor;
+ return (D) this;
+ }
+
+ @Override
+ protected void generateTableSpecConfig(Map<String, String> tableSpecConfig) {
+ super.generateTableSpecConfig(tableSpecConfig);
+
+ tableSpecConfig.put(INTERNAL_ENABLE_CHANGELOG, String.valueOf(enableChangelog));
+ if (enableChangelog) {
+ if (changelogStream != null) {
+ tableSpecConfig.put(INTERNAL_CHANGELOG_STREAM, changelogStream);
+ }
+ if (changelogReplicationFactor != null) {
+ tableSpecConfig.put(INTERNAL_CHANGELOG_REPLICATION_FACTOR, String.valueOf(changelogReplicationFactor));
+ }
+ }
+ }
+
+ /**
+ * Validate that this table descriptor is constructed properly
+ */
+ protected void validate() {
+ super.validate();
+ if (sideInputs != null || sideInputsProcessor != null) {
+ Preconditions.checkArgument(sideInputs != null && !sideInputs.isEmpty() && sideInputsProcessor != null,
+ String.format("Invalid side input configuration for table: %s. " +
+ "Both side inputs and the processor must be provided", tableId));
+ }
+ if (!enableChangelog) {
+ Preconditions.checkState(changelogStream == null,
+ String.format("Invalid changelog configuration for table: %s. Changelog " +
+ "must be enabled, when changelog stream name is provided", tableId));
+ Preconditions.checkState(changelogReplicationFactor == null,
+ String.format("Invalid changelog configuration for table: %s. Changelog " +
+ "must be enabled, when changelog replication factor is provided", tableId));
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/samza/blob/2159171e/samza-kv/src/main/java/org/apache/samza/storage/kv/descriptors/BaseLocalTableProvider.java
----------------------------------------------------------------------
diff --git a/samza-kv/src/main/java/org/apache/samza/storage/kv/descriptors/BaseLocalTableProvider.java b/samza-kv/src/main/java/org/apache/samza/storage/kv/descriptors/BaseLocalTableProvider.java
new file mode 100644
index 0000000..699b82b
--- /dev/null
+++ b/samza-kv/src/main/java/org/apache/samza/storage/kv/descriptors/BaseLocalTableProvider.java
@@ -0,0 +1,149 @@
+/*
+ * 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.samza.storage.kv.descriptors;
+
+import com.google.common.base.Preconditions;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.regex.Pattern;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.samza.SamzaException;
+import org.apache.samza.config.Config;
+import org.apache.samza.config.JavaStorageConfig;
+import org.apache.samza.config.JavaTableConfig;
+import org.apache.samza.config.JobConfig;
+import org.apache.samza.config.MapConfig;
+import org.apache.samza.config.StorageConfig;
+import org.apache.samza.context.Context;
+import org.apache.samza.storage.kv.KeyValueStore;
+import org.apache.samza.storage.kv.LocalReadWriteTable;
+import org.apache.samza.table.ReadableTable;
+import org.apache.samza.table.Table;
+import org.apache.samza.table.TableSpec;
+import org.apache.samza.table.utils.descriptors.BaseTableProvider;
+import org.apache.samza.table.utils.SerdeUtils;
+
+
+/**
+ * Base class for tables backed by Samza local stores. The backing stores are
+ * injected during initialization of the table. Since the lifecycle
+ * of the underlying stores are already managed by Samza container,
+ * the table provider will not manage the lifecycle of the backing
+ * stores.
+ */
+abstract public class BaseLocalTableProvider extends BaseTableProvider {
+ public static final Pattern SYSTEM_STREAM_NAME_PATTERN = Pattern.compile("[\\d\\w-_.]+");
+
+ protected KeyValueStore kvStore;
+
+ public BaseLocalTableProvider(TableSpec tableSpec) {
+ super(tableSpec);
+ }
+
+ @Override
+ public void init(Context context) {
+ super.init(context);
+
+ Preconditions.checkNotNull(this.context, "Must specify context for local tables.");
+
+ kvStore = (KeyValueStore) this.context.getTaskContext().getStore(tableSpec.getId());
+
+ if (kvStore == null) {
+ throw new SamzaException(String.format(
+ "Backing store for table %s was not injected by SamzaContainer", tableSpec.getId()));
+ }
+
+ logger.info("Initialized backing store for table " + tableSpec.getId());
+ }
+
+ @Override
+ public Table getTable() {
+ if (kvStore == null) {
+ throw new SamzaException("Store not initialized for table " + tableSpec.getId());
+ }
+ ReadableTable table = new LocalReadWriteTable(tableSpec.getId(), kvStore);
+ table.init(this.context);
+ return table;
+ }
+
+ protected Map<String, String> generateCommonStoreConfig(Config jobConfig, Map<String, String> generatedConfig) {
+
+ Map<String, String> storeConfig = new HashMap<>();
+
+ // serde configurations for tables are generated at top level by JobNodeConfigurationGenerator and are included
+ // in the global jobConfig. generatedConfig has all table specific configuration generated from TableSpec, such
+ // as TableProviderFactory, sideInputs, etc.
+ // Merge the global jobConfig and generatedConfig to get full access to configuration needed to create local
+ // store configuration
+ Map<String, String> mergedConfigMap = new HashMap<>(jobConfig);
+ mergedConfigMap.putAll(generatedConfig);
+ JobConfig mergedJobConfig = new JobConfig(new MapConfig(mergedConfigMap));
+ JavaTableConfig tableConfig = new JavaTableConfig(mergedJobConfig);
+
+ String keySerde = tableConfig.getKeySerde(tableSpec.getId());
+ storeConfig.put(String.format(StorageConfig.KEY_SERDE(), tableSpec.getId()), keySerde);
+
+ String valueSerde = tableConfig.getValueSerde(tableSpec.getId());
+ storeConfig.put(String.format(StorageConfig.MSG_SERDE(), tableSpec.getId()), valueSerde);
+
+ List<String> sideInputs = tableSpec.getSideInputs();
+ if (sideInputs != null && !sideInputs.isEmpty()) {
+ sideInputs.forEach(si -> Preconditions.checkState(isValidSystemStreamName(si), String.format(
+ "Side input stream %s doesn't confirm to pattern %s", si, SYSTEM_STREAM_NAME_PATTERN)));
+ String formattedSideInputs = String.join(",", sideInputs);
+ storeConfig.put(String.format(JavaStorageConfig.SIDE_INPUTS, tableSpec.getId()), formattedSideInputs);
+ storeConfig.put(String.format(JavaStorageConfig.SIDE_INPUTS_PROCESSOR_SERIALIZED_INSTANCE, tableSpec.getId()),
+ SerdeUtils.serialize("Side Inputs Processor", tableSpec.getSideInputsProcessor()));
+ }
+
+ // Changelog configuration
+ Boolean enableChangelog = Boolean.valueOf(
+ tableSpec.getConfig().get(BaseLocalTableDescriptor.INTERNAL_ENABLE_CHANGELOG));
+ if (enableChangelog) {
+ String changelogStream = tableSpec.getConfig().get(BaseLocalTableDescriptor.INTERNAL_CHANGELOG_STREAM);
+ if (StringUtils.isEmpty(changelogStream)) {
+ changelogStream = String.format("%s-%s-table-%s", mergedJobConfig.getName().get(), mergedJobConfig.getJobId(),
+ tableSpec.getId());
+ }
+
+ Preconditions.checkState(isValidSystemStreamName(changelogStream), String.format(
+ "Changelog stream %s doesn't confirm to pattern %s", changelogStream, SYSTEM_STREAM_NAME_PATTERN));
+ storeConfig.put(String.format(StorageConfig.CHANGELOG_STREAM(), tableSpec.getId()), changelogStream);
+
+ String changelogReplicationFactor = tableSpec.getConfig().get(
+ BaseLocalTableDescriptor.INTERNAL_CHANGELOG_REPLICATION_FACTOR);
+ if (changelogReplicationFactor != null) {
+ storeConfig.put(String.format(StorageConfig.CHANGELOG_REPLICATION_FACTOR(), tableSpec.getId()),
+ changelogReplicationFactor);
+ }
+ }
+
+ return storeConfig;
+ }
+
+ @Override
+ public void close() {
+ logger.info("Shutting down table provider for table " + tableSpec.getId());
+ }
+
+ private boolean isValidSystemStreamName(String name) {
+ return StringUtils.isNotBlank(name) && SYSTEM_STREAM_NAME_PATTERN.matcher(name).matches();
+ }
+}
http://git-wip-us.apache.org/repos/asf/samza/blob/2159171e/samza-kv/src/test/java/org/apache/samza/storage/kv/descriptors/TestBaseLocalStoreBackedTableProvider.java
----------------------------------------------------------------------
diff --git a/samza-kv/src/test/java/org/apache/samza/storage/kv/descriptors/TestBaseLocalStoreBackedTableProvider.java b/samza-kv/src/test/java/org/apache/samza/storage/kv/descriptors/TestBaseLocalStoreBackedTableProvider.java
deleted file mode 100644
index 559f2e9..0000000
--- a/samza-kv/src/test/java/org/apache/samza/storage/kv/descriptors/TestBaseLocalStoreBackedTableProvider.java
+++ /dev/null
@@ -1,150 +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.samza.storage.kv.descriptors;
-
-import java.util.HashMap;
-import java.util.Map;
-import junit.framework.Assert;
-import org.apache.samza.SamzaException;
-import org.apache.samza.config.Config;
-import org.apache.samza.config.JavaTableConfig;
-import org.apache.samza.config.JobConfig;
-import org.apache.samza.config.MapConfig;
-import org.apache.samza.config.StorageConfig;
-import org.apache.samza.context.Context;
-import org.apache.samza.context.TaskContext;
-import org.apache.samza.storage.kv.KeyValueStore;
-import org.apache.samza.table.descriptors.TableProvider;
-import org.apache.samza.table.TableSpec;
-import org.apache.samza.util.NoOpMetricsRegistry;
-import org.junit.Test;
-
-import static org.mockito.Matchers.any;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.when;
-
-
-public class TestBaseLocalStoreBackedTableProvider {
-
- @Test
- public void testInit() {
- Context context = mock(Context.class);
- TaskContext taskContext = mock(TaskContext.class);
- when(context.getTaskContext()).thenReturn(taskContext);
- when(taskContext.getStore(any())).thenReturn(mock(KeyValueStore.class));
- when(taskContext.getTaskMetricsRegistry()).thenReturn(new NoOpMetricsRegistry());
-
- TableSpec tableSpec = mock(TableSpec.class);
- when(tableSpec.getId()).thenReturn("t1");
-
- TableProvider tableProvider = createTableProvider(tableSpec);
- tableProvider.init(context);
- Assert.assertNotNull(tableProvider.getTable());
- }
-
- @Test(expected = SamzaException.class)
- public void testInitFail() {
- TableSpec tableSpec = mock(TableSpec.class);
- when(tableSpec.getId()).thenReturn("t1");
- TableProvider tableProvider = createTableProvider(tableSpec);
- Assert.assertNotNull(tableProvider.getTable());
- }
-
- @Test
- public void testGenerateCommonStoreConfig() {
- Map<String, String> generatedConfig = new HashMap<>();
- generatedConfig.put(String.format(JavaTableConfig.TABLE_KEY_SERDE, "t1"), "ks1");
- generatedConfig.put(String.format(JavaTableConfig.TABLE_VALUE_SERDE, "t1"), "vs1");
-
- TableSpec tableSpec = mock(TableSpec.class);
- when(tableSpec.getId()).thenReturn("t1");
-
- TableProvider tableProvider = createTableProvider(tableSpec);
- Map<String, String> tableConfig = tableProvider.generateConfig(new MapConfig(), generatedConfig);
- Assert.assertEquals("ks1", tableConfig.get(String.format(StorageConfig.KEY_SERDE(), "t1")));
- Assert.assertEquals("vs1", tableConfig.get(String.format(StorageConfig.MSG_SERDE(), "t1")));
- }
-
- @Test
- public void testChangelogDisabled() {
- TableSpec tableSpec = createTableDescriptor("t1")
- .getTableSpec();
-
- TableProvider tableProvider = createTableProvider(tableSpec);
- Map<String, String> tableConfig = tableProvider.generateConfig(new MapConfig(), new MapConfig());
- Assert.assertEquals(2, tableConfig.size());
- Assert.assertFalse(tableConfig.containsKey(String.format(StorageConfig.CHANGELOG_STREAM(), "t1")));
- }
-
- @Test
- public void testChangelogEnabled() {
- TableSpec tableSpec = createTableDescriptor("t1")
- .withChangelogEnabled()
- .getTableSpec();
-
- Map<String, String> jobConfig = new HashMap<>();
- jobConfig.put(JobConfig.JOB_NAME(), "test-job");
- jobConfig.put(JobConfig.JOB_ID(), "10");
-
- TableProvider tableProvider = createTableProvider(tableSpec);
- Map<String, String> tableConfig = tableProvider.generateConfig(new MapConfig(jobConfig), new MapConfig());
- Assert.assertEquals(3, tableConfig.size());
- Assert.assertEquals("test-job-10-table-t1", String.format(
- tableConfig.get(String.format(StorageConfig.CHANGELOG_STREAM(), "t1"))));
- }
-
- @Test
- public void testChangelogEnabledWithCustomParameters() {
- TableSpec tableSpec = createTableDescriptor("t1")
- .withChangelogStream("my-stream")
- .withChangelogReplicationFactor(100)
- .getTableSpec();
-
- TableProvider tableProvider = createTableProvider(tableSpec);
- Map<String, String> tableConfig = tableProvider.generateConfig(new MapConfig(), new MapConfig());
- Assert.assertEquals(4, tableConfig.size());
- Assert.assertEquals("my-stream", String.format(
- tableConfig.get(String.format(StorageConfig.CHANGELOG_STREAM(), "t1"))));
- Assert.assertEquals("100", String.format(
- tableConfig.get(String.format(StorageConfig.CHANGELOG_REPLICATION_FACTOR(), "t1"))));
- }
-
- private TableProvider createTableProvider(TableSpec tableSpec) {
- return new BaseLocalStoreBackedTableProvider(tableSpec) {
- @Override
- public Map<String, String> generateConfig(Config jobConfig, Map<String, String> generatedConfig) {
- return generateCommonStoreConfig(jobConfig, generatedConfig);
- }
- };
- }
-
- private BaseLocalStoreBackedTableDescriptor createTableDescriptor(String tableId) {
- return new BaseLocalStoreBackedTableDescriptor(tableId) {
- @Override
- public TableSpec getTableSpec() {
- validate();
- Map<String, String> tableSpecConfig = new HashMap<>();
- generateTableSpecConfig(tableSpecConfig);
- return new TableSpec(tableId, serde, null, tableSpecConfig,
- sideInputs, sideInputsProcessor);
- }
- };
- }
-}
http://git-wip-us.apache.org/repos/asf/samza/blob/2159171e/samza-kv/src/test/java/org/apache/samza/storage/kv/descriptors/TestBaseLocalTableProvider.java
----------------------------------------------------------------------
diff --git a/samza-kv/src/test/java/org/apache/samza/storage/kv/descriptors/TestBaseLocalTableProvider.java b/samza-kv/src/test/java/org/apache/samza/storage/kv/descriptors/TestBaseLocalTableProvider.java
new file mode 100644
index 0000000..1cb0197
--- /dev/null
+++ b/samza-kv/src/test/java/org/apache/samza/storage/kv/descriptors/TestBaseLocalTableProvider.java
@@ -0,0 +1,150 @@
+/*
+ * 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.samza.storage.kv.descriptors;
+
+import java.util.HashMap;
+import java.util.Map;
+import junit.framework.Assert;
+import org.apache.samza.SamzaException;
+import org.apache.samza.config.Config;
+import org.apache.samza.config.JavaTableConfig;
+import org.apache.samza.config.JobConfig;
+import org.apache.samza.config.MapConfig;
+import org.apache.samza.config.StorageConfig;
+import org.apache.samza.context.Context;
+import org.apache.samza.context.TaskContext;
+import org.apache.samza.storage.kv.KeyValueStore;
+import org.apache.samza.table.descriptors.TableProvider;
+import org.apache.samza.table.TableSpec;
+import org.apache.samza.util.NoOpMetricsRegistry;
+import org.junit.Test;
+
+import static org.mockito.Matchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+
+public class TestBaseLocalTableProvider {
+
+ @Test
+ public void testInit() {
+ Context context = mock(Context.class);
+ TaskContext taskContext = mock(TaskContext.class);
+ when(context.getTaskContext()).thenReturn(taskContext);
+ when(taskContext.getStore(any())).thenReturn(mock(KeyValueStore.class));
+ when(taskContext.getTaskMetricsRegistry()).thenReturn(new NoOpMetricsRegistry());
+
+ TableSpec tableSpec = mock(TableSpec.class);
+ when(tableSpec.getId()).thenReturn("t1");
+
+ TableProvider tableProvider = createTableProvider(tableSpec);
+ tableProvider.init(context);
+ Assert.assertNotNull(tableProvider.getTable());
+ }
+
+ @Test(expected = SamzaException.class)
+ public void testInitFail() {
+ TableSpec tableSpec = mock(TableSpec.class);
+ when(tableSpec.getId()).thenReturn("t1");
+ TableProvider tableProvider = createTableProvider(tableSpec);
+ Assert.assertNotNull(tableProvider.getTable());
+ }
+
+ @Test
+ public void testGenerateCommonStoreConfig() {
+ Map<String, String> generatedConfig = new HashMap<>();
+ generatedConfig.put(String.format(JavaTableConfig.TABLE_KEY_SERDE, "t1"), "ks1");
+ generatedConfig.put(String.format(JavaTableConfig.TABLE_VALUE_SERDE, "t1"), "vs1");
+
+ TableSpec tableSpec = mock(TableSpec.class);
+ when(tableSpec.getId()).thenReturn("t1");
+
+ TableProvider tableProvider = createTableProvider(tableSpec);
+ Map<String, String> tableConfig = tableProvider.generateConfig(new MapConfig(), generatedConfig);
+ Assert.assertEquals("ks1", tableConfig.get(String.format(StorageConfig.KEY_SERDE(), "t1")));
+ Assert.assertEquals("vs1", tableConfig.get(String.format(StorageConfig.MSG_SERDE(), "t1")));
+ }
+
+ @Test
+ public void testChangelogDisabled() {
+ TableSpec tableSpec = createTableDescriptor("t1")
+ .getTableSpec();
+
+ TableProvider tableProvider = createTableProvider(tableSpec);
+ Map<String, String> tableConfig = tableProvider.generateConfig(new MapConfig(), new MapConfig());
+ Assert.assertEquals(2, tableConfig.size());
+ Assert.assertFalse(tableConfig.containsKey(String.format(StorageConfig.CHANGELOG_STREAM(), "t1")));
+ }
+
+ @Test
+ public void testChangelogEnabled() {
+ TableSpec tableSpec = createTableDescriptor("t1")
+ .withChangelogEnabled()
+ .getTableSpec();
+
+ Map<String, String> jobConfig = new HashMap<>();
+ jobConfig.put(JobConfig.JOB_NAME(), "test-job");
+ jobConfig.put(JobConfig.JOB_ID(), "10");
+
+ TableProvider tableProvider = createTableProvider(tableSpec);
+ Map<String, String> tableConfig = tableProvider.generateConfig(new MapConfig(jobConfig), new MapConfig());
+ Assert.assertEquals(3, tableConfig.size());
+ Assert.assertEquals("test-job-10-table-t1", String.format(
+ tableConfig.get(String.format(StorageConfig.CHANGELOG_STREAM(), "t1"))));
+ }
+
+ @Test
+ public void testChangelogEnabledWithCustomParameters() {
+ TableSpec tableSpec = createTableDescriptor("t1")
+ .withChangelogStream("my-stream")
+ .withChangelogReplicationFactor(100)
+ .getTableSpec();
+
+ TableProvider tableProvider = createTableProvider(tableSpec);
+ Map<String, String> tableConfig = tableProvider.generateConfig(new MapConfig(), new MapConfig());
+ Assert.assertEquals(4, tableConfig.size());
+ Assert.assertEquals("my-stream", String.format(
+ tableConfig.get(String.format(StorageConfig.CHANGELOG_STREAM(), "t1"))));
+ Assert.assertEquals("100", String.format(
+ tableConfig.get(String.format(StorageConfig.CHANGELOG_REPLICATION_FACTOR(), "t1"))));
+ }
+
+ private TableProvider createTableProvider(TableSpec tableSpec) {
+ return new BaseLocalTableProvider(tableSpec) {
+ @Override
+ public Map<String, String> generateConfig(Config jobConfig, Map<String, String> generatedConfig) {
+ return generateCommonStoreConfig(jobConfig, generatedConfig);
+ }
+ };
+ }
+
+ private BaseLocalTableDescriptor createTableDescriptor(String tableId) {
+ return new BaseLocalTableDescriptor(tableId) {
+ @Override
+ public TableSpec getTableSpec() {
+ validate();
+ Map<String, String> tableSpecConfig = new HashMap<>();
+ generateTableSpecConfig(tableSpecConfig);
+ return new TableSpec(tableId, serde, null, tableSpecConfig,
+ sideInputs, sideInputsProcessor);
+ }
+ };
+ }
+}
http://git-wip-us.apache.org/repos/asf/samza/blob/2159171e/samza-test/src/test/java/org/apache/samza/test/table/TestLocalTable.java
----------------------------------------------------------------------
diff --git a/samza-test/src/test/java/org/apache/samza/test/table/TestLocalTable.java b/samza-test/src/test/java/org/apache/samza/test/table/TestLocalTable.java
index e94c62f..fa279f2 100644
--- a/samza-test/src/test/java/org/apache/samza/test/table/TestLocalTable.java
+++ b/samza-test/src/test/java/org/apache/samza/test/table/TestLocalTable.java
@@ -51,7 +51,7 @@ import org.apache.samza.serializers.NoOpSerde;
import org.apache.samza.standalone.PassthroughJobCoordinatorFactory;
import org.apache.samza.storage.kv.Entry;
import org.apache.samza.storage.kv.KeyValueStore;
-import org.apache.samza.storage.kv.LocalStoreBackedReadWriteTable;
+import org.apache.samza.storage.kv.LocalReadWriteTable;
import org.apache.samza.storage.kv.inmemory.descriptors.InMemoryTableDescriptor;
import org.apache.samza.system.IncomingMessageEnvelope;
import org.apache.samza.table.ReadWriteTable;
@@ -359,7 +359,7 @@ public class TestLocalTable extends AbstractIntegrationTestHarness {
@Test
public void testAsyncOperation() throws Exception {
KeyValueStore kvStore = mock(KeyValueStore.class);
- LocalStoreBackedReadWriteTable<String, String> table = new LocalStoreBackedReadWriteTable<>("table1", kvStore);
+ LocalReadWriteTable<String, String> table = new LocalReadWriteTable<>("table1", kvStore);
Context context = mock(Context.class);
TaskContext taskContext = mock(TaskContext.class);
when(context.getTaskContext()).thenReturn(taskContext);