DRPC #37
DRPC #37
Conversation
|
|
||
| Number connectTimeout = config.getRequiredConfigAs(DRPCConfig.DRPC_HTTP_CONNECT_TIMEOUT_MS, Number.class); | ||
| Number retryLimit = config.getRequiredConfigAs(DRPCConfig.DRPC_HTTP_CONNECT_RETRY_LIMIT, Number.class); | ||
| List<String> urls = (List<String>) config.getRequiredConfigAs(DRPCConfig.DRPC_SERVERS, List.class); |
There was a problem hiding this comment.
This local var name is the same as line 38 and make it a little bit hard to read. Can you change to a different name?
| bullet.topology.join.bolt.cpu.load: 100.0 | ||
| bullet.topology.join.bolt.memory.on.heap.load: 512.0 | ||
| bullet.topology.join.bolt.memory.off.heap.load: 160.0 | ||
| bullet.topology.join.bolt.parallelism: 20 | ||
| bullet.topology.join.bolt.parallelism: 2 |
There was a problem hiding this comment.
The numbers are changed a lot. Is there any reason for it?
There was a problem hiding this comment.
For defaults, these were too high relative to the filter bolt parallelisms, so I dropped it. This parallelism is sufficient for hundreds of simultaneous queries.
| } | ||
|
|
||
| @Override | ||
| public void nextTuple() { | ||
| PubSubMessage message = null; | ||
| try { | ||
| message = subscriber.receive(); | ||
| } catch (PubSubException e) { | ||
| } catch (Exception e) { |
There was a problem hiding this comment.
What else exception does it throw? I checked the interface definition, only PubSubException thrown.
There was a problem hiding this comment.
To catch RuntimeExceptions. Basically if anything bad happens for any reason, we don't want to restart the spout (ex. bad JSON in PubSubMessage throws a GSON RuntimeException)
| @@ -51,8 +71,9 @@ public void nextTuple() { | |||
| log.error(e.getMessage()); | |||
| } | |||
| if (message != null) { | |||
| // TODO: No need for two streams. Just send a unified Query stream. JoinBolt needs to not do that join. | |||
There was a problem hiding this comment.
"JoinBolt need not do that join."
Also, maybe call message.getId() just once and save to value to avoid calling the function 4 times?
| // The path that queries must be POSTed to. This generally is "drpc". | ||
| public static final String DRPC_HTTP_PATH = PREFIX + "http.path"; | ||
|
|
||
|
|
There was a problem hiding this comment.
Did you want two blank lines here?
| } | ||
| return IntStream.range(0, n).mapToObj(i -> new DRPCQuerySubscriber(config, maxUncommittedMessages)) | ||
| .collect(Collectors.toList()); | ||
|
|
| @Override | ||
| public void close() { | ||
| log.warn("Failing all pending requests: {}", emittedIDs); | ||
| emittedIDs.values().forEach(spout::fail); |
There was a problem hiding this comment.
As we discussed, messages are not removed from emittedIDs when a message is ack'ed... Override commit() and do that?
|
I tested this out on a Storm cluster with a custom build of bullet-service using the DRPC pubsub. Looks to be working. |
This builds off the initial DRPC work from @ShriramKumar and redoes it without replicating the Storm code. Feel free to take a look but I'll wire this up with an actual instance to test first.