You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@storm.apache.org by HeartSaVioR <gi...@git.apache.org> on 2016/09/13 03:46:08 UTC
[GitHub] storm pull request #1682: STORM-2089 Replace Consumer of ISqlTridentDataSour...
GitHub user HeartSaVioR opened a pull request:
https://github.com/apache/storm/pull/1682
STORM-2089 Replace Consumer of ISqlTridentDataSource with SqlTridentConsumer
* issue link: https://issues.apache.org/jira/browse/STORM-2089
This is for handling INSERT for tuples as batch if possible.
I also changed storm-sql-kafka to expose State directly, but this is effectively same as current since TridentKafkaState.updateState() does update via fully synchronous way.
(by looping each tuple and calling Future.get() immediately)
```
public void updateState(List<TridentTuple> tuples, TridentCollector collector) {
String topic = null;
for (TridentTuple tuple : tuples) {
try {
topic = topicSelector.getTopic(tuple);
if(topic != null) {
Future<RecordMetadata> result = producer.send(new ProducerRecord(topic,
mapper.getKeyFromTuple(tuple), mapper.getMessageFromTuple(tuple)));
try {
result.get();
} catch (ExecutionException e) {
String errorMsg = "Could not retrieve result for message with key = "
+ mapper.getKeyFromTuple(tuple) + " from topic = " + topic;
LOG.error(errorMsg, e);
throw new FailedException(errorMsg, e);
}
} else {
LOG.warn("skipping key = " + mapper.getKeyFromTuple(tuple) + ", topic selector returned null.");
}
} catch (Exception ex) {
String errorMsg = "Could not send message with key = " + mapper.getKeyFromTuple(tuple)
+ " to topic = " + topic;
LOG.warn(errorMsg, ex);
throw new FailedException(errorMsg, ex);
}
}
}
```
We could improve it, but it would be better to handle it as another issue if needed.
Please review and comment. Thanks!
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/HeartSaVioR/storm STORM-2089
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/storm/pull/1682.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 #1682
----
commit cc30363c7550ac34f9f8fed6abc4713a7680ee36
Author: Jungtaek Lim <ka...@gmail.com>
Date: 2016-09-13T02:40:13Z
STORM-2089 Replace Consumer of ISqlTridentDataSource with SqlTridentConsumer
* SqlTridentConsumer contains StateFactory and StateUpdater which is needed to store tuples to State via batch
* Apply the change to storm-sql-kafka
----
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm issue #1682: STORM-2089 Replace Consumer of ISqlTridentDataSource with...
Posted by HeartSaVioR <gi...@git.apache.org>.
Github user HeartSaVioR commented on the issue:
https://github.com/apache/storm/pull/1682
@vesense
Yeah we need to call get() to react for the result. We could still send all requests first, and sync all to reduce waiting time. It's a kind of trade-off since it doesn't do fail-fast.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm issue #1682: STORM-2089 Replace Consumer of ISqlTridentDataSource with...
Posted by HeartSaVioR <gi...@git.apache.org>.
Github user HeartSaVioR commented on the issue:
https://github.com/apache/storm/pull/1682
@arunmahadevan Thanks for the review. I addressed your comment.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #1682: STORM-2089 Replace Consumer of ISqlTridentDataSour...
Posted by vesense <gi...@git.apache.org>.
Github user vesense commented on a diff in the pull request:
https://github.com/apache/storm/pull/1682#discussion_r78509309
--- Diff: external/sql/storm-sql-runtime/src/jvm/org/apache/storm/sql/runtime/ISqlTridentDataSource.java ---
@@ -17,14 +17,32 @@
*/
package org.apache.storm.sql.runtime;
-import org.apache.storm.trident.operation.Function;
-import org.apache.storm.trident.spout.IBatchSpout;
import org.apache.storm.trident.spout.ITridentDataSource;
+import org.apache.storm.trident.state.StateFactory;
+import org.apache.storm.trident.state.StateUpdater;
/**
* A ISqlTridentDataSource specifies how an external data source produces and consumes data.
*/
public interface ISqlTridentDataSource {
+ class SqlTridentConsumer {
+ private StateFactory stateFactory;
+ private StateUpdater stateUpdater;
+
+ public SqlTridentConsumer(StateFactory stateFactory, StateUpdater stateUpdater) {
+ this.stateFactory = stateFactory;
+ this.stateUpdater = stateUpdater;
+ }
+
+ public StateFactory getStateFactory() {
+ return stateFactory;
+ }
+
+ public StateUpdater getStateUpdater() {
+ return stateUpdater;
+ }
+ }
+
ITridentDataSource getProducer();
--- End diff --
how about add some javadoc for interface methods in `ISqlTridentDataSource` and `DataSource`?
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm issue #1682: STORM-2089 Replace Consumer of ISqlTridentDataSource with...
Posted by arunmahadevan <gi...@git.apache.org>.
Github user arunmahadevan commented on the issue:
https://github.com/apache/storm/pull/1682
+1
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm issue #1682: STORM-2089 Replace Consumer of ISqlTridentDataSource with...
Posted by vesense <gi...@git.apache.org>.
Github user vesense commented on the issue:
https://github.com/apache/storm/pull/1682
@HeartSaVioR
one minor comment, others looks good to me.
about sync send `.get()`, please take a look at previous comments: https://github.com/apache/storm/pull/743#commitcomment-14195354
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #1682: STORM-2089 Replace Consumer of ISqlTridentDataSour...
Posted by HeartSaVioR <gi...@git.apache.org>.
Github user HeartSaVioR commented on a diff in the pull request:
https://github.com/apache/storm/pull/1682#discussion_r80415924
--- Diff: external/sql/storm-sql-runtime/src/jvm/org/apache/storm/sql/runtime/ISqlTridentDataSource.java ---
@@ -17,14 +17,62 @@
*/
package org.apache.storm.sql.runtime;
-import org.apache.storm.trident.operation.Function;
-import org.apache.storm.trident.spout.IBatchSpout;
import org.apache.storm.trident.spout.ITridentDataSource;
+import org.apache.storm.trident.state.StateFactory;
+import org.apache.storm.trident.state.StateUpdater;
/**
* A ISqlTridentDataSource specifies how an external data source produces and consumes data.
*/
public interface ISqlTridentDataSource {
+ /**
+ * SqlTridentConsumer is a data structure containing StateFactory and StateUpdater for consuming tuples with State.
+ *
+ * Please note that StateFactory and StateUpdater should use same class which implements State.
+ *
+ * @see org.apache.storm.trident.state.StateFactory
+ * @see org.apache.storm.trident.state.StateUpdater
+ */
+ class SqlTridentConsumer {
--- End diff --
Good point. Will address.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm issue #1682: STORM-2089 Replace Consumer of ISqlTridentDataSource with...
Posted by vesense <gi...@git.apache.org>.
Github user vesense commented on the issue:
https://github.com/apache/storm/pull/1682
+1
and I have filed a jira for improving the TridentKafkaState batch sending.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm issue #1682: STORM-2089 Replace Consumer of ISqlTridentDataSource with...
Posted by HeartSaVioR <gi...@git.apache.org>.
Github user HeartSaVioR commented on the issue:
https://github.com/apache/storm/pull/1682
@vesense Addressed javadoc.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #1682: STORM-2089 Replace Consumer of ISqlTridentDataSour...
Posted by HeartSaVioR <gi...@git.apache.org>.
Github user HeartSaVioR commented on a diff in the pull request:
https://github.com/apache/storm/pull/1682#discussion_r78509897
--- Diff: external/sql/storm-sql-runtime/src/jvm/org/apache/storm/sql/runtime/ISqlTridentDataSource.java ---
@@ -17,14 +17,32 @@
*/
package org.apache.storm.sql.runtime;
-import org.apache.storm.trident.operation.Function;
-import org.apache.storm.trident.spout.IBatchSpout;
import org.apache.storm.trident.spout.ITridentDataSource;
+import org.apache.storm.trident.state.StateFactory;
+import org.apache.storm.trident.state.StateUpdater;
/**
* A ISqlTridentDataSource specifies how an external data source produces and consumes data.
*/
public interface ISqlTridentDataSource {
+ class SqlTridentConsumer {
+ private StateFactory stateFactory;
+ private StateUpdater stateUpdater;
+
+ public SqlTridentConsumer(StateFactory stateFactory, StateUpdater stateUpdater) {
+ this.stateFactory = stateFactory;
+ this.stateUpdater = stateUpdater;
+ }
+
+ public StateFactory getStateFactory() {
+ return stateFactory;
+ }
+
+ public StateUpdater getStateUpdater() {
+ return stateUpdater;
+ }
+ }
+
ITridentDataSource getProducer();
--- End diff --
Makes sense. I'll add javadoc for methods and also SqlTridentConsumer.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #1682: STORM-2089 Replace Consumer of ISqlTridentDataSour...
Posted by vesense <gi...@git.apache.org>.
Github user vesense commented on a diff in the pull request:
https://github.com/apache/storm/pull/1682#discussion_r78509264
--- Diff: external/sql/storm-sql-runtime/src/jvm/org/apache/storm/sql/runtime/ISqlTridentDataSource.java ---
@@ -17,14 +17,32 @@
*/
package org.apache.storm.sql.runtime;
-import org.apache.storm.trident.operation.Function;
-import org.apache.storm.trident.spout.IBatchSpout;
import org.apache.storm.trident.spout.ITridentDataSource;
+import org.apache.storm.trident.state.StateFactory;
+import org.apache.storm.trident.state.StateUpdater;
/**
* A ISqlTridentDataSource specifies how an external data source produces and consumes data.
*/
public interface ISqlTridentDataSource {
+ class SqlTridentConsumer {
+ private StateFactory stateFactory;
+ private StateUpdater stateUpdater;
+
+ public SqlTridentConsumer(StateFactory stateFactory, StateUpdater stateUpdater) {
+ this.stateFactory = stateFactory;
+ this.stateUpdater = stateUpdater;
+ }
+
+ public StateFactory getStateFactory() {
+ return stateFactory;
+ }
+
+ public StateUpdater getStateUpdater() {
+ return stateUpdater;
+ }
+ }
+
ITridentDataSource getProducer();
--- End diff --
how about add some javadoc for interface methods in `ISqlTridentDataSource` and `DataSource`?
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #1682: STORM-2089 Replace Consumer of ISqlTridentDataSour...
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/storm/pull/1682
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] storm pull request #1682: STORM-2089 Replace Consumer of ISqlTridentDataSour...
Posted by arunmahadevan <gi...@git.apache.org>.
Github user arunmahadevan commented on a diff in the pull request:
https://github.com/apache/storm/pull/1682#discussion_r80413517
--- Diff: external/sql/storm-sql-runtime/src/jvm/org/apache/storm/sql/runtime/ISqlTridentDataSource.java ---
@@ -17,14 +17,62 @@
*/
package org.apache.storm.sql.runtime;
-import org.apache.storm.trident.operation.Function;
-import org.apache.storm.trident.spout.IBatchSpout;
import org.apache.storm.trident.spout.ITridentDataSource;
+import org.apache.storm.trident.state.StateFactory;
+import org.apache.storm.trident.state.StateUpdater;
/**
* A ISqlTridentDataSource specifies how an external data source produces and consumes data.
*/
public interface ISqlTridentDataSource {
+ /**
+ * SqlTridentConsumer is a data structure containing StateFactory and StateUpdater for consuming tuples with State.
+ *
+ * Please note that StateFactory and StateUpdater should use same class which implements State.
+ *
+ * @see org.apache.storm.trident.state.StateFactory
+ * @see org.apache.storm.trident.state.StateUpdater
+ */
+ class SqlTridentConsumer {
--- End diff --
Its better to have this as an interface with getStateFactory and getStateUpdater.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---