[jira] [Created] (STREAMS-480) Remove -incubating suffix from Release artifacts names
Suneel Marthi created STREAMS-480: - Summary: Remove -incubating suffix from Release artifacts names Key: STREAMS-480 URL: https://issues.apache.org/jira/browse/STREAMS-480 Project: Streams Issue Type: New Feature Components: Build Affects Versions: 0.4.1 Reporter: Suneel Marthi Assignee: Suneel Marthi Fix For: 0.5 Remove the -incubating suffix from future release artifacts - this stems from the discussion here - https://lists.apache.org/thread.html/17ecae7960572e3cad9ba734ea15d55a337422e27ce08d2297a1e830@%3Cgeneral.incubator.apache.org%3E -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Commented] (STREAMS-478) streams persist reader/writer for Apache Cassandra
[ 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)
[GitHub] incubator-streams issue #345: [STREAMS-478] streams persist reader/writer fo...
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. --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[jira] [Commented] (STREAMS-478) streams persist reader/writer for Apache Cassandra
[
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) {
+
[GitHub] incubator-streams pull request #345: [STREAMS-478] streams persist reader/wr...
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) {
+this.config = new ComponentConfigurator<>(CassandraConfiguration.class)
+
.detectConfiguration(StreamsConfigurator.getConfig().getConfig("cassandra"));
+this.persistQueue = persistQueue;
+ }
+
+ public void
[jira] [Commented] (STREAMS-478) streams persist reader/writer for Apache Cassandra
[
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;
+ }
+
+
[GitHub] incubator-streams pull request #345: [STREAMS-478] streams persist reader/wr...
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;
+ }
+
+ @Override
+ public String getId() {
+return STREAMS_ID;
+ }
+
+ @Override
+ public void write(StreamsDatum streamsDatum) {
+
+ObjectNode node;
+
+if (streamsDatum.getDocument() instanceof
[jira] [Commented] (STREAMS-478) streams persist reader/writer for Apache Cassandra
[
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) {
+
[GitHub] incubator-streams pull request #345: [STREAMS-478] streams persist reader/wr...
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) {
+this.config = new ComponentConfigurator<>(CassandraConfiguration.class)
+
.detectConfiguration(StreamsConfigurator.getConfig().getConfig("cassandra"));
+this.persistQueue = persistQueue;
+ }
+
+ public void
[jira] [Commented] (STREAMS-478) streams persist reader/writer for Apache Cassandra
[ 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
[
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)
[GitHub] incubator-streams issue #345: [STREAMS-478] streams persist reader/writer fo...
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
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[jira] [Commented] (STREAMS-477) Fix link for checkstyle.xml
[
https://issues.apache.org/jira/browse/STREAMS-477?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=15783354#comment-15783354
]
Steve Blackmon commented on STREAMS-477:
replace with site/latest/streams-master should work
> Fix link for checkstyle.xml
> ---
>
> Key: STREAMS-477
> URL: https://issues.apache.org/jira/browse/STREAMS-477
> Project: Streams
> Issue Type: Bug
>Affects Versions: 0.4
>Reporter: Suneel Marthi
>Assignee: Steve Blackmon
> Fix For: 0.5
>
>
> Fix the link for checkstyle.xml to be independent of ${project.version},
> since its very unlikely that the checkstyle would be changing frequently.
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
[jira] [Updated] (STREAMS-255) Merge streams-master into streams-project
[ https://issues.apache.org/jira/browse/STREAMS-255?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Suneel Marthi updated STREAMS-255: -- Fix Version/s: (was: 0.2) > Merge streams-master into streams-project > - > > Key: STREAMS-255 > URL: https://issues.apache.org/jira/browse/STREAMS-255 > Project: Streams > Issue Type: Sub-task >Reporter: Steve Blackmon >Assignee: Steve Blackmon > Fix For: 0.5 > > > Merge important parts of streams-master into streams-project and remove > parent relationship. -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Created] (STREAMS-479) build accurate NOTICE file(s) in streams-examples
Steve Blackmon created STREAMS-479: -- Summary: build accurate NOTICE file(s) in streams-examples Key: STREAMS-479 URL: https://issues.apache.org/jira/browse/STREAMS-479 Project: Streams Issue Type: Task Reporter: Steve Blackmon Assignee: Steve Blackmon Fix For: 0.5 containing licenses of shaded dependencies -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Commented] (STREAMS-478) streams persist reader/writer for Apache Cassandra
[
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)
[GitHub] incubator-streams pull request #345: [STREAMS-478] streams persist reader/wr...
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.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-streams pull request #345: [STREAMS-478] streams persist reader/wr...
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;
+ }
+
+ @Override
+ public String getId() {
+return STREAMS_ID;
+ }
+
+ @Override
+ public void write(StreamsDatum streamsDatum) {
+
+ObjectNode node;
+
+if (streamsDatum.getDocument() instanceof
[jira] [Commented] (STREAMS-478) streams persist reader/writer for Apache Cassandra
[
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
[ 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
[
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
[GitHub] incubator-streams pull request #345: [STREAMS-478] streams persist reader/wr...
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 --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] incubator-streams pull request #345: [STREAMS-478] streams persist reader/wr...
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
+ public String getId() {
+return STREAMS_ID;
+ }
+
+ @Override
+ public void write(StreamsDatum streamsDatum) {
+
+ObjectNode node;
+
+if (streamsDatum.getDocument() instanceof String) {
[jira] [Commented] (STREAMS-478) streams persist reader/writer for Apache Cassandra
[
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
[GitHub] incubator-streams pull request #345: [STREAMS-478] streams persist reader/wr...
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 = new ComponentConfigurator<>(CassandraConfiguration.class)
+
.detectConfiguration(StreamsConfigurator.getConfig().getConfig("cassandra"));
+this.persistQueue = persistQueue;
+ }
+
+ public void
