@Test public void shouldSendDatapointRequestsOnlyOnce() throws Exception { Datapoint datapoint = new Datapoint("metric", timestamp, Collections.emptyMap(), 123); MockedSources sources = new MockedSources(); Testing.withTrackedCluster(clusterParam, (cluster) -> { OpenTsdbTopology topology = new OpenTsdbTopology(makeLaunchEnvironment()); sources.addMockData(OpenTsdbTopology.OTSDB_SPOUT_ID, new Values(null, datapoint), new Values(null, datapoint)); completeTopologyParam.setMockedSources(sources); StormTopology stormTopology = topology.createTopology(); Testing.completeTopology(cluster, stormTopology, completeTopologyParam); }); mockServer.verify(REQUEST, VerificationTimes.exactly(1)); }
@Override public StormTopology createTopology() { logger.info("Creating OpenTsdbTopology - {}", topologyName); TopologyBuilder tb = new TopologyBuilder(); attachInput(tb); OpenTsdbConfig openTsdbConfig = topologyConfig.getOpenTsdbConfig(); tb.setBolt(OTSDB_PARSE_BOLT_ID, new DatapointParseBolt(), openTsdbConfig.getDatapointParseBoltExecutors()) .setNumTasks(openTsdbConfig.getDatapointParseBoltWorkers()) .shuffleGrouping(OTSDB_SPOUT_ID); tb.setBolt(OTSDB_FILTER_BOLT_ID, new OpenTSDBFilterBolt(), openTsdbConfig.getFilterBoltExecutors()) .fieldsGrouping(OTSDB_PARSE_BOLT_ID, new Fields("hash")); OpenTsdbClient.Builder tsdbBuilder = OpenTsdbClient .newBuilder(openTsdbConfig.getHosts()) .returnDetails(); if (openTsdbConfig.getClientChunkedRequestsEnabled()) { tsdbBuilder.enableChunkedEncoding(); } OpenTsdbBolt openTsdbBolt = new OpenTsdbBolt(tsdbBuilder, Collections.singletonList(TupleOpenTsdbDatapointMapper.DEFAULT_MAPPER)); openTsdbBolt.withBatchSize(openTsdbConfig.getBatchSize()).withFlushInterval(openTsdbConfig.getFlushInterval()); tb.setBolt(OTSDB_BOLT_ID, openTsdbBolt, openTsdbConfig.getBoltExecutors()) .setNumTasks(openTsdbConfig.getBoltWorkers()) .shuffleGrouping(OTSDB_FILTER_BOLT_ID); return tb.createTopology(); }
OpenTsdbTopology extends AbstractTopology<OpenTsdbTopologyConfig> { @Override public StormTopology createTopology() { logger.info("Creating OpenTsdbTopology - {}", topologyName); TopologyBuilder tb = new TopologyBuilder(); attachInput(tb); OpenTsdbConfig openTsdbConfig = topologyConfig.getOpenTsdbConfig(); tb.setBolt(OTSDB_PARSE_BOLT_ID, new DatapointParseBolt(), openTsdbConfig.getDatapointParseBoltExecutors()) .setNumTasks(openTsdbConfig.getDatapointParseBoltWorkers()) .shuffleGrouping(OTSDB_SPOUT_ID); tb.setBolt(OTSDB_FILTER_BOLT_ID, new OpenTSDBFilterBolt(), openTsdbConfig.getFilterBoltExecutors()) .fieldsGrouping(OTSDB_PARSE_BOLT_ID, new Fields("hash")); OpenTsdbClient.Builder tsdbBuilder = OpenTsdbClient .newBuilder(openTsdbConfig.getHosts()) .returnDetails(); if (openTsdbConfig.getClientChunkedRequestsEnabled()) { tsdbBuilder.enableChunkedEncoding(); } OpenTsdbBolt openTsdbBolt = new OpenTsdbBolt(tsdbBuilder, Collections.singletonList(TupleOpenTsdbDatapointMapper.DEFAULT_MAPPER)); openTsdbBolt.withBatchSize(openTsdbConfig.getBatchSize()).withFlushInterval(openTsdbConfig.getFlushInterval()); tb.setBolt(OTSDB_BOLT_ID, openTsdbBolt, openTsdbConfig.getBoltExecutors()) .setNumTasks(openTsdbConfig.getBoltWorkers()) .shuffleGrouping(OTSDB_FILTER_BOLT_ID); return tb.createTopology(); } }
OpenTsdbTopology extends AbstractTopology<OpenTsdbTopologyConfig> { @Override public StormTopology createTopology() { logger.info("Creating OpenTsdbTopology - {}", topologyName); TopologyBuilder tb = new TopologyBuilder(); attachInput(tb); OpenTsdbConfig openTsdbConfig = topologyConfig.getOpenTsdbConfig(); tb.setBolt(OTSDB_PARSE_BOLT_ID, new DatapointParseBolt(), openTsdbConfig.getDatapointParseBoltExecutors()) .setNumTasks(openTsdbConfig.getDatapointParseBoltWorkers()) .shuffleGrouping(OTSDB_SPOUT_ID); tb.setBolt(OTSDB_FILTER_BOLT_ID, new OpenTSDBFilterBolt(), openTsdbConfig.getFilterBoltExecutors()) .fieldsGrouping(OTSDB_PARSE_BOLT_ID, new Fields("hash")); OpenTsdbClient.Builder tsdbBuilder = OpenTsdbClient .newBuilder(openTsdbConfig.getHosts()) .returnDetails(); if (openTsdbConfig.getClientChunkedRequestsEnabled()) { tsdbBuilder.enableChunkedEncoding(); } OpenTsdbBolt openTsdbBolt = new OpenTsdbBolt(tsdbBuilder, Collections.singletonList(TupleOpenTsdbDatapointMapper.DEFAULT_MAPPER)); openTsdbBolt.withBatchSize(openTsdbConfig.getBatchSize()).withFlushInterval(openTsdbConfig.getFlushInterval()); tb.setBolt(OTSDB_BOLT_ID, openTsdbBolt, openTsdbConfig.getBoltExecutors()) .setNumTasks(openTsdbConfig.getBoltWorkers()) .shuffleGrouping(OTSDB_FILTER_BOLT_ID); return tb.createTopology(); } OpenTsdbTopology(LaunchEnvironment env); }
OpenTsdbTopology extends AbstractTopology<OpenTsdbTopologyConfig> { @Override public StormTopology createTopology() { logger.info("Creating OpenTsdbTopology - {}", topologyName); TopologyBuilder tb = new TopologyBuilder(); attachInput(tb); OpenTsdbConfig openTsdbConfig = topologyConfig.getOpenTsdbConfig(); tb.setBolt(OTSDB_PARSE_BOLT_ID, new DatapointParseBolt(), openTsdbConfig.getDatapointParseBoltExecutors()) .setNumTasks(openTsdbConfig.getDatapointParseBoltWorkers()) .shuffleGrouping(OTSDB_SPOUT_ID); tb.setBolt(OTSDB_FILTER_BOLT_ID, new OpenTSDBFilterBolt(), openTsdbConfig.getFilterBoltExecutors()) .fieldsGrouping(OTSDB_PARSE_BOLT_ID, new Fields("hash")); OpenTsdbClient.Builder tsdbBuilder = OpenTsdbClient .newBuilder(openTsdbConfig.getHosts()) .returnDetails(); if (openTsdbConfig.getClientChunkedRequestsEnabled()) { tsdbBuilder.enableChunkedEncoding(); } OpenTsdbBolt openTsdbBolt = new OpenTsdbBolt(tsdbBuilder, Collections.singletonList(TupleOpenTsdbDatapointMapper.DEFAULT_MAPPER)); openTsdbBolt.withBatchSize(openTsdbConfig.getBatchSize()).withFlushInterval(openTsdbConfig.getFlushInterval()); tb.setBolt(OTSDB_BOLT_ID, openTsdbBolt, openTsdbConfig.getBoltExecutors()) .setNumTasks(openTsdbConfig.getBoltWorkers()) .shuffleGrouping(OTSDB_FILTER_BOLT_ID); return tb.createTopology(); } OpenTsdbTopology(LaunchEnvironment env); @Override StormTopology createTopology(); static void main(String[] args); }
OpenTsdbTopology extends AbstractTopology<OpenTsdbTopologyConfig> { @Override public StormTopology createTopology() { logger.info("Creating OpenTsdbTopology - {}", topologyName); TopologyBuilder tb = new TopologyBuilder(); attachInput(tb); OpenTsdbConfig openTsdbConfig = topologyConfig.getOpenTsdbConfig(); tb.setBolt(OTSDB_PARSE_BOLT_ID, new DatapointParseBolt(), openTsdbConfig.getDatapointParseBoltExecutors()) .setNumTasks(openTsdbConfig.getDatapointParseBoltWorkers()) .shuffleGrouping(OTSDB_SPOUT_ID); tb.setBolt(OTSDB_FILTER_BOLT_ID, new OpenTSDBFilterBolt(), openTsdbConfig.getFilterBoltExecutors()) .fieldsGrouping(OTSDB_PARSE_BOLT_ID, new Fields("hash")); OpenTsdbClient.Builder tsdbBuilder = OpenTsdbClient .newBuilder(openTsdbConfig.getHosts()) .returnDetails(); if (openTsdbConfig.getClientChunkedRequestsEnabled()) { tsdbBuilder.enableChunkedEncoding(); } OpenTsdbBolt openTsdbBolt = new OpenTsdbBolt(tsdbBuilder, Collections.singletonList(TupleOpenTsdbDatapointMapper.DEFAULT_MAPPER)); openTsdbBolt.withBatchSize(openTsdbConfig.getBatchSize()).withFlushInterval(openTsdbConfig.getFlushInterval()); tb.setBolt(OTSDB_BOLT_ID, openTsdbBolt, openTsdbConfig.getBoltExecutors()) .setNumTasks(openTsdbConfig.getBoltWorkers()) .shuffleGrouping(OTSDB_FILTER_BOLT_ID); return tb.createTopology(); } OpenTsdbTopology(LaunchEnvironment env); @Override StormTopology createTopology(); static void main(String[] args);  }