openstack/monasca-thresh
Code Issues Proposed changes

Enable monasca-thresh to set topology.max.spout.pending f. Storm

Browse Source 
topology.max.spout.pending allows to limit the number of concurrent
entries sent from spout to worker(s). However, this requires the
usage of a unique id when sending messages (emit).

Change-Id: I907a4574b80e7c3347ba6a9f12c7836767dc3dd7
Story: 2005471
Task: 30550
This commit is contained in:
bandorf 2019-05-22 14:40:22 +02:00
parent de7f4c719c
commit e980ea71e0
1 changed files with 7 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
thresh/src/main/java/monasca/thresh/infrastructure/thresholding/MetricSpout.java
View File

@ -30,8 +30,10 @@ import org.apache.storm.tuple.Values;
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
import java.lang.Thread;
import java.util.Collections; import java.util.Collections;
import java.util.Map; import java.util.Map;
import java.util.concurrent.ThreadLocalRandom;
public class MetricSpout extends KafkaSpout { public class MetricSpout extends KafkaSpout {
@SuppressWarnings("unchecked") @SuppressWarnings("unchecked")
@ -70,8 +72,12 @@ public class MetricSpout extends KafkaSpout {
if (metric.dimensions == null) { if (metric.dimensions == null) {
metric.dimensions = EMPTY_DIMENSIONS; metric.dimensions = EMPTY_DIMENSIONS;
} }
// get unique identifier, required for storm
final String uId = Long.toString(Thread.currentThread().getId())
+ Integer.toString(ThreadLocalRandom.current().nextInt());
collector.emit(new Values(new TenantIdAndMetricName(tenantId, metricEnvelope.metric collector.emit(new Values(new TenantIdAndMetricName(tenantId, metricEnvelope.metric
.definition().name), metricEnvelope.creationTime, metric)); .definition().name), metricEnvelope.creationTime, metric), uId);
} }
@Override @Override