[jira] [Commented] (STREAMS-478) streams persist reader/writer for Apache Cassandra

2016-12-30 Thread ASF GitHub Bot (JIRA) 

[ 
https://issues.apache.org/jira/browse/STREAMS-478?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15788621#comment-15788621
 ] 

ASF GitHub Bot commented on STREAMS-478:


Github user asfgit closed the pull request at:

https://github.com/apache/incubator-streams/pull/345


> streams persist reader/writer for Apache Cassandra
> --
>
> Key: STREAMS-478
> URL: https://issues.apache.org/jira/browse/STREAMS-478
> Project: Streams
>  Issue Type: New Feature
>Reporter: Subhobrata Dey
>
> This issue is created to propose the streams persist reader/writer to 
> get/post documents from/to apache cassandra.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (STREAMS-478) streams persist reader/writer for Apache Cassandra

2016-12-29 Thread ASF GitHub Bot (JIRA) 

[ 
https://issues.apache.org/jira/browse/STREAMS-478?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15786402#comment-15786402
 ] 

ASF GitHub Bot commented on STREAMS-478:


Github user steveblackmon commented on the issue:

https://github.com/apache/incubator-streams/pull/345
  
 



> streams persist reader/writer for Apache Cassandra
> --
>
> Key: STREAMS-478
> URL: https://issues.apache.org/jira/browse/STREAMS-478
> Project: Streams
>  Issue Type: New Feature
>Reporter: Subhobrata Dey
>
> This issue is created to propose the streams persist reader/writer to 
> get/post documents from/to apache cassandra.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (STREAMS-478) streams persist reader/writer for Apache Cassandra

2016-12-29 Thread ASF GitHub Bot (JIRA) 

[ 
https://issues.apache.org/jira/browse/STREAMS-478?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15785905#comment-15785905
 ] 

ASF GitHub Bot commented on STREAMS-478:


Github user smarthi commented on the issue:

https://github.com/apache/incubator-streams/pull/345
  
+1 to merge


> streams persist reader/writer for Apache Cassandra
> --
>
> Key: STREAMS-478
> URL: https://issues.apache.org/jira/browse/STREAMS-478
> Project: Streams
>  Issue Type: New Feature
>Reporter: Subhobrata Dey
>
> This issue is created to propose the streams persist reader/writer to 
> get/post documents from/to apache cassandra.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (STREAMS-478) streams persist reader/writer for Apache Cassandra

2016-12-28 Thread ASF GitHub Bot (JIRA) 

[ 
https://issues.apache.org/jira/browse/STREAMS-478?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15784155#comment-15784155
 ] 

ASF GitHub Bot commented on STREAMS-478:


Github user sbcd90 commented on the issue:

https://github.com/apache/incubator-streams/pull/345
  
Hi @smarthi ,

I have refactored the code to remove the `printStackTrace` method 
occurrence by adding `LOGGER.warn` statements.


> streams persist reader/writer for Apache Cassandra
> --
>
> Key: STREAMS-478
> URL: https://issues.apache.org/jira/browse/STREAMS-478
> Project: Streams
>  Issue Type: New Feature
>Reporter: Subhobrata Dey
>
> This issue is created to propose the streams persist reader/writer to 
> get/post documents from/to apache cassandra.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (STREAMS-478) streams persist reader/writer for Apache Cassandra

2016-12-28 Thread ASF GitHub Bot (JIRA) 

[ 
https://issues.apache.org/jira/browse/STREAMS-478?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15784149#comment-15784149
 ] 

ASF GitHub Bot commented on STREAMS-478:


Github user sbcd90 commented on a diff in the pull request:

https://github.com/apache/incubator-streams/pull/345#discussion_r94092806
  
--- Diff: 
streams-contrib/streams-persist-cassandra/src/main/java/org/apache/streams/cassandra/CassandraPersistReader.java
 ---
@@ -0,0 +1,316 @@
+/*
+ * 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
+ *
+ *   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.streams.cassandra;
+
+import org.apache.streams.config.ComponentConfigurator;
+import org.apache.streams.config.StreamsConfigurator;
+import org.apache.streams.core.DatumStatusCounter;
+import org.apache.streams.core.StreamsDatum;
+import org.apache.streams.core.StreamsPersistReader;
+import org.apache.streams.core.StreamsResultSet;
+import org.apache.streams.jackson.StreamsJacksonMapper;
+
+import com.datastax.driver.core.Cluster;
+import com.datastax.driver.core.DataType;
+import com.datastax.driver.core.KeyspaceMetadata;
+import com.datastax.driver.core.Metadata;
+import com.datastax.driver.core.ResultSet;
+import com.datastax.driver.core.Row;
+import com.datastax.driver.core.Session;
+import com.datastax.driver.core.TableMetadata;
+import com.datastax.driver.core.querybuilder.QueryBuilder;
+import com.datastax.driver.core.schemabuilder.SchemaBuilder;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import com.google.common.collect.Queues;
+import org.apache.commons.lang3.StringUtils;
+import org.joda.time.DateTime;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.math.BigInteger;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Queue;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+/**
+ * CassandraPersistReader reads documents from cassandra.
+ */
+public class CassandraPersistReader implements StreamsPersistReader {
+
+  public static final String STREAMS_ID = "CassandraPersistReader";
+
+  public static final Logger LOGGER = 
LoggerFactory.getLogger(CassandraPersistReader.class);
+
+  protected volatile Queue persistQueue;
+
+  private ObjectMapper mapper = StreamsJacksonMapper.getInstance();
+
+  private ExecutorService executor;
+  private CompletableFuture readerTaskFuture = new 
CompletableFuture<>();
+
+  private CassandraConfiguration config;
+
+  protected Cluster cluster;
+  protected Session session;
+
+  protected String keyspace;
+  protected String table;
+  protected Iterator rowIterator;
+
+  protected final ReadWriteLock lock = new ReentrantReadWriteLock();
+
+  /**
+   * CassandraPersistReader constructor - resolves CassandraConfiguration 
from JVM 'cassandra'.
+   */
+  public CassandraPersistReader() {
+this.config = new ComponentConfigurator<>(CassandraConfiguration.class)
+  
.detectConfiguration(StreamsConfigurator.getConfig().getConfig("cassandra"));
+  }
+
+  /**
+   * CassandraPersistReader constructor - uses supplied 
CassandraConfiguration.
+   * @param config config
+   */
+  public CassandraPersistReader(CassandraConfiguration config) {
+this.config = config;
+  }
+
+  /**
+   * CassandraPersistReader constructor - uses supplied persistQueue.
+   * @param persistQueue persistQueue
+   */
+  public CassandraPersistReader(Queue persistQueue) {
+

[jira] [Commented] (STREAMS-478) streams persist reader/writer for Apache Cassandra

2016-12-28 Thread ASF GitHub Bot (JIRA) 

[ 
https://issues.apache.org/jira/browse/STREAMS-478?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15784005#comment-15784005
 ] 

ASF GitHub Bot commented on STREAMS-478:


Github user smarthi commented on a diff in the pull request:

https://github.com/apache/incubator-streams/pull/345#discussion_r94089087
  
--- Diff: 
streams-contrib/streams-persist-cassandra/src/main/java/org/apache/streams/cassandra/CassandraPersistWriter.java
 ---
@@ -0,0 +1,323 @@
+/*
+ * 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
+ *
+ *   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.streams.cassandra;
+
+import org.apache.streams.config.ComponentConfigurator;
+import org.apache.streams.config.StreamsConfigurator;
+import org.apache.streams.core.StreamsDatum;
+import org.apache.streams.core.StreamsPersistWriter;
+import org.apache.streams.jackson.StreamsJacksonMapper;
+import org.apache.streams.util.GuidUtils;
+
+import com.datastax.driver.core.BatchStatement;
+import com.datastax.driver.core.BoundStatement;
+import com.datastax.driver.core.Cluster;
+import com.datastax.driver.core.DataType;
+import com.datastax.driver.core.KeyspaceMetadata;
+import com.datastax.driver.core.Metadata;
+import com.datastax.driver.core.PreparedStatement;
+import com.datastax.driver.core.Session;
+import com.datastax.driver.core.TableMetadata;
+import com.datastax.driver.core.querybuilder.Insert;
+import com.datastax.driver.core.querybuilder.QueryBuilder;
+import com.datastax.driver.core.schemabuilder.SchemaBuilder;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import org.apache.commons.lang3.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.Closeable;
+import java.io.Flushable;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Queue;
+import java.util.Random;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+public class CassandraPersistWriter implements StreamsPersistWriter, 
Runnable, Flushable, Closeable {
+
+  public static final String STREAMS_ID = "CassandraPersistWriter";
+
+  private static final Logger LOGGER = 
LoggerFactory.getLogger(CassandraPersistWriter.class);
+
+  private static final long MAX_WRITE_LATENCY = 1000;
+
+  protected volatile Queue persistQueue;
+
+  private ObjectMapper mapper = StreamsJacksonMapper.getInstance();
+  private volatile AtomicLong lastWrite = new 
AtomicLong(System.currentTimeMillis());
+  private ScheduledExecutorService backgroundFlushTask = 
Executors.newSingleThreadScheduledExecutor();
+
+  private CassandraConfiguration config;
+
+  protected Cluster cluster;
+  protected Session session;
+
+  protected String keyspace;
+  protected String table;
+  protected PreparedStatement insertStatement;
+
+  protected List insertBatch = new ArrayList<>();
+
+  protected final ReadWriteLock lock = new ReentrantReadWriteLock();
+
+  public CassandraPersistWriter() {
+this(new ComponentConfigurator<>(CassandraConfiguration.class)
+
.detectConfiguration(StreamsConfigurator.getConfig().getConfig("cassandra")));
+  }
+
+  public CassandraPersistWriter(CassandraConfiguration config) {
+this.config = config;
+  }
+
+  public void setPersistQueue(Queue persistQueue) {
+this.persistQueue = persistQueue;
+  }
+
+  public Queue getPersistQueue() {
+return persistQueue;
+  }
+
+

[jira] [Commented] (STREAMS-478) streams persist reader/writer for Apache Cassandra

2016-12-28 Thread ASF GitHub Bot (JIRA) 

[ 
https://issues.apache.org/jira/browse/STREAMS-478?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15783907#comment-15783907
 ] 

ASF GitHub Bot commented on STREAMS-478:


Github user smarthi commented on a diff in the pull request:

https://github.com/apache/incubator-streams/pull/345#discussion_r94086200
  
--- Diff: 
streams-contrib/streams-persist-cassandra/src/main/java/org/apache/streams/cassandra/CassandraPersistReader.java
 ---
@@ -0,0 +1,316 @@
+/*
+ * 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
+ *
+ *   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.streams.cassandra;
+
+import org.apache.streams.config.ComponentConfigurator;
+import org.apache.streams.config.StreamsConfigurator;
+import org.apache.streams.core.DatumStatusCounter;
+import org.apache.streams.core.StreamsDatum;
+import org.apache.streams.core.StreamsPersistReader;
+import org.apache.streams.core.StreamsResultSet;
+import org.apache.streams.jackson.StreamsJacksonMapper;
+
+import com.datastax.driver.core.Cluster;
+import com.datastax.driver.core.DataType;
+import com.datastax.driver.core.KeyspaceMetadata;
+import com.datastax.driver.core.Metadata;
+import com.datastax.driver.core.ResultSet;
+import com.datastax.driver.core.Row;
+import com.datastax.driver.core.Session;
+import com.datastax.driver.core.TableMetadata;
+import com.datastax.driver.core.querybuilder.QueryBuilder;
+import com.datastax.driver.core.schemabuilder.SchemaBuilder;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import com.google.common.collect.Queues;
+import org.apache.commons.lang3.StringUtils;
+import org.joda.time.DateTime;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.math.BigInteger;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Queue;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+/**
+ * CassandraPersistReader reads documents from cassandra.
+ */
+public class CassandraPersistReader implements StreamsPersistReader {
+
+  public static final String STREAMS_ID = "CassandraPersistReader";
+
+  public static final Logger LOGGER = 
LoggerFactory.getLogger(CassandraPersistReader.class);
+
+  protected volatile Queue persistQueue;
+
+  private ObjectMapper mapper = StreamsJacksonMapper.getInstance();
+
+  private ExecutorService executor;
+  private CompletableFuture readerTaskFuture = new 
CompletableFuture<>();
+
+  private CassandraConfiguration config;
+
+  protected Cluster cluster;
+  protected Session session;
+
+  protected String keyspace;
+  protected String table;
+  protected Iterator rowIterator;
+
+  protected final ReadWriteLock lock = new ReentrantReadWriteLock();
+
+  /**
+   * CassandraPersistReader constructor - resolves CassandraConfiguration 
from JVM 'cassandra'.
+   */
+  public CassandraPersistReader() {
+this.config = new ComponentConfigurator<>(CassandraConfiguration.class)
+  
.detectConfiguration(StreamsConfigurator.getConfig().getConfig("cassandra"));
+  }
+
+  /**
+   * CassandraPersistReader constructor - uses supplied 
CassandraConfiguration.
+   * @param config config
+   */
+  public CassandraPersistReader(CassandraConfiguration config) {
+this.config = config;
+  }
+
+  /**
+   * CassandraPersistReader constructor - uses supplied persistQueue.
+   * @param persistQueue persistQueue
+   */
+  public CassandraPersistReader(Queue persistQueue) {
+

[jira] [Commented] (STREAMS-478) streams persist reader/writer for Apache Cassandra

2016-12-28 Thread Subhobrata Dey (JIRA) 

[ 
https://issues.apache.org/jira/browse/STREAMS-478?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15783616#comment-15783616
 ] 

Subhobrata Dey commented on STREAMS-478:


Hi [~jfrazee],

I have signed & email-ed the scanned copy of the agreement to 
secret...@apache.org.

> streams persist reader/writer for Apache Cassandra
> --
>
> Key: STREAMS-478
> URL: https://issues.apache.org/jira/browse/STREAMS-478
> Project: Streams
>  Issue Type: New Feature
>Reporter: Subhobrata Dey
>
> This issue is created to propose the streams persist reader/writer to 
> get/post documents from/to apache cassandra.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (STREAMS-478) streams persist reader/writer for Apache Cassandra

2016-12-28 Thread ASF GitHub Bot (JIRA) 

[ 
https://issues.apache.org/jira/browse/STREAMS-478?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15783565#comment-15783565
 ] 

ASF GitHub Bot commented on STREAMS-478:


Github user sbcd90 commented on the issue:

https://github.com/apache/incubator-streams/pull/345
  
Hi @smarthi , @steveblackmon ,

Thanks for the review.
I have refactored the code to handle the following issues:

- removing of guava APIs.
- no types specified in RHS.
- Switch to TestNG in IT tests
- used `streamsDatum.getMetadata().get("id")` if it exists, enabling an 
stream to get a write-then-overwrite behavior when the id is already known
- supported an array of strings for host


> streams persist reader/writer for Apache Cassandra
> --
>
> Key: STREAMS-478
> URL: https://issues.apache.org/jira/browse/STREAMS-478
> Project: Streams
>  Issue Type: New Feature
>Reporter: Subhobrata Dey
>
> This issue is created to propose the streams persist reader/writer to 
> get/post documents from/to apache cassandra.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (STREAMS-478) streams persist reader/writer for Apache Cassandra

2016-12-28 Thread ASF GitHub Bot (JIRA) 

[ 
https://issues.apache.org/jira/browse/STREAMS-478?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15783169#comment-15783169
 ] 

ASF GitHub Bot commented on STREAMS-478:


Github user steveblackmon commented on a diff in the pull request:

https://github.com/apache/incubator-streams/pull/345#discussion_r94047751
  
--- Diff: 
streams-contrib/streams-persist-cassandra/src/main/jsonschema/org/apache/streams/cassandra/CassandraConfiguration.json
 ---
@@ -0,0 +1,44 @@
+{
+  "$schema": "http://json-schema.org/draft-03/schema;,
+  "$license": [
+"http://www.apache.org/licenses/LICENSE-2.0;
+  ],
+  "id": "#",
+  "type": "object",
+  "javaType": "org.apache.streams.cassandra.CassandraConfiguration",
+  "javaInterfaces": ["java.io.Serializable"],
+  "properties": {
+"host": {
+  "type": "string",
--- End diff --

perhaps this should be an array of strings, so module can start-up with 
more than one seed host.


> streams persist reader/writer for Apache Cassandra
> --
>
> Key: STREAMS-478
> URL: https://issues.apache.org/jira/browse/STREAMS-478
> Project: Streams
>  Issue Type: New Feature
>Reporter: Subhobrata Dey
>
> This issue is created to propose the streams persist reader/writer to 
> get/post documents from/to apache cassandra.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (STREAMS-478) streams persist reader/writer for Apache Cassandra

2016-12-28 Thread ASF GitHub Bot (JIRA) 

[ 
https://issues.apache.org/jira/browse/STREAMS-478?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15783162#comment-15783162
 ] 

ASF GitHub Bot commented on STREAMS-478:


Github user steveblackmon commented on a diff in the pull request:

https://github.com/apache/incubator-streams/pull/345#discussion_r94047563
  
--- Diff: 
streams-contrib/streams-persist-cassandra/src/main/java/org/apache/streams/cassandra/CassandraPersistWriter.java
 ---
@@ -0,0 +1,312 @@
+/*
+ * 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
+ *
+ *   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.streams.cassandra;
+
+import org.apache.streams.config.ComponentConfigurator;
+import org.apache.streams.config.StreamsConfigurator;
+import org.apache.streams.core.StreamsDatum;
+import org.apache.streams.core.StreamsPersistWriter;
+import org.apache.streams.jackson.StreamsJacksonMapper;
+import org.apache.streams.util.GuidUtils;
+
+import com.datastax.driver.core.BatchStatement;
+import com.datastax.driver.core.BoundStatement;
+import com.datastax.driver.core.Cluster;
+import com.datastax.driver.core.DataType;
+import com.datastax.driver.core.KeyspaceMetadata;
+import com.datastax.driver.core.Metadata;
+import com.datastax.driver.core.PreparedStatement;
+import com.datastax.driver.core.Session;
+import com.datastax.driver.core.TableMetadata;
+import com.datastax.driver.core.querybuilder.Insert;
+import com.datastax.driver.core.querybuilder.QueryBuilder;
+import com.datastax.driver.core.schemabuilder.SchemaBuilder;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import com.google.common.base.Strings;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.Closeable;
+import java.io.Flushable;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Queue;
+import java.util.Random;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+public class CassandraPersistWriter implements StreamsPersistWriter, 
Runnable, Flushable, Closeable {
+
+  public static final String STREAMS_ID = "CassandraPersistWriter";
+
+  private static final Logger LOGGER = 
LoggerFactory.getLogger(CassandraPersistWriter.class);
+
+  private static final long MAX_WRITE_LATENCY = 1000;
+
+  protected volatile Queue persistQueue;
+
+  private ObjectMapper mapper = StreamsJacksonMapper.getInstance();
+  private volatile AtomicLong lastWrite = new 
AtomicLong(System.currentTimeMillis());
+  private ScheduledExecutorService backgroundFlushTask = 
Executors.newSingleThreadScheduledExecutor();
+
+  private CassandraConfiguration config;
+
+  protected Cluster cluster;
+  protected Session session;
+
+  protected String keyspace;
+  protected String table;
+  protected PreparedStatement insertStatement;
+
+  protected List insertBatch = new ArrayList<>();
+
+  protected final ReadWriteLock lock = new ReentrantReadWriteLock();
+
+  public CassandraPersistWriter() {
+this(new ComponentConfigurator<>(CassandraConfiguration.class)
+
.detectConfiguration(StreamsConfigurator.getConfig().getConfig("cassandra")));
+  }
+
+  public CassandraPersistWriter(CassandraConfiguration config) {
+this.config = config;
+  }
+
+  public void setPersistQueue(Queue persistQueue) {
+this.persistQueue = persistQueue;
+  }
+
+  public Queue getPersistQueue() {
+return persistQueue;
+  }
+
+

[jira] [Commented] (STREAMS-478) streams persist reader/writer for Apache Cassandra

2016-12-28 Thread ASF GitHub Bot (JIRA) 

[ 
https://issues.apache.org/jira/browse/STREAMS-478?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15783118#comment-15783118
 ] 

ASF GitHub Bot commented on STREAMS-478:


Github user smarthi commented on a diff in the pull request:

https://github.com/apache/incubator-streams/pull/345#discussion_r94044782
  
--- Diff: 
streams-contrib/streams-persist-cassandra/src/test/java/org/apache/streams/cassandra/test/CassandraPersistIT.java
 ---
@@ -0,0 +1,106 @@
+/*
+ * 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
+ *
+ *   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.streams.cassandra.test;
+
+import org.apache.streams.cassandra.CassandraConfiguration;
+import org.apache.streams.cassandra.CassandraPersistReader;
+import org.apache.streams.cassandra.CassandraPersistWriter;
+import org.apache.streams.config.ComponentConfigurator;
+import org.apache.streams.core.StreamsDatum;
+import org.apache.streams.core.StreamsResultSet;
+import org.apache.streams.jackson.StreamsJacksonMapper;
+import org.apache.streams.pojo.json.Activity;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.typesafe.config.Config;
+import com.typesafe.config.ConfigFactory;
+import com.typesafe.config.ConfigParseOptions;
+import org.apache.commons.io.IOUtils;
+import org.junit.Assert;
--- End diff --

Switch to TestNG in IT tests


> streams persist reader/writer for Apache Cassandra
> --
>
> Key: STREAMS-478
> URL: https://issues.apache.org/jira/browse/STREAMS-478
> Project: Streams
>  Issue Type: New Feature
>Reporter: Subhobrata Dey
>
> This issue is created to propose the streams persist reader/writer to 
> get/post documents from/to apache cassandra.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (STREAMS-478) streams persist reader/writer for Apache Cassandra

2016-12-28 Thread ASF GitHub Bot (JIRA) 

[ 
https://issues.apache.org/jira/browse/STREAMS-478?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15783116#comment-15783116
 ] 

ASF GitHub Bot commented on STREAMS-478:


Github user smarthi commented on a diff in the pull request:

https://github.com/apache/incubator-streams/pull/345#discussion_r94044606
  
--- Diff: 
streams-contrib/streams-persist-cassandra/src/main/java/org/apache/streams/cassandra/CassandraPersistWriter.java
 ---
@@ -0,0 +1,312 @@
+/*
+ * 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
+ *
+ *   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.streams.cassandra;
+
+import org.apache.streams.config.ComponentConfigurator;
+import org.apache.streams.config.StreamsConfigurator;
+import org.apache.streams.core.StreamsDatum;
+import org.apache.streams.core.StreamsPersistWriter;
+import org.apache.streams.jackson.StreamsJacksonMapper;
+import org.apache.streams.util.GuidUtils;
+
+import com.datastax.driver.core.BatchStatement;
+import com.datastax.driver.core.BoundStatement;
+import com.datastax.driver.core.Cluster;
+import com.datastax.driver.core.DataType;
+import com.datastax.driver.core.KeyspaceMetadata;
+import com.datastax.driver.core.Metadata;
+import com.datastax.driver.core.PreparedStatement;
+import com.datastax.driver.core.Session;
+import com.datastax.driver.core.TableMetadata;
+import com.datastax.driver.core.querybuilder.Insert;
+import com.datastax.driver.core.querybuilder.QueryBuilder;
+import com.datastax.driver.core.schemabuilder.SchemaBuilder;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import com.google.common.base.Strings;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.Closeable;
+import java.io.Flushable;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Queue;
+import java.util.Random;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+public class CassandraPersistWriter implements StreamsPersistWriter, 
Runnable, Flushable, Closeable {
+
+  public static final String STREAMS_ID = "CassandraPersistWriter";
+
+  private static final Logger LOGGER = 
LoggerFactory.getLogger(CassandraPersistWriter.class);
+
+  private static final long MAX_WRITE_LATENCY = 1000;
+
+  protected volatile Queue persistQueue;
+
+  private ObjectMapper mapper = StreamsJacksonMapper.getInstance();
+  private volatile AtomicLong lastWrite = new 
AtomicLong(System.currentTimeMillis());
+  private ScheduledExecutorService backgroundFlushTask = 
Executors.newSingleThreadScheduledExecutor();
+
+  private CassandraConfiguration config;
+
+  protected Cluster cluster;
+  protected Session session;
+
+  protected String keyspace;
+  protected String table;
+  protected PreparedStatement insertStatement;
+
+  protected List insertBatch = new ArrayList<>();
+
+  protected final ReadWriteLock lock = new ReentrantReadWriteLock();
+
+  public CassandraPersistWriter() {
+this(new ComponentConfigurator<>(CassandraConfiguration.class)
+
.detectConfiguration(StreamsConfigurator.getConfig().getConfig("cassandra")));
+  }
+
+  public CassandraPersistWriter(CassandraConfiguration config) {
+this.config = config;
+  }
+
+  public void setPersistQueue(Queue persistQueue) {
+this.persistQueue = persistQueue;
+  }
+
+  public Queue getPersistQueue() {
+return persistQueue;
+  }
+
+  @Override

[jira] [Commented] (STREAMS-478) streams persist reader/writer for Apache Cassandra

2016-12-28 Thread ASF GitHub Bot (JIRA) 

[ 
https://issues.apache.org/jira/browse/STREAMS-478?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15783113#comment-15783113
 ] 

ASF GitHub Bot commented on STREAMS-478:


Github user smarthi commented on a diff in the pull request:

https://github.com/apache/incubator-streams/pull/345#discussion_r94044472
  
--- Diff: 
streams-contrib/streams-persist-cassandra/src/main/java/org/apache/streams/cassandra/CassandraPersistReader.java
 ---
@@ -0,0 +1,315 @@
+/*
+ * 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
+ *
+ *   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.streams.cassandra;
+
+import org.apache.streams.config.ComponentConfigurator;
+import org.apache.streams.config.StreamsConfigurator;
+import org.apache.streams.core.DatumStatusCounter;
+import org.apache.streams.core.StreamsDatum;
+import org.apache.streams.core.StreamsPersistReader;
+import org.apache.streams.core.StreamsResultSet;
+import org.apache.streams.jackson.StreamsJacksonMapper;
+
+import com.datastax.driver.core.Cluster;
+import com.datastax.driver.core.DataType;
+import com.datastax.driver.core.KeyspaceMetadata;
+import com.datastax.driver.core.Metadata;
+import com.datastax.driver.core.ResultSet;
+import com.datastax.driver.core.Row;
+import com.datastax.driver.core.Session;
+import com.datastax.driver.core.TableMetadata;
+import com.datastax.driver.core.querybuilder.QueryBuilder;
+import com.datastax.driver.core.schemabuilder.SchemaBuilder;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import com.google.common.base.Strings;
+import com.google.common.collect.Queues;
+import org.joda.time.DateTime;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.math.BigInteger;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Queue;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+/**
+ * CassandraPersistReader reads documents from cassandra.
+ */
+public class CassandraPersistReader implements StreamsPersistReader {
+
+  public static final String STREAMS_ID = "CassandraPersistReader";
+
+  public static final Logger LOGGER = 
LoggerFactory.getLogger(CassandraPersistReader.class);
+
+  protected volatile Queue persistQueue;
+
+  private ObjectMapper mapper = StreamsJacksonMapper.getInstance();
+
+  private ExecutorService executor;
+  private CompletableFuture readerTaskFuture = new 
CompletableFuture<>();
+
+  private CassandraConfiguration config;
+
+  protected Cluster cluster;
+  protected Session session;
+
+  protected String keyspace;
+  protected String table;
+  protected Iterator rowIterator;
+
+  protected final ReadWriteLock lock = new ReentrantReadWriteLock();
+
+  /**
+   * CassandraPersistReader constructor - resolves CassandraConfiguration 
from JVM 'cassandra'.
+   */
+  public CassandraPersistReader() {
+this.config = new ComponentConfigurator<>(CassandraConfiguration.class)
+  
.detectConfiguration(StreamsConfigurator.getConfig().getConfig("cassandra"));
+  }
+
+  /**
+   * CassandraPersistReader constructor - uses supplied 
CassandraConfiguration.
+   * @param config config
+   */
+  public CassandraPersistReader(CassandraConfiguration config) {
+this.config = config;
+  }
+
+  /**
+   * CassandraPersistReader constructor - uses supplied persistQueue.
+   * @param persistQueue persistQueue
+   */
+  public CassandraPersistReader(Queue persistQueue) {
+this.config

[jira] [Commented] (STREAMS-478) streams persist reader/writer for Apache Cassandra

2016-12-27 Thread ASF GitHub Bot (JIRA) 

[ 
https://issues.apache.org/jira/browse/STREAMS-478?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15782313#comment-15782313
 ] 

ASF GitHub Bot commented on STREAMS-478:


Github user smarthi commented on a diff in the pull request:

https://github.com/apache/incubator-streams/pull/345#discussion_r94002347
  
--- Diff: 
streams-contrib/streams-persist-cassandra/src/main/java/org/apache/streams/cassandra/CassandraPersistReader.java
 ---
@@ -0,0 +1,315 @@
+/*
+ * 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
+ *
+ *   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.streams.cassandra;
+
+import org.apache.streams.config.ComponentConfigurator;
+import org.apache.streams.config.StreamsConfigurator;
+import org.apache.streams.core.DatumStatusCounter;
+import org.apache.streams.core.StreamsDatum;
+import org.apache.streams.core.StreamsPersistReader;
+import org.apache.streams.core.StreamsResultSet;
+import org.apache.streams.jackson.StreamsJacksonMapper;
+
+import com.datastax.driver.core.Cluster;
+import com.datastax.driver.core.DataType;
+import com.datastax.driver.core.KeyspaceMetadata;
+import com.datastax.driver.core.Metadata;
+import com.datastax.driver.core.ResultSet;
+import com.datastax.driver.core.Row;
+import com.datastax.driver.core.Session;
+import com.datastax.driver.core.TableMetadata;
+import com.datastax.driver.core.querybuilder.QueryBuilder;
+import com.datastax.driver.core.schemabuilder.SchemaBuilder;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import com.google.common.base.Strings;
--- End diff --

We have moved to Java 8 since the 0.4.1 release, there's no need to use 
Guava APIs.  Replace with equivalent Java 8 or Apace Commns api 


> streams persist reader/writer for Apache Cassandra
> --
>
> Key: STREAMS-478
> URL: https://issues.apache.org/jira/browse/STREAMS-478
> Project: Streams
>  Issue Type: New Feature
>Reporter: Subhobrata Dey
>
> This issue is created to propose the streams persist reader/writer to 
> get/post documents from/to apache cassandra.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (STREAMS-478) streams persist reader/writer for Apache Cassandra

2016-12-27 Thread Joey Frazee (JIRA) 

[ 
https://issues.apache.org/jira/browse/STREAMS-478?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15781845#comment-15781845
 ] 

Joey Frazee commented on STREAMS-478:
-

[~sbcd90] Thanks for the contrib! I'm going to start reviewing this but thanks 
in advance for updating docs and including unit tests and all that.

In the meantime, it doesn't look like you have an Apache ICLA 
[https://www.apache.org/licenses/icla.txt] on file, so I was wondering if you'd 
be able to fill that out and email it in?

> streams persist reader/writer for Apache Cassandra
> --
>
> Key: STREAMS-478
> URL: https://issues.apache.org/jira/browse/STREAMS-478
> Project: Streams
>  Issue Type: New Feature
>Reporter: Subhobrata Dey
>
> This issue is created to propose the streams persist reader/writer to 
> get/post documents from/to apache cassandra.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (STREAMS-478) streams persist reader/writer for Apache Cassandra

2016-12-24 Thread ASF GitHub Bot (JIRA) 

[ 
https://issues.apache.org/jira/browse/STREAMS-478?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15775763#comment-15775763
 ] 

ASF GitHub Bot commented on STREAMS-478:


GitHub user sbcd90 opened a pull request:

https://github.com/apache/incubator-streams/pull/345

[STREAMS-478] streams persist reader/writer for Apache Cassandra

This issue is created to propose the streams persist reader/writer to 
get/post documents from/to apache cassandra.

You can merge this pull request into a Git repository by running:

$ git pull https://github.com/sbcd90/incubator-streams persist-cassandra

Alternatively you can review and apply these changes as the patch at:

https://github.com/apache/incubator-streams/pull/345.patch

To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:

This closes #345


commit 09a5fa35a0da0fb04b450148d817eaebee70fa75
Author: Subhobrata Dey 
Date:   2016-12-24T03:10:46Z

[STREAMS-478] streams persist reader/writer for Apache Cassandra




> streams persist reader/writer for Apache Cassandra
> --
>
> Key: STREAMS-478
> URL: https://issues.apache.org/jira/browse/STREAMS-478
> Project: Streams
>  Issue Type: New Feature
>Reporter: Subhobrata Dey
>
> This issue is created to propose the streams persist reader/writer to 
> get/post documents from/to apache cassandra.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)