You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@storm.apache.org by rz...@apache.org on 2023/10/19 07:25:33 UTC
[storm] 02/13: STORM-3988 - Remove "storm-eventhubs"
This is an automated email from the ASF dual-hosted git repository.
rzo1 pushed a commit to branch STORM-3988
in repository https://gitbox.apache.org/repos/asf/storm.git
commit 7f97a03aa4208776a06b664c7f6bb262c09f8dea
Author: Richard Zowalla <ri...@hs-heilbronn.de>
AuthorDate: Thu Oct 19 08:57:13 2023 +0200
STORM-3988 - Remove "storm-eventhubs"
---
external/storm-eventhubs/README.md | 45 ----
external/storm-eventhubs/pom.xml | 85 -------
.../eventhubs/bolt/DefaultEventDataFormat.java | 47 ----
.../apache/storm/eventhubs/bolt/EventHubBolt.java | 139 -----------
.../storm/eventhubs/bolt/EventHubBoltConfig.java | 108 ---------
.../storm/eventhubs/bolt/IEventDataFormat.java | 29 ---
.../eventhubs/spout/BinaryEventDataScheme.java | 72 ------
.../storm/eventhubs/spout/EventDataScheme.java | 75 ------
.../storm/eventhubs/spout/EventDataWrap.java | 48 ----
.../storm/eventhubs/spout/EventHubException.java | 41 ----
.../storm/eventhubs/spout/EventHubFilter.java | 47 ----
.../eventhubs/spout/EventHubReceiverImpl.java | 169 -------------
.../storm/eventhubs/spout/EventHubSpout.java | 258 --------------------
.../storm/eventhubs/spout/EventHubSpoutConfig.java | 264 ---------------------
.../storm/eventhubs/spout/FieldConstants.java | 29 ---
.../storm/eventhubs/spout/IEventDataScheme.java | 44 ----
.../apache/storm/eventhubs/spout/IEventFilter.java | 28 ---
.../storm/eventhubs/spout/IEventHubReceiver.java | 34 ---
.../eventhubs/spout/IEventHubReceiverFactory.java | 25 --
.../eventhubs/spout/IPartitionCoordinator.java | 28 ---
.../storm/eventhubs/spout/IPartitionManager.java | 38 ---
.../eventhubs/spout/IPartitionManagerFactory.java | 28 ---
.../apache/storm/eventhubs/spout/IStateStore.java | 32 ---
.../apache/storm/eventhubs/spout/MessageId.java | 57 -----
.../storm/eventhubs/spout/PartitionManager.java | 101 --------
.../eventhubs/spout/SerializeDeserializeUtil.java | 34 ---
.../eventhubs/spout/SimplePartitionManager.java | 133 -----------
.../spout/StaticPartitionCoordinator.java | 84 -------
.../eventhubs/spout/StringEventDataScheme.java | 71 ------
.../storm/eventhubs/spout/ZookeeperStateStore.java | 96 --------
.../storm/eventhubs/trident/Coordinator.java | 59 -----
.../trident/ITridentPartitionManager.java | 37 ---
.../trident/ITridentPartitionManagerFactory.java | 26 --
.../trident/OpaqueTridentEventHubEmitter.java | 69 ------
.../trident/OpaqueTridentEventHubSpout.java | 63 -----
.../apache/storm/eventhubs/trident/Partition.java | 40 ----
.../apache/storm/eventhubs/trident/Partitions.java | 43 ----
.../TransactionalTridentEventHubEmitter.java | 161 -------------
.../trident/TransactionalTridentEventHubSpout.java | 64 -----
.../eventhubs/trident/TridentPartitionManager.java | 89 -------
.../src/main/resources/config.properties | 45 ----
.../eventhubs/samples/AtMostOnceEventCount.java | 54 -----
.../apache/storm/eventhubs/samples/EventCount.java | 144 -----------
.../storm/eventhubs/samples/EventHubLoop.java | 52 ----
.../eventhubs/samples/OpaqueTridentEventCount.java | 54 -----
.../samples/TransactionalTridentEventCount.java | 81 -------
.../eventhubs/samples/bolt/GlobalCountBolt.java | 87 -------
.../eventhubs/samples/bolt/PartialCountBolt.java | 67 ------
.../eventhubs/spout/EventHubReceiverMock.java | 84 -------
.../eventhubs/spout/EventHubSpoutCallerMock.java | 92 -------
.../spout/PartitionManagerCallerMock.java | 95 --------
.../eventhubs/spout/SpoutOutputCollectorMock.java | 71 ------
.../storm/eventhubs/spout/StateStoreMock.java | 54 -----
.../storm/eventhubs/spout/TestEventData.java | 48 ----
.../storm/eventhubs/spout/TestEventHubSpout.java | 75 ------
.../eventhubs/spout/TestPartitionManager.java | 119 ----------
.../trident/TestTransactionalTridentEmitter.java | 92 -------
.../eventhubs/trident/TridentCollectorMock.java | 57 -----
pom.xml | 1 -
59 files changed, 4312 deletions(-)
diff --git a/external/storm-eventhubs/README.md b/external/storm-eventhubs/README.md
deleted file mode 100755
index 98b43a5cb..000000000
--- a/external/storm-eventhubs/README.md
+++ /dev/null
@@ -1,45 +0,0 @@
-storm-eventhubs
-=====================
-
-Storm spout and bolt implementation for Microsoft Azure Eventhubs
-
-### build ###
- mvn clean package
-
-### run sample topology ###
-To run the sample topology, you need to modify the config.properties file with
-the eventhubs configurations. Here is an example:
-
- eventhubspout.username = [username: policy name in EventHubs Portal]
- eventhubspout.password = [password: shared access key in EventHubs Portal]
- eventhubspout.namespace = [namespace]
- eventhubspout.entitypath = [entitypath]
- eventhubspout.partitions.count = [partitioncount]
-
- # if not provided, will use storm's zookeeper settings
- # zookeeper.connectionstring=zookeeper0:2181,zookeeper1:2181,zookeeper2:2181
-
- eventhubspout.checkpoint.interval = 10
- eventhub.receiver.credits = 1024
-
-Then you can use storm.cmd to submit the sample topology:
- storm jar {jarfile} com.microsoft.eventhubs.samples.EventCount {topologyname} {spoutconffile}
- where the {jarfile} should be: eventhubs-storm-spout-{version}-jar-with-dependencies.jar
-
-### Run EventHubSendClient ###
-We have included a simple EventHubs send client for testing purpose. You can run the client like this:
- java -cp .\target\eventhubs-storm-spout-{version}-jar-with-dependencies.jar com.microsoft.eventhubs.client.EventHubSendClient
- [username] [password] [entityPath] [partitionId] [messageSize] [messageCount]
-If you want to send messages to all partitions, use "-1" as partitionId.
-
-### Serialization Scheme ###
- By default the serialization scheme is StringEventDataScheme where only the body of the eventdata is being sent.
- To have a more comprehensive Eventhub metadata exposure look into the BinaryEventDataScheme.
-
-### Windows Azure Eventhubs ###
- http://azure.microsoft.com/en-us/services/event-hubs/
-
-## Committer Sponsors
-
- * P. Taylor Goetz ([ptgoetz@apache.org](mailto:ptgoetz@apache.org))
-
diff --git a/external/storm-eventhubs/pom.xml b/external/storm-eventhubs/pom.xml
deleted file mode 100755
index 65f845ccb..000000000
--- a/external/storm-eventhubs/pom.xml
+++ /dev/null
@@ -1,85 +0,0 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<!--
- 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 with
- the License. You may obtain a copy of the License at
-
- 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.
--->
-<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
- <modelVersion>4.0.0</modelVersion>
-
- <parent>
- <artifactId>storm</artifactId>
- <groupId>org.apache.storm</groupId>
- <version>2.6.0-SNAPSHOT</version>
- <relativePath>../../pom.xml</relativePath>
- </parent>
-
- <artifactId>storm-eventhubs</artifactId>
- <packaging>jar</packaging>
- <name>storm-eventhubs</name>
- <description>EventHubs Storm Spout</description>
-
- <build>
- <plugins>
- <plugin>
- <artifactId>maven-antrun-plugin</artifactId>
- <executions>
- <execution>
- <phase>package</phase>
- <configuration>
- <target>
- <copy file="src/main/resources/config.properties" tofile="target/eventhubs-config.properties" />
- </target>
- </configuration>
- <goals>
- <goal>run</goal>
- </goals>
- </execution>
- </executions>
- </plugin>
- <plugin>
- <groupId>org.apache.maven.plugins</groupId>
- <artifactId>maven-checkstyle-plugin</artifactId>
- <!--Note - the version would be inherited-->
- </plugin>
- <plugin>
- <groupId>org.apache.maven.plugins</groupId>
- <artifactId>maven-pmd-plugin</artifactId>
- </plugin>
- </plugins>
- </build>
- <dependencies>
- <dependency>
- <groupId>com.microsoft.azure</groupId>
- <artifactId>azure-eventhubs</artifactId>
- <version>${azure-eventhubs.version}</version>
- </dependency>
- <dependency>
- <groupId>org.apache.storm</groupId>
- <artifactId>storm-client</artifactId>
- <version>${project.version}</version>
- <!-- keep storm out of the jar-with-dependencies -->
- <type>jar</type>
- <scope>${provided.scope}</scope>
- </dependency>
- <dependency>
- <groupId>org.apache.curator</groupId>
- <artifactId>curator-framework</artifactId>
- </dependency>
- <dependency>
- <groupId>org.apache.zookeeper</groupId>
- <artifactId>zookeeper</artifactId>
- </dependency>
- </dependencies>
-</project>
diff --git a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/bolt/DefaultEventDataFormat.java b/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/bolt/DefaultEventDataFormat.java
deleted file mode 100644
index 094ab9a6c..000000000
--- a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/bolt/DefaultEventDataFormat.java
+++ /dev/null
@@ -1,47 +0,0 @@
-/*******************************************************************************
- * 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
- * with the License. You may obtain a copy of the License at
- *
- * 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.storm.eventhubs.bolt;
-
-import org.apache.storm.tuple.Tuple;
-
-/**
- * A default implementation of IEventDataFormat that converts the tuple into a delimited string.
- */
-public class DefaultEventDataFormat implements IEventDataFormat {
- private static final long serialVersionUID = 1L;
- private String delimiter = ",";
-
- public DefaultEventDataFormat withFieldDelimiter(String delimiter) {
- this.delimiter = delimiter;
- return this;
- }
-
- @Override
- public byte[] serialize(Tuple tuple) {
- StringBuilder sb = new StringBuilder();
- for (Object obj : tuple.getValues()) {
- if (sb.length() != 0) {
- sb.append(delimiter);
- }
- sb.append(obj.toString());
- }
- return sb.toString().getBytes();
- }
-
-}
diff --git a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/bolt/EventHubBolt.java b/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/bolt/EventHubBolt.java
deleted file mode 100755
index 4c1e06843..000000000
--- a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/bolt/EventHubBolt.java
+++ /dev/null
@@ -1,139 +0,0 @@
-/*******************************************************************************
- * 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
- * with the License. You may obtain a copy of the License at
- *
- * 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.storm.eventhubs.bolt;
-
-import com.microsoft.azure.eventhubs.EventData;
-import com.microsoft.azure.eventhubs.EventHubClient;
-import com.microsoft.azure.eventhubs.PartitionSender;
-import com.microsoft.azure.servicebus.ServiceBusException;
-import java.util.Map;
-import java.util.concurrent.ExecutionException;
-import org.apache.storm.eventhubs.spout.EventHubException;
-import org.apache.storm.task.OutputCollector;
-import org.apache.storm.task.TopologyContext;
-import org.apache.storm.topology.OutputFieldsDeclarer;
-import org.apache.storm.topology.base.BaseRichBolt;
-import org.apache.storm.tuple.Tuple;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * A bolt that writes event message to EventHub.
- */
-public class EventHubBolt extends BaseRichBolt {
- private static final long serialVersionUID = 1L;
- private static final Logger logger = LoggerFactory
- .getLogger(EventHubBolt.class);
-
- protected OutputCollector collector;
- protected PartitionSender sender;
- protected EventHubClient ehClient;
- protected EventHubBoltConfig boltConfig;
-
- public EventHubBolt(String connectionString, String entityPath) {
- boltConfig = new EventHubBoltConfig(connectionString, entityPath);
- }
-
- public EventHubBolt(String userName, String password, String namespace,
- String entityPath, boolean partitionMode) {
- boltConfig = new EventHubBoltConfig(userName, password, namespace,
- entityPath, partitionMode);
- }
-
- public EventHubBolt(EventHubBoltConfig config) {
- boltConfig = config;
- }
-
- @Override
- public void prepare(Map<String, Object> config, TopologyContext context,
- OutputCollector collector) {
- this.collector = collector;
- String myPartitionId = null;
- if (boltConfig.getPartitionMode()) {
- // We can use the task index (starting from 0) as the partition ID
- myPartitionId = "" + context.getThisTaskIndex();
- }
- logger.info("creating sender: " + boltConfig.getConnectionString()
- + ", " + boltConfig.getEntityPath() + ", " + myPartitionId);
- try {
- ehClient = EventHubClient.createFromConnectionStringSync(boltConfig.getConnectionString());
- if (boltConfig.getPartitionMode()) {
- sender = ehClient.createPartitionSenderSync(Integer.toString(context.getThisTaskIndex()));
- }
- } catch (Exception ex) {
- collector.reportError(ex);
- throw new RuntimeException(ex);
- }
-
- }
-
- @Override
- public void execute(Tuple tuple) {
- try {
- EventData sendEvent = new EventData(boltConfig.getEventDataFormat().serialize(tuple));
- if (boltConfig.getPartitionMode() && sender != null) {
- sender.sendSync(sendEvent);
- } else if (boltConfig.getPartitionMode() && sender == null) {
- throw new EventHubException("Sender is null");
- } else if (!boltConfig.getPartitionMode() && ehClient != null) {
- ehClient.sendSync(sendEvent);
- } else if (!boltConfig.getPartitionMode() && ehClient == null) {
- throw new EventHubException("ehclient is null");
- }
- collector.ack(tuple);
- } catch (EventHubException ex) {
- collector.reportError(ex);
- collector.fail(tuple);
- } catch (ServiceBusException e) {
- collector.reportError(e);
- collector.fail(tuple);
- }
- }
-
- @Override
- public void cleanup() {
- if (sender != null) {
- try {
- sender.close().whenComplete((voidargs, error) -> {
- try {
- if (error != null) {
- logger.error("Exception during sender cleanup phase" + error.toString());
- }
- ehClient.closeSync();
- } catch (Exception e) {
- logger.error("Exception during ehclient cleanup phase" + e.toString());
- }
- }).get();
- } catch (InterruptedException e) {
- logger.error("Exception occured during cleanup phase" + e.toString());
- } catch (ExecutionException e) {
- logger.error("Exception occured during cleanup phase" + e.toString());
- }
- logger.info("Eventhub Bolt cleaned up");
- sender = null;
- ehClient = null;
- }
- }
-
- @Override
- public void declareOutputFields(OutputFieldsDeclarer declarer) {
-
- }
-
-}
diff --git a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/bolt/EventHubBoltConfig.java b/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/bolt/EventHubBoltConfig.java
deleted file mode 100644
index 41e39e4e8..000000000
--- a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/bolt/EventHubBoltConfig.java
+++ /dev/null
@@ -1,108 +0,0 @@
-/*******************************************************************************
- * 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
- * with the License. You may obtain a copy of the License at
- *
- * 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.storm.eventhubs.bolt;
-
-import com.microsoft.azure.servicebus.ConnectionStringBuilder;
-import java.io.Serializable;
-import org.apache.storm.eventhubs.spout.EventHubSpoutConfig;
-
-/*
- * EventHubs bolt configurations
- *
- * Partition mode:
- * With partitionMode=true you need to create the same number of tasks as the number of
- * EventHubs partitions, and each bolt task will only send data to one partition.
- * The partition ID is the task ID of the bolt.
- *
- * Event format:
- * The formatter to convert tuple to bytes for EventHubs.
- * if null, the default format is common delimited tuple fields.
- */
-public class EventHubBoltConfig implements Serializable {
- private static final long serialVersionUID = 1L;
- private final String entityPath;
- protected boolean partitionMode;
- protected IEventDataFormat dataFormat;
- private String connectionString;
-
- public EventHubBoltConfig(String connectionString, String entityPath) {
- this(connectionString, entityPath, false, null);
- }
-
- public EventHubBoltConfig(String connectionString, String entityPath,
- boolean partitionMode) {
- this(connectionString, entityPath, partitionMode, null);
- }
-
- public EventHubBoltConfig(String userName, String password, String namespace,
- String entityPath, boolean partitionMode) {
- this(userName, password, namespace,
- EventHubSpoutConfig.EH_SERVICE_FQDN_SUFFIX, entityPath, partitionMode);
- }
-
- public EventHubBoltConfig(String connectionString, String entityPath,
- boolean partitionMode, IEventDataFormat dataFormat) {
- this.connectionString = connectionString;
- this.entityPath = entityPath;
- this.partitionMode = partitionMode;
- this.dataFormat = dataFormat;
- if (this.dataFormat == null) {
- this.dataFormat = new DefaultEventDataFormat();
- }
- }
-
- public EventHubBoltConfig(String userName, String password, String namespace,
- String targetFqnAddress, String entityPath) {
- this(userName, password, namespace, targetFqnAddress, entityPath, false, null);
- }
-
- public EventHubBoltConfig(String userName, String password, String namespace,
- String targetFqnAddress, String entityPath, boolean partitionMode) {
- this(userName, password, namespace, targetFqnAddress, entityPath, partitionMode, null);
- }
-
- public EventHubBoltConfig(String userName, String password, String namespace,
- String targetFqnAddress, String entityPath, boolean partitionMode,
- IEventDataFormat dataFormat) {
- this.connectionString = new ConnectionStringBuilder(namespace, entityPath,
- userName, password).toString();
- this.entityPath = entityPath;
- this.partitionMode = partitionMode;
- this.dataFormat = dataFormat;
- if (this.dataFormat == null) {
- this.dataFormat = new DefaultEventDataFormat();
- }
- }
-
- public String getConnectionString() {
- return connectionString;
- }
-
- public String getEntityPath() {
- return entityPath;
- }
-
- public boolean getPartitionMode() {
- return partitionMode;
- }
-
- public IEventDataFormat getEventDataFormat() {
- return dataFormat;
- }
-}
diff --git a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/bolt/IEventDataFormat.java b/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/bolt/IEventDataFormat.java
deleted file mode 100644
index fb03a7cb6..000000000
--- a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/bolt/IEventDataFormat.java
+++ /dev/null
@@ -1,29 +0,0 @@
-/*******************************************************************************
- * 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
- * with the License. You may obtain a copy of the License at
- *
- * 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.storm.eventhubs.bolt;
-
-import java.io.Serializable;
-import org.apache.storm.tuple.Tuple;
-
-/**
- * Serialize a tuple to a byte array to be sent to EventHubs.
- */
-public interface IEventDataFormat extends Serializable {
- byte[] serialize(Tuple tuple);
-}
diff --git a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/BinaryEventDataScheme.java b/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/BinaryEventDataScheme.java
deleted file mode 100644
index 40db61a73..000000000
--- a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/BinaryEventDataScheme.java
+++ /dev/null
@@ -1,72 +0,0 @@
-/*******************************************************************************
- * 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
- * with the License. You may obtain a copy of the License at
- *
- * 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.storm.eventhubs.spout;
-
-import com.microsoft.azure.eventhubs.EventData;
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Map;
-import org.apache.storm.tuple.Fields;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-
-/**
- * An Event Data Scheme which deserializes message payload into the raw bytes.
- *
- * <p>The resulting tuple would contain three items, the first being the message bytes, and the second a map of properties that include
- * metadata, which can be used to determine who processes the message, and how it is processed.The third is the system properties which
- * exposes information like enqueue-time, offset and sequence number
- */
-public class BinaryEventDataScheme implements IEventDataScheme {
-
- private static final Logger logger = LoggerFactory.getLogger(BinaryEventDataScheme.class);
-
- @Override
- public List<Object> deserialize(EventData eventData) {
- final List<Object> fieldContents = new ArrayList<Object>();
- byte[] messageData = null;
- if (eventData.getBytes() != null) {
- messageData = eventData.getBytes();
- } else if (eventData.getObject() != null) {
- try {
- messageData = SerializeDeserializeUtil.serialize(eventData.getObject());
- } catch (IOException e) {
- logger.error("Failed to serialize EventData payload class"
- + eventData.getObject().getClass());
- logger.error("Exception encountered while serializing EventData payload is"
- + e.toString());
- throw new RuntimeException(e);
- }
- }
- Map<String, Object> metaDataMap = eventData.getProperties();
- Map<String, Object> systemMetaDataMap = eventData.getSystemProperties();
- fieldContents.add(messageData);
- fieldContents.add(metaDataMap);
- fieldContents.add(systemMetaDataMap);
- return fieldContents;
- }
-
- @Override
- public Fields getOutputFields() {
- return new Fields(FieldConstants.Message, FieldConstants.META_DATA,
- FieldConstants.SYSTEM_META_DATA);
- }
-}
diff --git a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/EventDataScheme.java b/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/EventDataScheme.java
deleted file mode 100755
index 40b803420..000000000
--- a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/EventDataScheme.java
+++ /dev/null
@@ -1,75 +0,0 @@
-/*******************************************************************************
- * 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
- * with the License. You may obtain a copy of the License at
- *
- * 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.storm.eventhubs.spout;
-
-import com.microsoft.azure.eventhubs.EventData;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Map;
-import org.apache.storm.tuple.Fields;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * An Event Data Scheme which deserializes message payload into the Strings. No encoding is assumed. The receiver will need to handle
- * parsing of the string data in appropriate encoding.
- *
- * <p>The resulting tuple would contain two items: the the message string, and a map of properties that include
- * metadata, which can be used to determine who processes the message, and how it is processed.
- *
- * <p>For passing the raw bytes of a messsage to Bolts, refer to {@link BinaryEventDataScheme}.
- */
-public class EventDataScheme implements IEventDataScheme {
-
- private static final long serialVersionUID = 1L;
- private static final Logger logger = LoggerFactory.getLogger(EventDataScheme.class);
-
- @Override
- public List<Object> deserialize(EventData eventData) {
- final List<Object> fieldContents = new ArrayList<Object>();
- String messageData = "";
- if (eventData.getBytes() != null) {
- messageData = new String(eventData.getBytes());
- } else if (eventData.getObject() != null) {
- /*Will only serialize AMQPValue type*/
- try {
- if (!(eventData.getObject() instanceof List)) {
- messageData = eventData.getObject().toString();
- } else {
- throw new RuntimeException("Cannot serialize the given AMQP type");
- }
- } catch (RuntimeException e) {
- logger.error("Failed to serialize EventData payload class"
- + eventData.getObject().getClass());
- logger.error("Exception encountered while serializing EventData payload is"
- + e.toString());
- throw e;
- }
- }
- Map<String, Object> metaDataMap = eventData.getProperties();
- fieldContents.add(messageData);
- fieldContents.add(metaDataMap);
- return fieldContents;
- }
-
- @Override
- public Fields getOutputFields() {
- return new Fields(FieldConstants.Message, FieldConstants.META_DATA);
- }
-}
diff --git a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/EventDataWrap.java b/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/EventDataWrap.java
deleted file mode 100644
index a331c6919..000000000
--- a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/EventDataWrap.java
+++ /dev/null
@@ -1,48 +0,0 @@
-/*******************************************************************************
- * 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
- * with the License. You may obtain a copy of the License at
- *
- * 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.storm.eventhubs.spout;
-
-import com.microsoft.azure.eventhubs.EventData;
-
-public class EventDataWrap implements Comparable<EventDataWrap> {
- private final EventData eventData;
- private final MessageId messageId;
-
- public EventDataWrap(EventData eventdata, MessageId messageId) {
- this.eventData = eventdata;
- this.messageId = messageId;
- }
-
- public static EventDataWrap create(EventData eventData, MessageId messageId) {
- return new EventDataWrap(eventData, messageId);
- }
-
- public EventData getEventData() {
- return this.eventData;
- }
-
- public MessageId getMessageId() {
- return this.messageId;
- }
-
- @Override
- public int compareTo(EventDataWrap ed) {
- return messageId.getSequenceNumber().compareTo(ed.getMessageId().getSequenceNumber());
- }
-}
diff --git a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/EventHubException.java b/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/EventHubException.java
deleted file mode 100644
index 6c7824b44..000000000
--- a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/EventHubException.java
+++ /dev/null
@@ -1,41 +0,0 @@
-/*******************************************************************************
- * 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
- * with the License. You may obtain a copy of the License at
- *
- * 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.storm.eventhubs.spout;
-
-public class EventHubException extends Exception {
-
- private static final long serialVersionUID = 1L;
-
- public EventHubException() {
- super();
- }
-
- public EventHubException(String message) {
- super(message);
- }
-
- public EventHubException(Throwable cause) {
- super(cause);
- }
-
- public EventHubException(String message, Throwable cause) {
- super(message, cause);
- }
-
-}
diff --git a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/EventHubFilter.java b/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/EventHubFilter.java
deleted file mode 100644
index 509191d08..000000000
--- a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/EventHubFilter.java
+++ /dev/null
@@ -1,47 +0,0 @@
-/*******************************************************************************
- * 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
- * with the License. You may obtain a copy of the License at
- *
- * 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.storm.eventhubs.spout;
-
-import java.time.Instant;
-
-public class EventHubFilter implements IEventFilter {
-
- String offset = null;
- Instant time = null;
-
- public EventHubFilter(String offset) {
- this.offset = offset;
- this.time = null;
- }
-
- public EventHubFilter(Instant time) {
- this.time = time;
- this.offset = null;
- }
-
- @Override
- public String getOffset() {
- return offset;
- }
-
- @Override
- public Instant getTime() {
- return time;
- }
-}
diff --git a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/EventHubReceiverImpl.java b/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/EventHubReceiverImpl.java
deleted file mode 100755
index 392adaf93..000000000
--- a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/EventHubReceiverImpl.java
+++ /dev/null
@@ -1,169 +0,0 @@
-/*******************************************************************************
- * 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
- * with the License. You may obtain a copy of the License at
- *
- * 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.storm.eventhubs.spout;
-
-import com.microsoft.azure.eventhubs.EventData;
-import com.microsoft.azure.eventhubs.EventHubClient;
-import com.microsoft.azure.eventhubs.PartitionReceiver;
-import com.microsoft.azure.servicebus.ServiceBusException;
-import java.io.IOException;
-import java.time.Duration;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.concurrent.ExecutionException;
-import org.apache.storm.metric.api.CountMetric;
-import org.apache.storm.metric.api.MeanReducer;
-import org.apache.storm.metric.api.ReducedMetric;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class EventHubReceiverImpl implements IEventHubReceiver {
- private static final Logger logger = LoggerFactory.getLogger(EventHubReceiverImpl.class);
-
- private final String connectionString;
- private final String entityName;
- private final String partitionId;
- private final String consumerGroupName;
- private final int receiverTimeoutInMillis;
-
- private PartitionReceiver receiver;
- private EventHubClient ehClient;
- private ReducedMetric receiveApiLatencyMean;
- private CountMetric receiveApiCallCount;
- private CountMetric receiveMessageCount;
-
- public EventHubReceiverImpl(EventHubSpoutConfig config, String partitionId) {
- this.connectionString = config.getConnectionString();
- this.entityName = config.getEntityPath();
- this.partitionId = partitionId;
- this.consumerGroupName = config.getConsumerGroupName();
- this.receiverTimeoutInMillis = config.getReceiverTimeoutInMillis();
- receiveApiLatencyMean = new ReducedMetric(new MeanReducer());
- receiveApiCallCount = new CountMetric();
- receiveMessageCount = new CountMetric();
- }
-
- @Override
- public void open(IEventFilter filter) throws EventHubException {
- logger.info("creating eventhub receiver: partitionId=" + partitionId
- + ", filter=" + filter.getOffset() != null
- ? filter.getOffset()
- : Long.toString(filter.getTime().toEpochMilli()));
- long start = System.currentTimeMillis();
- try {
- ehClient = EventHubClient.createFromConnectionStringSync(connectionString);
-
- if (filter.getOffset() != null) {
- receiver = ehClient.createEpochReceiverSync(
- consumerGroupName,
- partitionId,
- filter.getOffset(),
- false,
- 1);
- } else if (filter.getTime() != null) {
- receiver = ehClient.createEpochReceiverSync(
- consumerGroupName,
- partitionId,
- filter.getTime(),
- 1);
- } else {
- throw new RuntimeException("Eventhub receiver must have an offset or time to be created");
- }
- if (receiver != null) {
- receiver.setReceiveTimeout(Duration.ofMillis(receiverTimeoutInMillis));
- }
- } catch (IOException e) {
- logger.error("Exception in creating ehclient" + e.toString());
- throw new EventHubException(e);
- } catch (ServiceBusException e) {
- logger.error("Exception in creating Receiver" + e.toString());
- throw new EventHubException(e);
- }
- long end = System.currentTimeMillis();
- logger.info("created eventhub receiver, time taken(ms): " + (end - start));
- }
-
- @Override
- public void close() {
- if (receiver != null) {
- try {
- receiver.close().whenComplete((voidargs, error) -> {
- try {
- if (error != null) {
- logger.error("Exception during receiver close phase" + error.toString());
- }
- ehClient.closeSync();
- } catch (Exception e) {
- logger.error("Exception during ehclient close phase" + e.toString());
- }
- }).get();
- } catch (InterruptedException e) {
- logger.error("Exception occured during close phase" + e.toString());
- } catch (ExecutionException e) {
- logger.error("Exception occured during close phase" + e.toString());
- }
- logger.info("closed eventhub receiver: partitionId=" + partitionId);
- receiver = null;
- ehClient = null;
- }
- }
-
-
- @Override
- public boolean isOpen() {
- return (receiver != null);
- }
-
- @Override
- public EventDataWrap receive() {
- long start = System.currentTimeMillis();
- Iterable<EventData> receivedEvents = null;
- /*Get one message at a time for backward compatibility behaviour*/
- try {
- receivedEvents = receiver.receiveSync(1);
- } catch (ServiceBusException e) {
- logger.error("Exception occured during receive" + e.toString());
- return null;
- }
- long end = System.currentTimeMillis();
- long millis = (end - start);
- receiveApiLatencyMean.update(millis);
- receiveApiCallCount.incr();
-
- if (receivedEvents == null || receivedEvents.spliterator().getExactSizeIfKnown() == 0) {
- return null;
- }
- receiveMessageCount.incr();
- EventData receivedEvent = receivedEvents.iterator().next();
- MessageId messageId = new MessageId(partitionId,
- receivedEvent.getSystemProperties().getOffset(),
- receivedEvent.getSystemProperties().getSequenceNumber());
-
- return EventDataWrap.create(receivedEvent, messageId);
- }
-
- @Override
- public Map<String, Object> getMetricsData() {
- Map<String, Object> ret = new HashMap<>();
- ret.put(partitionId + "/receiveApiLatencyMean", receiveApiLatencyMean.getValueAndReset());
- ret.put(partitionId + "/receiveApiCallCount", receiveApiCallCount.getValueAndReset());
- ret.put(partitionId + "/receiveMessageCount", receiveMessageCount.getValueAndReset());
- return ret;
- }
-}
diff --git a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/EventHubSpout.java b/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/EventHubSpout.java
deleted file mode 100755
index ea12d0952..000000000
--- a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/EventHubSpout.java
+++ /dev/null
@@ -1,258 +0,0 @@
-/*******************************************************************************
- * 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
- * with the License. You may obtain a copy of the License at
- *
- * 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.storm.eventhubs.spout;
-
-import com.google.common.base.Strings;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.UUID;
-import org.apache.storm.Config;
-import org.apache.storm.metric.api.IMetric;
-import org.apache.storm.spout.SpoutOutputCollector;
-import org.apache.storm.task.TopologyContext;
-import org.apache.storm.topology.OutputFieldsDeclarer;
-import org.apache.storm.topology.base.BaseRichSpout;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class EventHubSpout extends BaseRichSpout {
-
- private static final Logger logger = LoggerFactory.getLogger(EventHubSpout.class);
-
- private final UUID instanceId;
- private final EventHubSpoutConfig eventHubConfig;
- private final IEventDataScheme scheme;
- private final int checkpointIntervalInSeconds;
-
- private IStateStore stateStore;
- private IPartitionCoordinator partitionCoordinator;
- private IPartitionManagerFactory pmFactory;
- private IEventHubReceiverFactory recvFactory;
- private SpoutOutputCollector collector;
- private long lastCheckpointTime;
- private int currentPartitionIndex = -1;
-
- public EventHubSpout(String username, String password, String namespace,
- String entityPath, int partitionCount) {
- this(new EventHubSpoutConfig(username, password, namespace, entityPath, partitionCount));
- }
-
- public EventHubSpout(EventHubSpoutConfig spoutConfig) {
- this(spoutConfig, null, null, null);
- }
-
- public EventHubSpout(EventHubSpoutConfig spoutConfig,
- IStateStore store,
- IPartitionManagerFactory pmFactory,
- IEventHubReceiverFactory recvFactory) {
- this.eventHubConfig = spoutConfig;
- this.scheme = spoutConfig.getEventDataScheme();
- this.instanceId = UUID.randomUUID();
- this.checkpointIntervalInSeconds = spoutConfig.getCheckpointIntervalInSeconds();
- this.lastCheckpointTime = System.currentTimeMillis();
- stateStore = store;
- this.pmFactory = pmFactory;
- if (this.pmFactory == null) {
- this.pmFactory = new IPartitionManagerFactory() {
- @Override
- public IPartitionManager create(EventHubSpoutConfig spoutConfig,
- String partitionId, IStateStore stateStore,
- IEventHubReceiver receiver) {
- return new PartitionManager(spoutConfig, partitionId,
- stateStore, receiver);
- }
- };
- }
- this.recvFactory = recvFactory;
- if (this.recvFactory == null) {
- this.recvFactory = new IEventHubReceiverFactory() {
- @Override
- public IEventHubReceiver create(EventHubSpoutConfig spoutConfig,
- String partitionId) {
- return new EventHubReceiverImpl(spoutConfig, partitionId);
- }
- };
- }
-
- }
-
- /**
- * This is a extracted method that is easy to test.
- */
- public void preparePartitions(Map<String, Object> config, int totalTasks, int taskIndex, SpoutOutputCollector collector) throws
- Exception {
- this.collector = collector;
- if (stateStore == null) {
- String zkEndpointAddress = eventHubConfig.getZkConnectionString();
- if (zkEndpointAddress == null || zkEndpointAddress.length() == 0) {
- //use storm's zookeeper servers if not specified.
- @SuppressWarnings("unchecked")
- List<String> zkServers = (List<String>) config.get(Config.STORM_ZOOKEEPER_SERVERS);
- Integer zkPort = ((Number) config.get(Config.STORM_ZOOKEEPER_PORT)).intValue();
- StringBuilder sb = new StringBuilder();
- for (String zk : zkServers) {
- if (sb.length() > 0) {
- sb.append(',');
- }
- sb.append(zk + ":" + zkPort);
- }
- zkEndpointAddress = sb.toString();
- }
- stateStore = new ZookeeperStateStore(zkEndpointAddress,
- Integer.parseInt(config.get(Config.STORM_ZOOKEEPER_RETRY_TIMES).toString()),
- Integer.parseInt(config.get(Config.STORM_ZOOKEEPER_RETRY_INTERVAL).toString()));
- }
- stateStore.open();
-
- partitionCoordinator = new StaticPartitionCoordinator(
- eventHubConfig, taskIndex, totalTasks, stateStore, pmFactory, recvFactory);
-
- for (IPartitionManager partitionManager :
- partitionCoordinator.getMyPartitionManagers()) {
- partitionManager.open();
- }
- }
-
- @Override
- public void open(Map<String, Object> config, TopologyContext context, SpoutOutputCollector collector) {
- logger.info("begin:start open()");
- String topologyName = (String) config.get(Config.TOPOLOGY_NAME);
- eventHubConfig.setTopologyName(topologyName);
-
- int totalTasks = context.getComponentTasks(context.getThisComponentId()).size();
- int taskIndex = context.getThisTaskIndex();
- if (totalTasks > eventHubConfig.getPartitionCount()) {
- throw new RuntimeException("total tasks of EventHubSpout is greater than partition count.");
- }
-
- logger.info(String.format("topologyName: %s, totalTasks: %d, taskIndex: %d", topologyName, totalTasks, taskIndex));
-
- try {
- preparePartitions(config, totalTasks, taskIndex, collector);
- } catch (Exception e) {
- collector.reportError(e);
- throw new RuntimeException(e);
- }
-
- //register metrics
- context.registerMetric("EventHubReceiver", new IMetric() {
- @Override
- public Object getValueAndReset() {
- Map<String, Object> concatMetricsDataMaps = new HashMap<>();
- for (IPartitionManager partitionManager :
- partitionCoordinator.getMyPartitionManagers()) {
- concatMetricsDataMaps.putAll(partitionManager.getMetricsData());
- }
- return concatMetricsDataMaps;
- }
- }, Integer.parseInt(config.get(Config.TOPOLOGY_BUILTIN_METRICS_BUCKET_SIZE_SECS).toString()));
- logger.info("end open()");
- }
-
- @Override
- public void nextTuple() {
- EventDataWrap eventDatawrap = null;
-
- List<IPartitionManager> partitionManagers = partitionCoordinator.getMyPartitionManagers();
- for (int i = 0; i < partitionManagers.size(); i++) {
- currentPartitionIndex = (currentPartitionIndex + 1) % partitionManagers.size();
- IPartitionManager partitionManager = partitionManagers.get(currentPartitionIndex);
-
- if (partitionManager == null) {
- throw new RuntimeException("partitionManager doesn't exist.");
- }
-
- eventDatawrap = partitionManager.receive();
-
- if (eventDatawrap != null) {
- break;
- }
- }
-
- if (eventDatawrap != null) {
- MessageId messageId = eventDatawrap.getMessageId();
- List<Object> tuples = scheme.deserialize(eventDatawrap.getEventData());
- if (tuples != null) {
- collector.emit(tuples, messageId);
- }
- }
-
- checkpointIfNeeded();
-
- // We don't need to sleep here because the IPartitionManager.receive() is
- // a blocked call so it's fine to call this function in a tight loop.
- }
-
- @Override
- public void ack(Object msgId) {
- MessageId messageId = (MessageId) msgId;
- IPartitionManager partitionManager = partitionCoordinator.getPartitionManager(messageId.getPartitionId());
- String offset = messageId.getOffset();
- partitionManager.ack(offset);
- }
-
- @Override
- public void fail(Object msgId) {
- MessageId messageId = (MessageId) msgId;
- IPartitionManager partitionManager = partitionCoordinator.getPartitionManager(messageId.getPartitionId());
- String offset = messageId.getOffset();
- partitionManager.fail(offset);
- }
-
- @Override
- public void deactivate() {
- // let's checkpoint so that we can get the last checkpoint when restarting.
- checkpoint();
- }
-
- @Override
- public void close() {
- for (IPartitionManager partitionManager :
- partitionCoordinator.getMyPartitionManagers()) {
- partitionManager.close();
- }
- stateStore.close();
- }
-
- @Override
- public void declareOutputFields(OutputFieldsDeclarer declarer) {
- if (Strings.isNullOrEmpty(eventHubConfig.getOutputStreamId())) {
- declarer.declare(scheme.getOutputFields());
- } else {
- declarer.declareStream(eventHubConfig.getOutputStreamId(), scheme.getOutputFields());
- }
- }
-
- private void checkpointIfNeeded() {
- long nextCheckpointTime = lastCheckpointTime + checkpointIntervalInSeconds * 1000;
- if (nextCheckpointTime < System.currentTimeMillis()) {
-
- checkpoint();
- lastCheckpointTime = System.currentTimeMillis();
- }
- }
-
- private void checkpoint() {
- for (IPartitionManager partitionManager :
- partitionCoordinator.getMyPartitionManagers()) {
- partitionManager.checkpoint();
- }
- }
-}
diff --git a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/EventHubSpoutConfig.java b/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/EventHubSpoutConfig.java
deleted file mode 100755
index 3e486b500..000000000
--- a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/EventHubSpoutConfig.java
+++ /dev/null
@@ -1,264 +0,0 @@
-/*******************************************************************************
- * 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
- * with the License. You may obtain a copy of the License at
- *
- * 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.storm.eventhubs.spout;
-
-import com.microsoft.azure.eventhubs.EventHubClient;
-import com.microsoft.azure.servicebus.ConnectionStringBuilder;
-import java.io.Serializable;
-import java.util.ArrayList;
-import java.util.List;
-
-public class EventHubSpoutConfig implements Serializable {
- public static final String EH_SERVICE_FQDN_SUFFIX = "servicebus.windows.net";
- private static final long serialVersionUID = 1L;
- private final String userName;
- private final String password;
- private final String namespace;
- private final String entityPath;
- private final int partitionCount;
-
- private String zkConnectionString = null; // if null then use zookeeper used
- // by Storm
- private int checkpointIntervalInSeconds = 10;
- private int receiverCredits = 1024;
- private int maxPendingMsgsPerPartition = 1024;
- private long enqueueTimeFilter = 0; // timestamp in millisecond, 0 means
- // disabling filter
- private String connectionString;
- private String topologyName;
- private int receiverTimeoutInMillis = 1000; // default
- private IEventDataScheme scheme = new StringEventDataScheme();
- private String consumerGroupName = EventHubClient.DEFAULT_CONSUMER_GROUP_NAME;
- private String outputStreamId;
-
-
- // These are mandatory parameters
- public EventHubSpoutConfig(String username, String password,
- String namespace, String entityPath, int partitionCount) {
- this.userName = username;
- this.password = password;
- this.connectionString = new ConnectionStringBuilder(namespace, entityPath,
- username, password).toString();
- this.namespace = namespace;
- this.entityPath = entityPath;
- this.partitionCount = partitionCount;
- }
-
- // Keep this constructor for backward compatibility
- public EventHubSpoutConfig(String username, String password,
- String namespace, String entityPath, int partitionCount,
- String zkConnectionString) {
- this(username, password, namespace, entityPath, partitionCount);
- setZkConnectionString(zkConnectionString);
- }
-
- // Keep this constructor for backward compatibility
- public EventHubSpoutConfig(String username, String password,
- String namespace, String entityPath, int partitionCount,
- String zkConnectionString, int checkpointIntervalInSeconds,
- int receiverCredits) {
- this(username, password, namespace, entityPath, partitionCount,
- zkConnectionString);
- setCheckpointIntervalInSeconds(checkpointIntervalInSeconds);
- setReceiverCredits(receiverCredits);
- }
-
- public EventHubSpoutConfig(String username, String password,
- String namespace, String entityPath, int partitionCount,
- String zkConnectionString, int checkpointIntervalInSeconds,
- int receiverCredits, long enqueueTimeFilter) {
- this(username, password, namespace, entityPath, partitionCount,
- zkConnectionString, checkpointIntervalInSeconds,
- receiverCredits);
- setEnqueueTimeFilter(enqueueTimeFilter);
- }
-
- // Keep this constructor for backward compatibility
- public EventHubSpoutConfig(String username, String password,
- String namespace, String entityPath, int partitionCount,
- String zkConnectionString, int checkpointIntervalInSeconds,
- int receiverCredits, int maxPendingMsgsPerPartition,
- long enqueueTimeFilter) {
-
- this(username, password, namespace, entityPath, partitionCount,
- zkConnectionString, checkpointIntervalInSeconds,
- receiverCredits);
- setMaxPendingMsgsPerPartition(maxPendingMsgsPerPartition);
- setEnqueueTimeFilter(enqueueTimeFilter);
- }
-
- public String getNamespace() {
- return namespace;
- }
-
- public String getEntityPath() {
- return entityPath;
- }
-
- public int getPartitionCount() {
- return partitionCount;
- }
-
- public String getZkConnectionString() {
- return zkConnectionString;
- }
-
- public void setZkConnectionString(String value) {
- zkConnectionString = value;
- }
-
- public EventHubSpoutConfig withZkConnectionString(String value) {
- setZkConnectionString(value);
- return this;
- }
-
- public int getCheckpointIntervalInSeconds() {
- return checkpointIntervalInSeconds;
- }
-
- public void setCheckpointIntervalInSeconds(int value) {
- checkpointIntervalInSeconds = value;
- }
-
- public EventHubSpoutConfig withCheckpointIntervalInSeconds(int value) {
- setCheckpointIntervalInSeconds(value);
- return this;
- }
-
- public int getReceiverCredits() {
- return receiverCredits;
- }
-
- public void setReceiverCredits(int value) {
- receiverCredits = value;
- }
-
- public EventHubSpoutConfig withReceiverCredits(int value) {
- setReceiverCredits(value);
- return this;
- }
-
- public int getMaxPendingMsgsPerPartition() {
- return maxPendingMsgsPerPartition;
- }
-
- public void setMaxPendingMsgsPerPartition(int value) {
- maxPendingMsgsPerPartition = value;
- }
-
- public EventHubSpoutConfig withMaxPendingMsgsPerPartition(int value) {
- setMaxPendingMsgsPerPartition(value);
- return this;
- }
-
- public long getEnqueueTimeFilter() {
- return enqueueTimeFilter;
- }
-
- public void setEnqueueTimeFilter(long value) {
- enqueueTimeFilter = value;
- }
-
- public EventHubSpoutConfig withEnqueueTimeFilter(long value) {
- setEnqueueTimeFilter(value);
- return this;
- }
-
- public String getTopologyName() {
- return topologyName;
- }
-
- public void setTopologyName(String value) {
- topologyName = value;
- }
-
- public EventHubSpoutConfig withTopologyName(String value) {
- setTopologyName(value);
- return this;
- }
-
- public IEventDataScheme getEventDataScheme() {
- return scheme;
- }
-
- public void setEventDataScheme(IEventDataScheme scheme) {
- this.scheme = scheme;
- }
-
- public EventHubSpoutConfig withEventDataScheme(IEventDataScheme value) {
- setEventDataScheme(value);
- return this;
- }
-
- public String getConsumerGroupName() {
- return consumerGroupName;
- }
-
- public void setConsumerGroupName(String value) {
- consumerGroupName = value;
- }
-
- public EventHubSpoutConfig withConsumerGroupName(String value) {
- setConsumerGroupName(value);
- return this;
- }
-
- public List<String> getPartitionList() {
- List<String> partitionList = new ArrayList<String>();
-
- for (int i = 0; i < this.partitionCount; i++) {
- partitionList.add(Integer.toString(i));
- }
-
- return partitionList;
- }
-
- public String getConnectionString() {
- return connectionString;
- }
-
- /*Keeping it for backward compatibility*/
- public void setTargetAddress(String targetFqnAddress) {
- }
-
- public void setTargetAddress() {
-
- }
-
- public EventHubSpoutConfig withTargetAddress(String targetFqnAddress) {
- setTargetAddress(targetFqnAddress);
- return this;
- }
-
- public String getOutputStreamId() {
- return outputStreamId;
- }
-
- public void setOutputStreamId(String outputStreamId) {
- this.outputStreamId = outputStreamId;
- }
-
- public int getReceiverTimeoutInMillis() {
- return receiverTimeoutInMillis;
- }
-
- public void setReceiverTimeoutInMillis(int receiverTimeoutInMillis) {
- this.receiverTimeoutInMillis = receiverTimeoutInMillis;
- }
-}
diff --git a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/FieldConstants.java b/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/FieldConstants.java
deleted file mode 100755
index 1baf16fc3..000000000
--- a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/FieldConstants.java
+++ /dev/null
@@ -1,29 +0,0 @@
-/*******************************************************************************
- * 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
- * with the License. You may obtain a copy of the License at
- *
- * 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.storm.eventhubs.spout;
-
-public class FieldConstants {
-
- public static final String PartitionKey = "partitionKey";
- public static final String Offset = "offset";
- public static final String Message = "message";
- public static final String META_DATA = "metadata";
- public static final String SYSTEM_META_DATA = "eventdata_system_properties";
- public static final String DefaultStartingOffset = "-1";
-}
diff --git a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/IEventDataScheme.java b/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/IEventDataScheme.java
deleted file mode 100755
index f2aa1582d..000000000
--- a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/IEventDataScheme.java
+++ /dev/null
@@ -1,44 +0,0 @@
-/*******************************************************************************
- * 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
- * with the License. You may obtain a copy of the License at
- *
- * 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.storm.eventhubs.spout;
-
-import com.microsoft.azure.eventhubs.EventData;
-import java.io.Serializable;
-import java.util.List;
-import org.apache.storm.tuple.Fields;
-
-public interface IEventDataScheme extends Serializable {
-
- /**
- * Deserialize an AMQP Message into a Tuple.
- *
- * @param eventData The EventData to Deserialize.
- * @return A tuple containing the deserialized fields of the message.
- *
- * @see #getOutputFields() for the list of fields the tuple will contain.
- */
- List<Object> deserialize(EventData eventData);
-
- /**
- * Retrieve the Fields that are present on tuples created by this object.
- *
- * @return The Fields that are present on tuples created by this object.
- */
- Fields getOutputFields();
-}
diff --git a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/IEventFilter.java b/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/IEventFilter.java
deleted file mode 100644
index c67823d59..000000000
--- a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/IEventFilter.java
+++ /dev/null
@@ -1,28 +0,0 @@
-/*******************************************************************************
- * 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
- * with the License. You may obtain a copy of the License at
- *
- * 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.storm.eventhubs.spout;
-
-import java.time.Instant;
-
-public interface IEventFilter {
-
- String getOffset();
-
- Instant getTime();
-}
diff --git a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/IEventHubReceiver.java b/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/IEventHubReceiver.java
deleted file mode 100755
index 4af967f6c..000000000
--- a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/IEventHubReceiver.java
+++ /dev/null
@@ -1,34 +0,0 @@
-/*******************************************************************************
- * 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
- * with the License. You may obtain a copy of the License at
- *
- * 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.storm.eventhubs.spout;
-
-import java.util.Map;
-
-public interface IEventHubReceiver {
-
- void open(IEventFilter filter) throws EventHubException;
-
- void close();
-
- boolean isOpen();
-
- EventDataWrap receive();
-
- Map<String, Object> getMetricsData();
-}
diff --git a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/IEventHubReceiverFactory.java b/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/IEventHubReceiverFactory.java
deleted file mode 100755
index ef4d9b822..000000000
--- a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/IEventHubReceiverFactory.java
+++ /dev/null
@@ -1,25 +0,0 @@
-/**
- * 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 with the License. You may obtain a copy of the License at
- *
- * 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.storm.eventhubs.spout;
-
-import java.io.Serializable;
-
-/**
- * An abstract factory to generate EventHubReceiver.
- */
-public interface IEventHubReceiverFactory extends Serializable {
- IEventHubReceiver create(EventHubSpoutConfig config, String partitionId);
-}
diff --git a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/IPartitionCoordinator.java b/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/IPartitionCoordinator.java
deleted file mode 100755
index d83c0cbb9..000000000
--- a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/IPartitionCoordinator.java
+++ /dev/null
@@ -1,28 +0,0 @@
-/*******************************************************************************
- * 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
- * with the License. You may obtain a copy of the License at
- *
- * 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.storm.eventhubs.spout;
-
-import java.util.List;
-
-public interface IPartitionCoordinator {
-
- List<IPartitionManager> getMyPartitionManagers();
-
- IPartitionManager getPartitionManager(String partitionId);
-}
diff --git a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/IPartitionManager.java b/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/IPartitionManager.java
deleted file mode 100755
index 345fd483d..000000000
--- a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/IPartitionManager.java
+++ /dev/null
@@ -1,38 +0,0 @@
-/*******************************************************************************
- * 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
- * with the License. You may obtain a copy of the License at
- *
- * 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.storm.eventhubs.spout;
-
-import java.util.Map;
-
-public interface IPartitionManager {
-
- void open() throws Exception;
-
- void close();
-
- EventDataWrap receive();
-
- void checkpoint();
-
- void ack(String offset);
-
- void fail(String offset);
-
- Map<String, Object> getMetricsData();
-}
diff --git a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/IPartitionManagerFactory.java b/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/IPartitionManagerFactory.java
deleted file mode 100755
index 4a87b8676..000000000
--- a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/IPartitionManagerFactory.java
+++ /dev/null
@@ -1,28 +0,0 @@
-/**
- * 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 with the License. You may obtain a copy of the License at
- *
- * 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.storm.eventhubs.spout;
-
-import java.io.Serializable;
-
-/**
- * An interface of factory method to create IPartitionManager.
- */
-public interface IPartitionManagerFactory extends Serializable {
- IPartitionManager create(EventHubSpoutConfig spoutConfig,
- String partitionId,
- IStateStore stateStore,
- IEventHubReceiver receiver);
-}
diff --git a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/IStateStore.java b/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/IStateStore.java
deleted file mode 100755
index a2c4d4fe0..000000000
--- a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/IStateStore.java
+++ /dev/null
@@ -1,32 +0,0 @@
-/*******************************************************************************
- * 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
- * with the License. You may obtain a copy of the License at
- *
- * 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.storm.eventhubs.spout;
-
-import java.io.Serializable;
-
-public interface IStateStore extends Serializable {
-
- void open();
-
- void close();
-
- void saveData(String path, String data);
-
- String readData(String path);
-}
diff --git a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/MessageId.java b/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/MessageId.java
deleted file mode 100755
index 0bb1f100a..000000000
--- a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/MessageId.java
+++ /dev/null
@@ -1,57 +0,0 @@
-/*******************************************************************************
- * 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
- * with the License. You may obtain a copy of the License at
- *
- * 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.storm.eventhubs.spout;
-
-public class MessageId {
-
- private final String partitionId;
- private final String offset;
- private final long sequenceNumber;
-
- public MessageId(
- String partitionId,
- String offset,
- long sequenceNumber) {
- this.partitionId = partitionId;
- this.offset = offset;
- this.sequenceNumber = sequenceNumber;
- }
-
- public static MessageId create(String partitionId, String offset, long sequenceNumber) {
- return new MessageId(partitionId, offset, sequenceNumber);
- }
-
- public String getPartitionId() {
- return this.partitionId;
- }
-
- public String getOffset() {
- return this.offset;
- }
-
- public Long getSequenceNumber() {
- return this.sequenceNumber;
- }
-
- @Override
- public String toString() {
- return String.format("PartitionId: %s, Offset: %s, SequenceNumber: %s",
- this.partitionId, this.offset, this.sequenceNumber);
- }
-}
diff --git a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/PartitionManager.java b/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/PartitionManager.java
deleted file mode 100755
index c5c5084d8..000000000
--- a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/PartitionManager.java
+++ /dev/null
@@ -1,101 +0,0 @@
-/*******************************************************************************
- * 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
- * with the License. You may obtain a copy of the License at
- *
- * 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.storm.eventhubs.spout;
-
-import java.util.LinkedHashMap;
-import java.util.Map;
-import java.util.TreeSet;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class PartitionManager extends SimplePartitionManager {
- private static final Logger logger = LoggerFactory.getLogger(PartitionManager.class);
- private final int ehReceiveTimeoutMs = 5000;
-
- //all sent events are stored in pending
- private final Map<String, EventDataWrap> pending;
- //all failed events are put in toResend, which is sorted by event's offset
- private final TreeSet<EventDataWrap> toResend;
-
- public PartitionManager(
- EventHubSpoutConfig spoutConfig,
- String partitionId,
- IStateStore stateStore,
- IEventHubReceiver receiver) {
-
- super(spoutConfig, partitionId, stateStore, receiver);
-
- this.pending = new LinkedHashMap<String, EventDataWrap>();
- this.toResend = new TreeSet<EventDataWrap>();
- }
-
- @Override
- public EventDataWrap receive() {
- if (pending.size() >= config.getMaxPendingMsgsPerPartition()) {
- return null;
- }
-
- EventDataWrap eventDatawrap;
- if (toResend.isEmpty()) {
- eventDatawrap = receiver.receive();
- } else {
- eventDatawrap = toResend.pollFirst();
- }
-
- if (eventDatawrap != null) {
- lastOffset = eventDatawrap.getMessageId().getOffset();
- pending.put(lastOffset, eventDatawrap);
- }
-
- return eventDatawrap;
- }
-
- @Override
- public void ack(String offset) {
- pending.remove(offset);
- }
-
- @Override
- public void fail(String offset) {
- logger.warn("fail on " + offset);
- EventDataWrap eventDataWrap = pending.remove(offset);
- toResend.add(eventDataWrap);
- }
-
- @Override
- protected String getCompletedOffset() {
- String offset = null;
-
- if (pending.size() > 0) {
- //find the smallest offset in pending list
- offset = pending.keySet().iterator().next();
- }
- if (toResend.size() > 0) {
- //find the smallest offset in toResend list
- String offset2 = toResend.first().getMessageId().getOffset();
- if (offset == null || offset2.compareTo(offset) < 0) {
- offset = offset2;
- }
- }
- if (offset == null) {
- offset = lastOffset;
- }
- return offset;
- }
-}
diff --git a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/SerializeDeserializeUtil.java b/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/SerializeDeserializeUtil.java
deleted file mode 100644
index 0439a4fe4..000000000
--- a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/SerializeDeserializeUtil.java
+++ /dev/null
@@ -1,34 +0,0 @@
-/*******************************************************************************
- * 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
- * with the License. You may obtain a copy of the License at
- *
- * 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.storm.eventhubs.spout;
-
-import java.io.ByteArrayOutputStream;
-import java.io.IOException;
-import java.io.ObjectOutputStream;
-
-public class SerializeDeserializeUtil {
- public static byte[] serialize(Object obj) throws IOException {
- try (ByteArrayOutputStream b = new ByteArrayOutputStream()) {
- try (ObjectOutputStream o = new ObjectOutputStream(b)) {
- o.writeObject(obj);
- }
- return b.toByteArray();
- }
- }
-}
diff --git a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/SimplePartitionManager.java b/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/SimplePartitionManager.java
deleted file mode 100755
index 3c98735da..000000000
--- a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/SimplePartitionManager.java
+++ /dev/null
@@ -1,133 +0,0 @@
-/*******************************************************************************
- * 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
- * with the License. You may obtain a copy of the License at
- *
- * 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.storm.eventhubs.spout;
-
-import java.time.Instant;
-import java.util.Map;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * A simple partition manager that does not re-send failed messages.
- */
-public class SimplePartitionManager implements IPartitionManager {
- private static final Logger logger = LoggerFactory.getLogger(SimplePartitionManager.class);
- private static final String statePathPrefix = "/eventhubspout";
-
- protected final IEventHubReceiver receiver;
- protected final EventHubSpoutConfig config;
- private final String partitionId;
- private final IStateStore stateStore;
- private final String statePath;
- protected String lastOffset = "-1";
- protected String committedOffset = "-1";
-
- public SimplePartitionManager(
- EventHubSpoutConfig spoutConfig,
- String partitionId,
- IStateStore stateStore,
- IEventHubReceiver receiver) {
- this.receiver = receiver;
- this.config = spoutConfig;
- this.partitionId = partitionId;
- this.statePath = this.getPartitionStatePath();
- this.stateStore = stateStore;
- }
-
- @Override
- public void open() throws Exception {
-
- //read from state store, if not found, use startingOffset
- String offset = stateStore.readData(statePath);
- logger.info("read offset from state store: " + offset);
- if (offset == null) {
- offset = FieldConstants.DefaultStartingOffset;
- }
- IEventFilter filter;
- if (offset.equals(FieldConstants.DefaultStartingOffset)
- && config.getEnqueueTimeFilter() != 0) {
- filter = new EventHubFilter(Instant.ofEpochMilli(config.getEnqueueTimeFilter()));
- } else {
- filter = new EventHubFilter(offset);
- }
-
- receiver.open(filter);
- }
-
- @Override
- public void close() {
- this.receiver.close();
- this.checkpoint();
- }
-
- @Override
- public void checkpoint() {
- String completedOffset = getCompletedOffset();
- if (!committedOffset.equals(completedOffset)) {
- logger.info("saving state " + completedOffset);
- stateStore.saveData(statePath, completedOffset);
- committedOffset = completedOffset;
- }
- }
-
- protected String getCompletedOffset() {
- return lastOffset;
- }
-
- @Override
- public EventDataWrap receive() {
- EventDataWrap eventDatawrap = receiver.receive();
- if (eventDatawrap != null) {
- lastOffset = eventDatawrap.getEventData().getSystemProperties().getOffset();
- }
- return eventDatawrap;
- }
-
- @Override
- public void ack(String offset) {
- //do nothing
- }
-
- @Override
- public void fail(String offset) {
- logger.warn("fail on " + offset);
- //do nothing
- }
-
- private String getPartitionStatePath() {
-
- // Partition state path =
- // "/{prefix}/{topologyName}/{namespace}/{entityPath}/partitions/{partitionId}/state";
- String namespace = config.getNamespace();
- String entityPath = config.getEntityPath();
- String topologyName = config.getTopologyName();
-
- String partitionStatePath =
- statePathPrefix + "/" + topologyName + "/" + namespace + "/" + entityPath + "/partitions/" + this.partitionId;
-
- logger.info("partition state path: " + partitionStatePath);
-
- return partitionStatePath;
- }
-
- @Override
- public Map<String, Object> getMetricsData() {
- return receiver.getMetricsData();
- }
-}
diff --git a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/StaticPartitionCoordinator.java b/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/StaticPartitionCoordinator.java
deleted file mode 100755
index 54c0f7d61..000000000
--- a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/StaticPartitionCoordinator.java
+++ /dev/null
@@ -1,84 +0,0 @@
-/*******************************************************************************
- * 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
- * with the License. You may obtain a copy of the License at
- *
- * 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.storm.eventhubs.spout;
-
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-
-public class StaticPartitionCoordinator implements IPartitionCoordinator {
-
- private static final Logger logger = LoggerFactory.getLogger(StaticPartitionCoordinator.class);
-
- protected final EventHubSpoutConfig config;
- protected final int taskIndex;
- protected final int totalTasks;
- protected final List<IPartitionManager> partitionManagers;
- protected final Map<String, IPartitionManager> partitionManagerMap;
- protected final IStateStore stateStore;
-
- public StaticPartitionCoordinator(
- EventHubSpoutConfig spoutConfig,
- int taskIndex,
- int totalTasks,
- IStateStore stateStore,
- IPartitionManagerFactory pmFactory,
- IEventHubReceiverFactory recvFactory) {
-
- this.config = spoutConfig;
- this.taskIndex = taskIndex;
- this.totalTasks = totalTasks;
- this.stateStore = stateStore;
- List<String> partitionIds = calculateParititionIdsToOwn();
- partitionManagerMap = new HashMap<String, IPartitionManager>();
- partitionManagers = new ArrayList<IPartitionManager>();
-
- for (String partitionId : partitionIds) {
- IEventHubReceiver receiver = recvFactory.create(config, partitionId);
- IPartitionManager partitionManager = pmFactory.create(
- config, partitionId, stateStore, receiver);
- partitionManagerMap.put(partitionId, partitionManager);
- partitionManagers.add(partitionManager);
- }
- }
-
- @Override
- public List<IPartitionManager> getMyPartitionManagers() {
- return partitionManagers;
- }
-
- @Override
- public IPartitionManager getPartitionManager(String partitionId) {
- return partitionManagerMap.get(partitionId);
- }
-
- protected List<String> calculateParititionIdsToOwn() {
- List<String> taskPartitions = new ArrayList<String>();
- for (int i = this.taskIndex; i < config.getPartitionCount(); i += this.totalTasks) {
- taskPartitions.add(Integer.toString(i));
- logger.info(String.format("taskIndex %d owns partitionId %d.", this.taskIndex, i));
- }
-
- return taskPartitions;
- }
-}
diff --git a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/StringEventDataScheme.java b/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/StringEventDataScheme.java
deleted file mode 100644
index 3bd1ba8bf..000000000
--- a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/StringEventDataScheme.java
+++ /dev/null
@@ -1,71 +0,0 @@
-/*******************************************************************************
- * 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
- * with the License. You may obtain a copy of the License at
- *
- * 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.storm.eventhubs.spout;
-
-import com.microsoft.azure.eventhubs.EventData;
-import java.util.ArrayList;
-import java.util.List;
-import org.apache.storm.tuple.Fields;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * An Event Data Scheme which deserializes message payload into the Strings. No encoding is assumed. The receiver will need to handle
- * parsing of the string data in appropriate encoding.
- *
- * <p>Note: Unlike other schemes provided, this scheme does not include any metadata.
- *
- * <p>For metadata please refer to {@link BinaryEventDataScheme}, {@link EventDataScheme}
- */
-public class StringEventDataScheme implements IEventDataScheme {
-
- private static final long serialVersionUID = 1L;
- private static final Logger logger = LoggerFactory.getLogger(StringEventDataScheme.class);
-
- @Override
- public List<Object> deserialize(EventData eventData) {
- final List<Object> fieldContents = new ArrayList<Object>();
- String messageData = "";
- if (eventData.getBytes() != null) {
- messageData = new String(eventData.getBytes());
- } else if (eventData.getObject() != null) {
- /*Will only serialize AMQPValue type*/
- try {
- if (!(eventData.getObject() instanceof List)) {
- messageData = eventData.getObject().toString();
- } else {
- throw new RuntimeException("Cannot serialize the given AMQP type.");
- }
- } catch (RuntimeException e) {
- logger.error("Failed to serialize EventData payload class"
- + eventData.getObject().getClass());
- logger.error("Exception encountered while serializing EventData payload is"
- + e.toString());
- throw e;
- }
- }
- fieldContents.add(messageData);
- return fieldContents;
- }
-
- @Override
- public Fields getOutputFields() {
- return new Fields(FieldConstants.Message);
- }
-}
diff --git a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/ZookeeperStateStore.java b/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/ZookeeperStateStore.java
deleted file mode 100755
index 4e2ece9a4..000000000
--- a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/spout/ZookeeperStateStore.java
+++ /dev/null
@@ -1,96 +0,0 @@
-/*******************************************************************************
- * 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
- * with the License. You may obtain a copy of the License at
- *
- * 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.storm.eventhubs.spout;
-
-import org.apache.curator.RetryPolicy;
-import org.apache.curator.framework.CuratorFramework;
-import org.apache.curator.framework.CuratorFrameworkFactory;
-import org.apache.curator.retry.RetryNTimes;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class ZookeeperStateStore implements IStateStore {
- private static final long serialVersionUID = 1L;
- private static final Logger logger = LoggerFactory.getLogger(ZookeeperStateStore.class);
-
- private final String zookeeperConnectionString;
- private final CuratorFramework curatorFramework;
-
- public ZookeeperStateStore(String zookeeperConnectionString) {
- this(zookeeperConnectionString, 3, 100);
- }
-
- public ZookeeperStateStore(String connectionString, int retries, int retryInterval) {
- if (connectionString == null) {
- zookeeperConnectionString = "localhost:2181";
- } else {
- zookeeperConnectionString = connectionString;
- }
-
- RetryPolicy retryPolicy = new RetryNTimes(retries, retryInterval);
- curatorFramework = CuratorFrameworkFactory.newClient(zookeeperConnectionString, retryPolicy);
- }
-
- @Override
- public void open() {
- curatorFramework.start();
- }
-
- @Override
- public void close() {
- curatorFramework.close();
- }
-
- @Override
- public void saveData(String statePath, String data) {
- data = data == null ? "" : data;
- byte[] bytes = data.getBytes();
-
- try {
- if (curatorFramework.checkExists().forPath(statePath) == null) {
- curatorFramework.create().creatingParentsIfNeeded().forPath(statePath, bytes);
- } else {
- curatorFramework.setData().forPath(statePath, bytes);
- }
-
- logger.info(String.format("data was saved. path: %s, data: %s.", statePath, data));
- } catch (Exception e) {
- throw new RuntimeException(e);
- }
- }
-
- @Override
- public String readData(String statePath) {
- try {
- if (curatorFramework.checkExists().forPath(statePath) == null) {
- // do we want to throw an exception if path doesn't exist??
- return null;
- } else {
- byte[] bytes = curatorFramework.getData().forPath(statePath);
- String data = new String(bytes);
-
- logger.info(String.format("data was retrieved. path: %s, data: %s.", statePath, data));
-
- return data;
- }
- } catch (Exception e) {
- throw new RuntimeException(e);
- }
- }
-}
diff --git a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/trident/Coordinator.java b/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/trident/Coordinator.java
deleted file mode 100755
index b87c81ba7..000000000
--- a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/trident/Coordinator.java
+++ /dev/null
@@ -1,59 +0,0 @@
-/*******************************************************************************
- * 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
- * with the License. You may obtain a copy of the License at
- *
- * 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.storm.eventhubs.trident;
-
-import org.apache.storm.eventhubs.spout.EventHubSpoutConfig;
-import org.apache.storm.trident.spout.IOpaquePartitionedTridentSpout;
-import org.apache.storm.trident.spout.IPartitionedTridentSpout;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class Coordinator implements IPartitionedTridentSpout.Coordinator<Partitions>,
- IOpaquePartitionedTridentSpout.Coordinator<Partitions> {
- private static final Logger logger = LoggerFactory.getLogger(Coordinator.class);
- private final EventHubSpoutConfig spoutConfig;
- Partitions partitions;
-
- public Coordinator(EventHubSpoutConfig spoutConfig) {
- this.spoutConfig = spoutConfig;
- }
-
- @Override
- public void close() {
- }
-
- @Override
- public Partitions getPartitionsForBatch() {
- if (partitions != null) {
- return partitions;
- }
-
- partitions = new Partitions();
- for (int i = 0; i < spoutConfig.getPartitionCount(); ++i) {
- partitions.addPartition(new Partition(spoutConfig, Integer.toString(i)));
- }
- logger.info("created partitions, size=" + spoutConfig.getPartitionCount());
- return partitions;
- }
-
- @Override
- public boolean isReady(long txid) {
- return true;
- }
-}
diff --git a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/trident/ITridentPartitionManager.java b/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/trident/ITridentPartitionManager.java
deleted file mode 100755
index aa1e567c6..000000000
--- a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/trident/ITridentPartitionManager.java
+++ /dev/null
@@ -1,37 +0,0 @@
-/*******************************************************************************
- * 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
- * with the License. You may obtain a copy of the License at
- *
- * 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.storm.eventhubs.trident;
-
-import java.util.List;
-import org.apache.storm.eventhubs.spout.EventDataWrap;
-
-public interface ITridentPartitionManager {
- boolean open(String offset);
-
- void close();
-
- /**
- * receive a batch of messages from EvenHub up to "count" messages.
- *
- * @param offset the starting offset
- * @param count max number of messages in this batch
- * @return list of EventData, if failed to receive, return empty list
- */
- List<EventDataWrap> receiveBatch(String offset, int count);
-}
diff --git a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/trident/ITridentPartitionManagerFactory.java b/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/trident/ITridentPartitionManagerFactory.java
deleted file mode 100755
index 15a83cc31..000000000
--- a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/trident/ITridentPartitionManagerFactory.java
+++ /dev/null
@@ -1,26 +0,0 @@
-/*******************************************************************************
- * 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
- * with the License. You may obtain a copy of the License at
- *
- * 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.storm.eventhubs.trident;
-
-import java.io.Serializable;
-import org.apache.storm.eventhubs.spout.IEventHubReceiver;
-
-public interface ITridentPartitionManagerFactory extends Serializable {
- ITridentPartitionManager create(IEventHubReceiver receiver);
-}
diff --git a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/trident/OpaqueTridentEventHubEmitter.java b/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/trident/OpaqueTridentEventHubEmitter.java
deleted file mode 100755
index 2e8837351..000000000
--- a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/trident/OpaqueTridentEventHubEmitter.java
+++ /dev/null
@@ -1,69 +0,0 @@
-/*******************************************************************************
- * 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
- * with the License. You may obtain a copy of the License at
- *
- * 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.storm.eventhubs.trident;
-
-import java.util.List;
-import java.util.Map;
-import org.apache.storm.eventhubs.spout.EventHubSpoutConfig;
-import org.apache.storm.eventhubs.spout.IEventHubReceiverFactory;
-import org.apache.storm.trident.operation.TridentCollector;
-import org.apache.storm.trident.spout.IOpaquePartitionedTridentSpout;
-import org.apache.storm.trident.topology.TransactionAttempt;
-
-/**
- * A thin wrapper of TransactionalTridentEventHubEmitter for OpaqueTridentEventHubSpout.
- */
-public class OpaqueTridentEventHubEmitter implements IOpaquePartitionedTridentSpout.Emitter<Partitions, Partition, Map> {
- private final TransactionalTridentEventHubEmitter transactionalEmitter;
-
- public OpaqueTridentEventHubEmitter(EventHubSpoutConfig spoutConfig) {
- transactionalEmitter = new TransactionalTridentEventHubEmitter(spoutConfig);
- }
-
- public OpaqueTridentEventHubEmitter(EventHubSpoutConfig spoutConfig,
- int batchSize,
- ITridentPartitionManagerFactory pmFactory,
- IEventHubReceiverFactory recvFactory) {
- transactionalEmitter = new TransactionalTridentEventHubEmitter(spoutConfig,
- batchSize,
- pmFactory,
- recvFactory);
- }
-
- @Override
- public void close() {
- transactionalEmitter.close();
- }
-
- @Override
- public Map emitPartitionBatch(TransactionAttempt attempt, TridentCollector collector,
- Partition partition, Map meta) {
- return transactionalEmitter.emitPartitionBatchNew(attempt, collector, partition, meta);
- }
-
- @Override
- public List<Partition> getOrderedPartitions(Partitions partitions) {
- return transactionalEmitter.getOrderedPartitions(partitions);
- }
-
- @Override
- public void refreshPartitions(List<Partition> partitionList) {
- transactionalEmitter.refreshPartitions(partitionList);
- }
-}
diff --git a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/trident/OpaqueTridentEventHubSpout.java b/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/trident/OpaqueTridentEventHubSpout.java
deleted file mode 100755
index d8fb338ce..000000000
--- a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/trident/OpaqueTridentEventHubSpout.java
+++ /dev/null
@@ -1,63 +0,0 @@
-/*******************************************************************************
- * 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
- * with the License. You may obtain a copy of the License at
- *
- * 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.storm.eventhubs.trident;
-
-import java.util.Map;
-import org.apache.storm.eventhubs.spout.EventHubSpoutConfig;
-import org.apache.storm.eventhubs.spout.IEventDataScheme;
-import org.apache.storm.task.TopologyContext;
-import org.apache.storm.trident.spout.IOpaquePartitionedTridentSpout;
-import org.apache.storm.tuple.Fields;
-
-/**
- * Opaque Trident EventHubs Spout.
- */
-public class OpaqueTridentEventHubSpout implements IOpaquePartitionedTridentSpout<Partitions, Partition, Map> {
- private static final long serialVersionUID = 1L;
- private final IEventDataScheme scheme;
- private final EventHubSpoutConfig spoutConfig;
-
- public OpaqueTridentEventHubSpout(EventHubSpoutConfig config) {
- spoutConfig = config;
- scheme = spoutConfig.getEventDataScheme();
- }
-
- @Override
- public Map<String, Object> getComponentConfiguration() {
- return null;
- }
-
- @Override
- public IOpaquePartitionedTridentSpout.Coordinator<Partitions> getCoordinator(
- Map<String, Object> conf, TopologyContext context) {
- return new org.apache.storm.eventhubs.trident.Coordinator(spoutConfig);
- }
-
- @Override
- public IOpaquePartitionedTridentSpout.Emitter<Partitions, Partition, Map> getEmitter(
- Map<String, Object> conf, TopologyContext context) {
- return new OpaqueTridentEventHubEmitter(spoutConfig);
- }
-
- @Override
- public Fields getOutputFields() {
- return scheme.getOutputFields();
- }
-
-}
diff --git a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/trident/Partition.java b/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/trident/Partition.java
deleted file mode 100755
index c47e6160b..000000000
--- a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/trident/Partition.java
+++ /dev/null
@@ -1,40 +0,0 @@
-/*******************************************************************************
- * 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
- * with the License. You may obtain a copy of the License at
- *
- * 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.storm.eventhubs.trident;
-
-import java.io.Serializable;
-import org.apache.storm.eventhubs.spout.EventHubSpoutConfig;
-import org.apache.storm.trident.spout.ISpoutPartition;
-
-/**
- * Represents an EventHub partition.
- */
-public class Partition implements ISpoutPartition, Serializable {
- private static final long serialVersionUID = 1L;
- String partitionId;
-
- public Partition(EventHubSpoutConfig config, String partitionId) {
- this.partitionId = partitionId;
- }
-
- @Override
- public String getId() {
- return partitionId;
- }
-}
diff --git a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/trident/Partitions.java b/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/trident/Partitions.java
deleted file mode 100755
index 375f39db8..000000000
--- a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/trident/Partitions.java
+++ /dev/null
@@ -1,43 +0,0 @@
-/*******************************************************************************
- * 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
- * with the License. You may obtain a copy of the License at
- *
- * 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.storm.eventhubs.trident;
-
-import java.io.Serializable;
-import java.util.ArrayList;
-import java.util.List;
-
-/**
- * Represents all EventHub partitions a spout is receiving messages from.
- */
-public class Partitions implements Serializable {
- private static final long serialVersionUID = 1L;
- private List<Partition> partitionList;
-
- public Partitions() {
- partitionList = new ArrayList<Partition>();
- }
-
- public void addPartition(Partition partition) {
- partitionList.add(partition);
- }
-
- public List<Partition> getPartitions() {
- return partitionList;
- }
-}
diff --git a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/trident/TransactionalTridentEventHubEmitter.java b/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/trident/TransactionalTridentEventHubEmitter.java
deleted file mode 100755
index 23216177b..000000000
--- a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/trident/TransactionalTridentEventHubEmitter.java
+++ /dev/null
@@ -1,161 +0,0 @@
-/*******************************************************************************
- * 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
- * with the License. You may obtain a copy of the License at
- *
- * 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.storm.eventhubs.trident;
-
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import org.apache.storm.eventhubs.spout.EventDataWrap;
-import org.apache.storm.eventhubs.spout.EventHubReceiverImpl;
-import org.apache.storm.eventhubs.spout.EventHubSpoutConfig;
-import org.apache.storm.eventhubs.spout.FieldConstants;
-import org.apache.storm.eventhubs.spout.IEventHubReceiver;
-import org.apache.storm.eventhubs.spout.IEventHubReceiverFactory;
-import org.apache.storm.trident.operation.TridentCollector;
-import org.apache.storm.trident.spout.IPartitionedTridentSpout;
-import org.apache.storm.trident.topology.TransactionAttempt;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-
-public class TransactionalTridentEventHubEmitter
- implements IPartitionedTridentSpout.Emitter<Partitions, Partition, Map<String, Object>> {
- private static final Logger logger = LoggerFactory.getLogger(TransactionalTridentEventHubEmitter.class);
- private final int batchSize;
- private final EventHubSpoutConfig spoutConfig;
- private Map<String, ITridentPartitionManager> pmMap;
- private ITridentPartitionManagerFactory pmFactory;
- private IEventHubReceiverFactory recvFactory;
-
- public TransactionalTridentEventHubEmitter(EventHubSpoutConfig spoutConfig) {
- //use batch size that matches the default credit size
- this(spoutConfig, spoutConfig.getReceiverCredits(), null, null);
- }
-
- public TransactionalTridentEventHubEmitter(final EventHubSpoutConfig spoutConfig,
- int batchSize,
- ITridentPartitionManagerFactory pmFactory,
- IEventHubReceiverFactory recvFactory) {
- this.spoutConfig = spoutConfig;
- this.batchSize = batchSize;
- this.pmFactory = pmFactory;
- this.recvFactory = recvFactory;
- pmMap = new HashMap<String, ITridentPartitionManager>();
- if (this.pmFactory == null) {
- this.pmFactory = new ITridentPartitionManagerFactory() {
- @Override
- public ITridentPartitionManager create(IEventHubReceiver receiver) {
- return new TridentPartitionManager(spoutConfig, receiver);
- }
- };
- }
- if (this.recvFactory == null) {
- this.recvFactory = new IEventHubReceiverFactory() {
- @Override
- public IEventHubReceiver create(EventHubSpoutConfig config,
- String partitionId) {
- return new EventHubReceiverImpl(config, partitionId);
- }
- };
- }
- }
-
- @Override
- public void close() {
- for (ITridentPartitionManager pm : pmMap.values()) {
- pm.close();
- }
- }
-
- /**
- * Check if partition manager for a given partiton is created if not, create it.
- */
- private ITridentPartitionManager getOrCreatePartitionManager(Partition partition) {
- ITridentPartitionManager pm;
- if (!pmMap.containsKey(partition.getId())) {
- IEventHubReceiver receiver = recvFactory.create(spoutConfig, partition.getId());
- pm = pmFactory.create(receiver);
- pmMap.put(partition.getId(), pm);
- } else {
- pm = pmMap.get(partition.getId());
- }
- return pm;
- }
-
- @Override
- public void emitPartitionBatch(TransactionAttempt attempt,
- TridentCollector collector, Partition partition, Map<String, Object> meta) {
- String offset = (String) meta.get("offset");
- int count = Integer.parseInt((String) meta.get("count"));
- logger.info("re-emit for partition " + partition.getId() + ", offset=" + offset + ", count=" + count);
- ITridentPartitionManager pm = getOrCreatePartitionManager(partition);
- List<EventDataWrap> listEvents = pm.receiveBatch(offset, count);
- if (listEvents.size() != count) {
- logger.error("failed to refetch eventhub messages, new count=" + listEvents.size());
- return;
- }
-
- for (EventDataWrap ed : listEvents) {
- List<Object> tuples =
- spoutConfig.getEventDataScheme().deserialize(ed.getEventData());
- collector.emit(tuples);
- }
- }
-
- @Override
- public Map<String, Object> emitPartitionBatchNew(TransactionAttempt attempt,
- TridentCollector collector, Partition partition, Map<String, Object> meta) {
- ITridentPartitionManager pm = getOrCreatePartitionManager(partition);
- String offset = FieldConstants.DefaultStartingOffset;
- if (meta != null && meta.containsKey("nextOffset")) {
- offset = (String) meta.get("nextOffset");
- }
- //logger.info("emit for partition " + partition.getId() + ", offset=" + offset);
- String nextOffset = offset;
-
- List<EventDataWrap> listEvents = pm.receiveBatch(offset, batchSize);
-
- for (EventDataWrap ed : listEvents) {
- //update nextOffset;
- nextOffset = ed.getMessageId().getOffset();
- List<Object> tuples =
- spoutConfig.getEventDataScheme().deserialize(ed.getEventData());
- collector.emit(tuples);
- }
- //logger.info("emitted new batches: " + listEvents.size());
-
- Map<String, Object> newMeta = new HashMap<>();
- newMeta.put("offset", offset);
- newMeta.put("nextOffset", nextOffset);
- newMeta.put("count", "" + listEvents.size());
- return newMeta;
- }
-
- @Override
- public List<Partition> getOrderedPartitions(Partitions partitions) {
- return partitions.getPartitions();
- }
-
- @Override
- public void refreshPartitions(List<Partition> partitionList) {
- //partition info does not change in EventHub
- return;
- }
-
-}
diff --git a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/trident/TransactionalTridentEventHubSpout.java b/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/trident/TransactionalTridentEventHubSpout.java
deleted file mode 100755
index e9a22e7e4..000000000
--- a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/trident/TransactionalTridentEventHubSpout.java
+++ /dev/null
@@ -1,64 +0,0 @@
-/*******************************************************************************
- * 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
- * with the License. You may obtain a copy of the License at
- *
- * 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.storm.eventhubs.trident;
-
-import java.util.Map;
-import org.apache.storm.eventhubs.spout.EventHubSpoutConfig;
-import org.apache.storm.eventhubs.spout.IEventDataScheme;
-import org.apache.storm.task.TopologyContext;
-import org.apache.storm.trident.spout.IPartitionedTridentSpout;
-import org.apache.storm.tuple.Fields;
-
-/**
- * Transactional Trident EventHub Spout.
- */
-public class TransactionalTridentEventHubSpout implements
- IPartitionedTridentSpout<Partitions, Partition, Map<String, Object>> {
- private static final long serialVersionUID = 1L;
- private final IEventDataScheme scheme;
- private final EventHubSpoutConfig spoutConfig;
-
- public TransactionalTridentEventHubSpout(EventHubSpoutConfig config) {
- spoutConfig = config;
- scheme = spoutConfig.getEventDataScheme();
- }
-
- @Override
- public Map<String, Object> getComponentConfiguration() {
- return null;
- }
-
- @Override
- public IPartitionedTridentSpout.Coordinator<Partitions> getCoordinator(
- Map<String, Object> conf, TopologyContext context) {
- return new org.apache.storm.eventhubs.trident.Coordinator(spoutConfig);
- }
-
- @Override
- public IPartitionedTridentSpout.Emitter<Partitions, Partition, Map<String, Object>> getEmitter(
- Map<String, Object> conf, TopologyContext context) {
- return new TransactionalTridentEventHubEmitter(spoutConfig);
- }
-
- @Override
- public Fields getOutputFields() {
- return scheme.getOutputFields();
- }
-
-}
diff --git a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/trident/TridentPartitionManager.java b/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/trident/TridentPartitionManager.java
deleted file mode 100755
index 550e1064d..000000000
--- a/external/storm-eventhubs/src/main/java/org/apache/storm/eventhubs/trident/TridentPartitionManager.java
+++ /dev/null
@@ -1,89 +0,0 @@
-/*******************************************************************************
- * 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
- * with the License. You may obtain a copy of the License at
- *
- * 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.storm.eventhubs.trident;
-
-import java.time.Instant;
-import java.util.ArrayList;
-import java.util.List;
-import org.apache.storm.eventhubs.spout.EventDataWrap;
-import org.apache.storm.eventhubs.spout.EventHubException;
-import org.apache.storm.eventhubs.spout.EventHubFilter;
-import org.apache.storm.eventhubs.spout.EventHubSpoutConfig;
-import org.apache.storm.eventhubs.spout.FieldConstants;
-import org.apache.storm.eventhubs.spout.IEventHubReceiver;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class TridentPartitionManager implements ITridentPartitionManager {
- private static final Logger logger = LoggerFactory.getLogger(TridentPartitionManager.class);
- private final int receiveTimeoutMs = 5000;
- private final IEventHubReceiver receiver;
- private final EventHubSpoutConfig spoutConfig;
- private String lastOffset = FieldConstants.DefaultStartingOffset;
-
- public TridentPartitionManager(EventHubSpoutConfig spoutConfig, IEventHubReceiver receiver) {
- this.receiver = receiver;
- this.spoutConfig = spoutConfig;
- }
-
- @Override
- public boolean open(String offset) {
- try {
- if ((offset == null || offset.equals(FieldConstants.DefaultStartingOffset))
- && spoutConfig.getEnqueueTimeFilter() != 0) {
- receiver.open(new EventHubFilter(Instant.ofEpochMilli(spoutConfig.getEnqueueTimeFilter())));
- } else {
- receiver.open(new EventHubFilter(offset));
- }
- lastOffset = offset;
- return true;
- } catch (EventHubException ex) {
- logger.error("failed to open eventhub receiver: " + ex.getMessage());
- return false;
- }
- }
-
- @Override
- public void close() {
- receiver.close();
- }
-
- @Override
- public List<EventDataWrap> receiveBatch(String offset, int count) {
- List<EventDataWrap> batch = new ArrayList<EventDataWrap>(count);
- if (!offset.equals(lastOffset) || !receiver.isOpen()) {
- //re-establish connection to eventhub servers using the right offset
- //TBD: might be optimized with cache.
- close();
- if (!open(offset)) {
- return batch;
- }
- }
-
- for (int i = 0; i < count; ++i) {
- EventDataWrap ed = receiver.receive();
- if (ed == null) {
- break;
- }
- batch.add(ed);
- lastOffset = ed.getMessageId().getOffset();
- }
- return batch;
- }
-}
diff --git a/external/storm-eventhubs/src/main/resources/config.properties b/external/storm-eventhubs/src/main/resources/config.properties
deleted file mode 100755
index 2062e2473..000000000
--- a/external/storm-eventhubs/src/main/resources/config.properties
+++ /dev/null
@@ -1,45 +0,0 @@
-# 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
-# with the License. You may obtain a copy of the License at
-#
-# 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.
-eventhubspout.username = [username]
-
-eventhubspout.password = [password]
-
-eventhubspout.namespace = [namespace]
-
-eventhubspout.entitypath = [entitypath]
-
-eventhubspout.partitions.count = [partitioncount]
-
-# if not provided, will use storm's zookeeper settings
-# zookeeper.connectionstring=localhost:2181
-
-eventhubspout.checkpoint.interval = 10
-
-eventhub.receiver.credits = 1024
-
-# Defines how many messages can be pending (sent but not acked by storm)
-# per partition in eventhubspout. If set too large, your spout may throw
-# OutOfMemoryError because all pending messages are cached in the spout.
-eventhubspout.max.pending.messages.per.partition = 1024
-
-# Defines how many seconds in the past the spout uses to filter events in
-# the EventHubs entity when we first create the Storm topology. If offsets
-# have been saved in Zookeeper, we'll ignore this configuration.
-# A value of 0 means disable time based filtering when creating the receiver.
-eventhub.receiver.filter.timediff = 0
-
-# Uncomment to specify consumer group name here, or use the default
-#eventhubspout.consumer.group.name = yourconsumergroupname
\ No newline at end of file
diff --git a/external/storm-eventhubs/src/test/java/org/apache/storm/eventhubs/samples/AtMostOnceEventCount.java b/external/storm-eventhubs/src/test/java/org/apache/storm/eventhubs/samples/AtMostOnceEventCount.java
deleted file mode 100755
index 2c8ef2eb3..000000000
--- a/external/storm-eventhubs/src/test/java/org/apache/storm/eventhubs/samples/AtMostOnceEventCount.java
+++ /dev/null
@@ -1,54 +0,0 @@
-/*******************************************************************************
- * 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
- * with the License. You may obtain a copy of the License at
- *
- * 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.storm.eventhubs.samples;
-
-import java.io.Serializable;
-import org.apache.storm.eventhubs.spout.EventHubSpout;
-import org.apache.storm.eventhubs.spout.EventHubSpoutConfig;
-import org.apache.storm.eventhubs.spout.IEventHubReceiver;
-import org.apache.storm.eventhubs.spout.IPartitionManager;
-import org.apache.storm.eventhubs.spout.IPartitionManagerFactory;
-import org.apache.storm.eventhubs.spout.IStateStore;
-import org.apache.storm.eventhubs.spout.SimplePartitionManager;
-
-public class AtMostOnceEventCount extends EventCount implements Serializable {
- public static void main(String[] args) throws Exception {
- AtMostOnceEventCount scenario = new AtMostOnceEventCount();
-
- scenario.runScenario(args);
- }
-
- @Override
- protected EventHubSpout createEventHubSpout() {
- IPartitionManagerFactory pmFactory = new IPartitionManagerFactory() {
- private static final long serialVersionUID = 1L;
-
- @Override
- public IPartitionManager create(EventHubSpoutConfig spoutConfig,
- String partitionId, IStateStore stateStore,
- IEventHubReceiver receiver) {
- return new SimplePartitionManager(spoutConfig, partitionId,
- stateStore, receiver);
- }
- };
- EventHubSpout eventHubSpout = new EventHubSpout(
- spoutConfig, null, pmFactory, null);
- return eventHubSpout;
- }
-}
diff --git a/external/storm-eventhubs/src/test/java/org/apache/storm/eventhubs/samples/EventCount.java b/external/storm-eventhubs/src/test/java/org/apache/storm/eventhubs/samples/EventCount.java
deleted file mode 100755
index 8d00397a0..000000000
--- a/external/storm-eventhubs/src/test/java/org/apache/storm/eventhubs/samples/EventCount.java
+++ /dev/null
@@ -1,144 +0,0 @@
-/*******************************************************************************
- * 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
- * with the License. You may obtain a copy of the License at
- *
- * 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.storm.eventhubs.samples;
-
-import java.io.FileReader;
-import java.util.Properties;
-import org.apache.storm.Config;
-import org.apache.storm.StormSubmitter;
-import org.apache.storm.eventhubs.samples.bolt.GlobalCountBolt;
-import org.apache.storm.eventhubs.samples.bolt.PartialCountBolt;
-import org.apache.storm.eventhubs.spout.EventHubSpout;
-import org.apache.storm.eventhubs.spout.EventHubSpoutConfig;
-import org.apache.storm.generated.StormTopology;
-import org.apache.storm.topology.TopologyBuilder;
-
-/**
- * The basic scenario topology that uses EventHubSpout with PartialCountBolt and GlobalCountBolt. To submit this topology: storm jar
- * {jarfile} {classname} {topologyname} {spoutconffile}
- */
-public class EventCount {
- protected EventHubSpoutConfig spoutConfig;
- protected int numWorkers;
-
- public EventCount() {
- }
-
- public static void main(String[] args) throws Exception {
- EventCount scenario = new EventCount();
- scenario.runScenario(args);
- }
-
- protected void readEHConfig(String[] args) throws Exception {
- Properties properties = new Properties();
- if (args.length > 1) {
- properties.load(new FileReader(args[1]));
- } else {
- properties.load(EventCount.class.getClassLoader().getResourceAsStream(
- "Config.properties"));
- }
-
- String username = properties.getProperty("eventhubspout.username");
- String password = properties.getProperty("eventhubspout.password");
- String namespaceName = properties.getProperty("eventhubspout.namespace");
- String entityPath = properties.getProperty("eventhubspout.entitypath");
- String targetFqnAddress = properties.getProperty("eventhubspout.targetfqnaddress");
- String zkEndpointAddress = properties.getProperty("zookeeper.connectionstring");
- int partitionCount = Integer.parseInt(properties.getProperty("eventhubspout.partitions.count"));
- int checkpointIntervalInSeconds = Integer.parseInt(properties.getProperty("eventhubspout.checkpoint.interval"));
- int receiverCredits = Integer.parseInt(properties.getProperty("eventhub.receiver.credits"));
- String maxPendingMsgsPerPartitionStr = properties.getProperty("eventhubspout.max.pending.messages.per.partition");
- if (maxPendingMsgsPerPartitionStr == null) {
- maxPendingMsgsPerPartitionStr = "1024";
- }
- int maxPendingMsgsPerPartition = Integer.parseInt(maxPendingMsgsPerPartitionStr);
- String enqueueTimeDiffStr = properties.getProperty("eventhub.receiver.filter.timediff");
- if (enqueueTimeDiffStr == null) {
- enqueueTimeDiffStr = "0";
- }
- int enqueueTimeDiff = Integer.parseInt(enqueueTimeDiffStr);
- long enqueueTimeFilter = 0;
- if (enqueueTimeDiff != 0) {
- enqueueTimeFilter = System.currentTimeMillis() - enqueueTimeDiff * 1000;
- }
- String consumerGroupName = properties.getProperty("eventhubspout.consumer.group.name");
-
- System.out.println("Eventhub spout config: ");
- System.out.println(" partition count: " + partitionCount);
- System.out.println(" checkpoint interval: " + checkpointIntervalInSeconds);
- System.out.println(" receiver credits: " + receiverCredits);
- spoutConfig = new EventHubSpoutConfig(username, password,
- namespaceName, entityPath, partitionCount, zkEndpointAddress,
- checkpointIntervalInSeconds, receiverCredits, maxPendingMsgsPerPartition,
- enqueueTimeFilter);
-
- if (targetFqnAddress != null) {
- spoutConfig.setTargetAddress(targetFqnAddress);
- }
- spoutConfig.setConsumerGroupName(consumerGroupName);
-
- //set the number of workers to be the same as partition number.
- //the idea is to have a spout and a partial count bolt co-exist in one
- //worker to avoid shuffling messages across workers in storm cluster.
- numWorkers = spoutConfig.getPartitionCount();
-
- if (args.length > 0) {
- //set topology name so that sample Trident topology can use it as stream name.
- spoutConfig.setTopologyName(args[0]);
- }
- }
-
- protected EventHubSpout createEventHubSpout() {
- EventHubSpout eventHubSpout = new EventHubSpout(spoutConfig);
- return eventHubSpout;
- }
-
- protected StormTopology buildTopology(EventHubSpout eventHubSpout) {
- TopologyBuilder topologyBuilder = new TopologyBuilder();
-
- topologyBuilder.setSpout("EventHubsSpout", eventHubSpout, spoutConfig.getPartitionCount())
- .setNumTasks(spoutConfig.getPartitionCount());
- topologyBuilder.setBolt("PartialCountBolt", new PartialCountBolt(), spoutConfig.getPartitionCount())
- .localOrShuffleGrouping("EventHubsSpout").setNumTasks(spoutConfig.getPartitionCount());
- topologyBuilder.setBolt("GlobalCountBolt", new GlobalCountBolt(), 1)
- .globalGrouping("PartialCountBolt").setNumTasks(1);
- return topologyBuilder.createTopology();
- }
-
- protected void submitTopology(String[] args, StormTopology topology) throws Exception {
- Config config = new Config();
- config.setDebug(false);
- //Enable metrics
- config.registerMetricsConsumer(org.apache.storm.metric.LoggingMetricsConsumer.class, 1);
-
- String topoName = "test";
- if (args.length > 0) {
- topoName = args[0];
- }
- config.setNumWorkers(numWorkers);
- StormSubmitter.submitTopology(topoName, config, topology);
- }
-
- protected void runScenario(String[] args) throws Exception {
- readEHConfig(args);
- EventHubSpout eventHubSpout = createEventHubSpout();
- StormTopology topology = buildTopology(eventHubSpout);
- submitTopology(args, topology);
- }
-}
diff --git a/external/storm-eventhubs/src/test/java/org/apache/storm/eventhubs/samples/EventHubLoop.java b/external/storm-eventhubs/src/test/java/org/apache/storm/eventhubs/samples/EventHubLoop.java
deleted file mode 100755
index 55d4cf221..000000000
--- a/external/storm-eventhubs/src/test/java/org/apache/storm/eventhubs/samples/EventHubLoop.java
+++ /dev/null
@@ -1,52 +0,0 @@
-/*******************************************************************************
- * 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
- * with the License. You may obtain a copy of the License at
- *
- * 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.storm.eventhubs.samples;
-
-import org.apache.storm.eventhubs.bolt.EventHubBolt;
-import org.apache.storm.eventhubs.bolt.EventHubBoltConfig;
-import org.apache.storm.eventhubs.spout.EventHubSpout;
-import org.apache.storm.generated.StormTopology;
-import org.apache.storm.topology.TopologyBuilder;
-
-/**
- * A sample topology that loops message back to EventHub
- */
-public class EventHubLoop extends EventCount {
-
- public static void main(String[] args) throws Exception {
- EventHubLoop scenario = new EventHubLoop();
- scenario.runScenario(args);
- }
-
- @Override
- protected StormTopology buildTopology(EventHubSpout eventHubSpout) {
- TopologyBuilder topologyBuilder = new TopologyBuilder();
-
- topologyBuilder.setSpout("EventHubsSpout", eventHubSpout, spoutConfig.getPartitionCount())
- .setNumTasks(spoutConfig.getPartitionCount());
- EventHubBoltConfig boltConfig = new EventHubBoltConfig(spoutConfig.getConnectionString(),
- spoutConfig.getEntityPath(), true);
-
- EventHubBolt eventHubBolt = new EventHubBolt(boltConfig);
- int boltTasks = spoutConfig.getPartitionCount();
- topologyBuilder.setBolt("EventHubsBolt", eventHubBolt, boltTasks)
- .localOrShuffleGrouping("EventHubsSpout").setNumTasks(boltTasks);
- return topologyBuilder.createTopology();
- }
-}
diff --git a/external/storm-eventhubs/src/test/java/org/apache/storm/eventhubs/samples/OpaqueTridentEventCount.java b/external/storm-eventhubs/src/test/java/org/apache/storm/eventhubs/samples/OpaqueTridentEventCount.java
deleted file mode 100755
index abcad5ae1..000000000
--- a/external/storm-eventhubs/src/test/java/org/apache/storm/eventhubs/samples/OpaqueTridentEventCount.java
+++ /dev/null
@@ -1,54 +0,0 @@
-/*******************************************************************************
- * 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
- * with the License. You may obtain a copy of the License at
- *
- * 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.storm.eventhubs.samples;
-
-import org.apache.storm.eventhubs.samples.TransactionalTridentEventCount.LoggingFilter;
-import org.apache.storm.eventhubs.spout.EventHubSpout;
-import org.apache.storm.eventhubs.trident.OpaqueTridentEventHubSpout;
-import org.apache.storm.generated.StormTopology;
-import org.apache.storm.trident.TridentState;
-import org.apache.storm.trident.TridentTopology;
-import org.apache.storm.trident.operation.builtin.Count;
-import org.apache.storm.trident.operation.builtin.Sum;
-import org.apache.storm.trident.testing.MemoryMapState;
-import org.apache.storm.tuple.Fields;
-
-/**
- * A simple Trident topology uses OpaqueTridentEventHubSpout
- */
-public class OpaqueTridentEventCount extends EventCount {
- public static void main(String[] args) throws Exception {
- OpaqueTridentEventCount scenario = new OpaqueTridentEventCount();
- scenario.runScenario(args);
- }
-
- @Override
- protected StormTopology buildTopology(EventHubSpout eventHubSpout) {
- TridentTopology topology = new TridentTopology();
-
- OpaqueTridentEventHubSpout spout = new OpaqueTridentEventHubSpout(spoutConfig);
- TridentState state = topology.newStream("stream-" + spoutConfig.getTopologyName(), spout)
- .parallelismHint(spoutConfig.getPartitionCount())
- .aggregate(new Count(), new Fields("partial-count"))
- .persistentAggregate(new MemoryMapState.Factory(), new Fields("partial-count"), new Sum(),
- new Fields("count"));
- state.newValuesStream().each(new Fields("count"), new LoggingFilter("got count: ", 10000));
- return topology.build();
- }
-}
diff --git a/external/storm-eventhubs/src/test/java/org/apache/storm/eventhubs/samples/TransactionalTridentEventCount.java b/external/storm-eventhubs/src/test/java/org/apache/storm/eventhubs/samples/TransactionalTridentEventCount.java
deleted file mode 100755
index 646dc9cd2..000000000
--- a/external/storm-eventhubs/src/test/java/org/apache/storm/eventhubs/samples/TransactionalTridentEventCount.java
+++ /dev/null
@@ -1,81 +0,0 @@
-/*******************************************************************************
- * 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
- * with the License. You may obtain a copy of the License at
- *
- * 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.storm.eventhubs.samples;
-
-import org.apache.storm.eventhubs.spout.EventHubSpout;
-import org.apache.storm.eventhubs.trident.TransactionalTridentEventHubSpout;
-import org.apache.storm.generated.StormTopology;
-import org.apache.storm.trident.TridentState;
-import org.apache.storm.trident.TridentTopology;
-import org.apache.storm.trident.operation.BaseFilter;
-import org.apache.storm.trident.operation.builtin.Count;
-import org.apache.storm.trident.operation.builtin.Sum;
-import org.apache.storm.trident.testing.MemoryMapState;
-import org.apache.storm.trident.tuple.TridentTuple;
-import org.apache.storm.tuple.Fields;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * A simple Trident topology uses TransactionalTridentEventHubSpout
- */
-public class TransactionalTridentEventCount extends EventCount {
- public static void main(String[] args) throws Exception {
- TransactionalTridentEventCount scenario = new TransactionalTridentEventCount();
- scenario.runScenario(args);
- }
-
- @Override
- protected StormTopology buildTopology(EventHubSpout eventHubSpout) {
- TridentTopology topology = new TridentTopology();
-
- TransactionalTridentEventHubSpout spout = new TransactionalTridentEventHubSpout(spoutConfig);
- TridentState state = topology.newStream("stream-" + spoutConfig.getTopologyName(), spout)
- .parallelismHint(spoutConfig.getPartitionCount())
- .aggregate(new Count(), new Fields("partial-count"))
- .persistentAggregate(new MemoryMapState.Factory(), new Fields("partial-count"), new Sum(),
- new Fields("count"));
- state.newValuesStream().each(new Fields("count"), new LoggingFilter("got count: ", 10000));
- return topology.build();
- }
-
- public static class LoggingFilter extends BaseFilter {
- private static final long serialVersionUID = 1L;
- private static final Logger logger = LoggerFactory.getLogger(LoggingFilter.class);
- private final String prefix;
- private final long logIntervalMs;
- private long lastTime;
-
- public LoggingFilter(String prefix, int logIntervalMs) {
- this.prefix = prefix;
- this.logIntervalMs = logIntervalMs;
- lastTime = System.nanoTime();
- }
-
- @Override
- public boolean isKeep(TridentTuple tuple) {
- long now = System.nanoTime();
- if (logIntervalMs < (now - lastTime) / 1000000) {
- logger.info(prefix + tuple.toString());
- lastTime = now;
- }
- return false;
- }
- }
-}
diff --git a/external/storm-eventhubs/src/test/java/org/apache/storm/eventhubs/samples/bolt/GlobalCountBolt.java b/external/storm-eventhubs/src/test/java/org/apache/storm/eventhubs/samples/bolt/GlobalCountBolt.java
deleted file mode 100755
index cf3fb7647..000000000
--- a/external/storm-eventhubs/src/test/java/org/apache/storm/eventhubs/samples/bolt/GlobalCountBolt.java
+++ /dev/null
@@ -1,87 +0,0 @@
-/*******************************************************************************
- * 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
- * with the License. You may obtain a copy of the License at
- *
- * 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.storm.eventhubs.samples.bolt;
-
-import java.util.HashMap;
-import java.util.Map;
-import org.apache.storm.Config;
-import org.apache.storm.metric.api.IMetric;
-import org.apache.storm.task.TopologyContext;
-import org.apache.storm.topology.BasicOutputCollector;
-import org.apache.storm.topology.OutputFieldsDeclarer;
-import org.apache.storm.topology.base.BaseBasicBolt;
-import org.apache.storm.tuple.Tuple;
-import org.apache.storm.utils.TupleUtils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * Globally count number of messages
- */
-public class GlobalCountBolt extends BaseBasicBolt {
- private static final long serialVersionUID = 1L;
- private static final Logger logger = LoggerFactory
- .getLogger(GlobalCountBolt.class);
- private long globalCount;
- private long globalCountDiff;
- private long lastMetricsTime;
- private long throughput;
-
- @Override
- public void prepare(Map<String, Object> config, TopologyContext context) {
- globalCount = 0;
- globalCountDiff = 0;
- lastMetricsTime = System.nanoTime();
- context.registerMetric("GlobalMessageCount", new IMetric() {
- @Override
- public Object getValueAndReset() {
- long now = System.nanoTime();
- long millis = (now - lastMetricsTime) / 1000000;
- throughput = globalCountDiff / millis * 1000;
- Map<String, Object> values = new HashMap<>();
- values.put("global_count", globalCount);
- values.put("throughput", throughput);
- lastMetricsTime = now;
- globalCountDiff = 0;
- return values;
- }
- }, (Integer) config.get(Config.TOPOLOGY_BUILTIN_METRICS_BUCKET_SIZE_SECS));
- }
-
- @Override
- public void execute(Tuple tuple, BasicOutputCollector collector) {
- if (TupleUtils.isTick(tuple)) {
- return;
- }
-
- int partial = (Integer) tuple.getValueByField("partial_count");
- globalCount += partial;
- globalCountDiff += partial;
- if ((globalCountDiff == partial) && (globalCount != globalCountDiff)) {
- //metrics has just been collected, let's also log it
- logger.info("Current throughput (messages/second): " + throughput);
- }
- }
-
- @Override
- public void declareOutputFields(OutputFieldsDeclarer declarer) {
-
- }
-
-}
diff --git a/external/storm-eventhubs/src/test/java/org/apache/storm/eventhubs/samples/bolt/PartialCountBolt.java b/external/storm-eventhubs/src/test/java/org/apache/storm/eventhubs/samples/bolt/PartialCountBolt.java
deleted file mode 100755
index e51dc0851..000000000
--- a/external/storm-eventhubs/src/test/java/org/apache/storm/eventhubs/samples/bolt/PartialCountBolt.java
+++ /dev/null
@@ -1,67 +0,0 @@
-/*******************************************************************************
- * 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
- * with the License. You may obtain a copy of the License at
- *
- * 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.storm.eventhubs.samples.bolt;
-
-import java.util.Map;
-import org.apache.storm.task.TopologyContext;
-import org.apache.storm.topology.BasicOutputCollector;
-import org.apache.storm.topology.OutputFieldsDeclarer;
-import org.apache.storm.topology.base.BaseBasicBolt;
-import org.apache.storm.tuple.Fields;
-import org.apache.storm.tuple.Tuple;
-import org.apache.storm.tuple.Values;
-import org.apache.storm.utils.TupleUtils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * Partially count number of messages from EventHubs
- */
-public class PartialCountBolt extends BaseBasicBolt {
- private static final long serialVersionUID = 1L;
- private static final Logger logger = LoggerFactory
- .getLogger(PartialCountBolt.class);
- private static final int PartialCountBatchSize = 1000;
-
- private int partialCount;
-
- @Override
- public void prepare(Map<String, Object> topoConf, TopologyContext context) {
- partialCount = 0;
- }
-
- @Override
- public void execute(Tuple tuple, BasicOutputCollector collector) {
- if (TupleUtils.isTick(tuple)) {
- return;
- }
-
- partialCount++;
- if (partialCount == PartialCountBatchSize) {
- collector.emit(new Values(PartialCountBatchSize));
- partialCount = 0;
- }
- }
-
- @Override
- public void declareOutputFields(OutputFieldsDeclarer declarer) {
- declarer.declare(new Fields("partial_count"));
- }
-
-}
diff --git a/external/storm-eventhubs/src/test/java/org/apache/storm/eventhubs/spout/EventHubReceiverMock.java b/external/storm-eventhubs/src/test/java/org/apache/storm/eventhubs/spout/EventHubReceiverMock.java
deleted file mode 100755
index 7f6e6976a..000000000
--- a/external/storm-eventhubs/src/test/java/org/apache/storm/eventhubs/spout/EventHubReceiverMock.java
+++ /dev/null
@@ -1,84 +0,0 @@
-/*******************************************************************************
- * 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
- * with the License. You may obtain a copy of the License at
- *
- * 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.storm.eventhubs.spout;
-
-import com.microsoft.azure.eventhubs.EventData;
-import java.util.Map;
-
-/**
- * A mock receiver that emits fake data with offset starting from given offset and increase by 1 each time.
- */
-public class EventHubReceiverMock implements IEventHubReceiver {
- private static boolean isPaused = false;
- private final String partitionId;
- private long currentOffset;
- private boolean isOpen;
-
- public EventHubReceiverMock(String pid) {
- partitionId = pid;
- isPaused = false;
- }
-
- /**
- * Use this method to pause/resume all the receivers. If paused all receiver will return null.
- *
- * @param val
- */
- public static void setPause(boolean val) {
- isPaused = val;
- }
-
- @Override
- public void open(IEventFilter filter) throws EventHubException {
- currentOffset = filter.getOffset() != null ?
- Long.parseLong(filter.getOffset()) :
- filter.getTime().toEpochMilli();
- isOpen = true;
- }
-
- @Override
- public void close() {
- isOpen = false;
- }
-
- @Override
- public boolean isOpen() {
- return isOpen;
- }
-
- @Override
- public EventDataWrap receive() {
- if (isPaused) {
- return null;
- }
-
- currentOffset++;
-
- //the body of the message is "message" + currentOffset, e.g. "message123"
-
- MessageId mid = new MessageId(partitionId, "" + currentOffset, currentOffset);
- EventData ed = new EventData(("message" + currentOffset).getBytes());
- return EventDataWrap.create(ed, mid);
- }
-
- @Override
- public Map<String, Object> getMetricsData() {
- return null;
- }
-}
diff --git a/external/storm-eventhubs/src/test/java/org/apache/storm/eventhubs/spout/EventHubSpoutCallerMock.java b/external/storm-eventhubs/src/test/java/org/apache/storm/eventhubs/spout/EventHubSpoutCallerMock.java
deleted file mode 100755
index 16f3b80cb..000000000
--- a/external/storm-eventhubs/src/test/java/org/apache/storm/eventhubs/spout/EventHubSpoutCallerMock.java
+++ /dev/null
@@ -1,92 +0,0 @@
-/*******************************************************************************
- * 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
- * with the License. You may obtain a copy of the License at
- *
- * 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.storm.eventhubs.spout;
-
-import org.apache.storm.spout.SpoutOutputCollector;
-
-/**
- * Mocks EventHubSpout's caller (storm framework)
- */
-public class EventHubSpoutCallerMock {
- public static final String statePathPrefix = "/eventhubspout/TestTopo/namespace/entityname/partitions/";
- EventHubSpout spout;
- private IStateStore stateStore;
- private SpoutOutputCollectorMock collector;
-
- public EventHubSpoutCallerMock(int totalPartitions,
- int totalTasks, int taskIndex, int checkpointInterval) {
- stateStore = new StateStoreMock();
- EventHubSpoutConfig conf = new EventHubSpoutConfig("username", "password",
- "namespace", "entityname", totalPartitions, "zookeeper", checkpointInterval,
- 1024);
- conf.setTopologyName("TestTopo");
-
- IEventHubReceiverFactory recvFactory = new IEventHubReceiverFactory() {
- @Override
- public IEventHubReceiver create(EventHubSpoutConfig config,
- String partitionId) {
- return new EventHubReceiverMock(partitionId);
- }
- };
- //mock state store and receiver
- spout = new EventHubSpout(conf, stateStore, null, recvFactory);
-
- collector = new SpoutOutputCollectorMock();
-
- try {
- spout.preparePartitions(null, totalTasks, taskIndex, new SpoutOutputCollector(collector));
- } catch (Exception ex) {
- }
- }
-
- /**
- * Execute a sequence of calls to EventHubSpout.
- *
- * @param callSequence: is represented as a string of commands, e.g. "r,r,r,r,a1,f2,...". The commands are: r[N]: receive() called N
- * times aP_X: ack(P_X), partition: P, offset: X fP_Y: fail(P_Y), partition: P, offset: Y
- */
- public String execute(String callSequence) {
- String[] cmds = callSequence.split(",");
- for (String cmd : cmds) {
- if (cmd.startsWith("r")) {
- int count = 1;
- if (cmd.length() > 1) {
- count = Integer.parseInt(cmd.substring(1));
- }
- for (int i = 0; i < count; ++i) {
- spout.nextTuple();
- }
- } else if (cmd.startsWith("a")) {
- String[] midStrs = cmd.substring(1).split("_");
- MessageId msgId = new MessageId(midStrs[0], midStrs[1], Long.parseLong(midStrs[1]));
- spout.ack(msgId);
- } else if (cmd.startsWith("f")) {
- String[] midStrs = cmd.substring(1).split("_");
- MessageId msgId = new MessageId(midStrs[0], midStrs[1], Long.parseLong(midStrs[1]));
- spout.fail(msgId);
- }
- }
- return collector.getOffsetSequenceAndReset();
- }
-
- public String getCheckpoint(int partitionIndex) {
- String statePath = statePathPrefix + partitionIndex;
- return stateStore.readData(statePath);
- }
-}
diff --git a/external/storm-eventhubs/src/test/java/org/apache/storm/eventhubs/spout/PartitionManagerCallerMock.java b/external/storm-eventhubs/src/test/java/org/apache/storm/eventhubs/spout/PartitionManagerCallerMock.java
deleted file mode 100755
index 364990e57..000000000
--- a/external/storm-eventhubs/src/test/java/org/apache/storm/eventhubs/spout/PartitionManagerCallerMock.java
+++ /dev/null
@@ -1,95 +0,0 @@
-/*******************************************************************************
- * 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
- * with the License. You may obtain a copy of the License at
- *
- * 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.storm.eventhubs.spout;
-
-/**
- * This mock exercises PartitionManager
- */
-public class PartitionManagerCallerMock {
- public static final String statePath = "/eventhubspout/TestTopo/namespace/entityname/partitions/1";
- private IPartitionManager pm;
- private IStateStore stateStore;
-
- public PartitionManagerCallerMock(String partitionId) {
- this(partitionId, 0);
- }
-
- public PartitionManagerCallerMock(String partitionId, long enqueueTimeFilter) {
- EventHubReceiverMock receiver = new EventHubReceiverMock(partitionId);
- EventHubSpoutConfig conf = new EventHubSpoutConfig("username", "password",
- "namespace", "entityname", 16, "zookeeper", 10, 1024, 1024, enqueueTimeFilter);
- conf.setTopologyName("TestTopo");
- stateStore = new StateStoreMock();
- this.pm = new PartitionManager(conf, partitionId, stateStore, receiver);
-
- stateStore.open();
- try {
- pm.open();
- } catch (Exception ex) {
- }
- }
-
- /**
- * Execute a sequence of calls to Partition Manager.
- *
- * @param callSequence: is represented as a string of commands, e.g. "r,r,r,r,a1,f2,...". The commands are: r[N]: receive() called N
- * times aX: ack(X) fY: fail(Y)
- * @return the sequence of messages the receive call returns
- */
- public String execute(String callSequence) {
-
- String[] cmds = callSequence.split(",");
- StringBuilder ret = new StringBuilder();
- for (String cmd : cmds) {
- if (cmd.startsWith("r")) {
- int count = 1;
- if (cmd.length() > 1) {
- count = Integer.parseInt(cmd.substring(1));
- }
- for (int i = 0; i < count; ++i) {
- EventDataWrap ed = pm.receive();
- if (ed == null) {
- ret.append("null,");
- } else {
- ret.append(ed.getMessageId().getOffset());
- ret.append(",");
- }
- }
- } else if (cmd.startsWith("a")) {
- pm.ack(cmd.substring(1));
- } else if (cmd.startsWith("f")) {
- pm.fail(cmd.substring(1));
- }
- }
- if (ret.length() > 0) {
- ret.setLength(ret.length() - 1);
- }
- return ret.toString();
- }
-
- /**
- * Exercise the IPartitionManager.checkpoint() method
- *
- * @return the offset that we write to state store
- */
- public String checkpoint() {
- pm.checkpoint();
- return stateStore.readData(statePath);
- }
-}
diff --git a/external/storm-eventhubs/src/test/java/org/apache/storm/eventhubs/spout/SpoutOutputCollectorMock.java b/external/storm-eventhubs/src/test/java/org/apache/storm/eventhubs/spout/SpoutOutputCollectorMock.java
deleted file mode 100755
index 37dfdd5cd..000000000
--- a/external/storm-eventhubs/src/test/java/org/apache/storm/eventhubs/spout/SpoutOutputCollectorMock.java
+++ /dev/null
@@ -1,71 +0,0 @@
-/*******************************************************************************
- * 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
- * with the License. You may obtain a copy of the License at
- *
- * 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.storm.eventhubs.spout;
-
-import java.util.List;
-import org.apache.storm.spout.ISpoutOutputCollector;
-
-/**
- * Mock of ISpoutOutputCollector
- */
-public class SpoutOutputCollectorMock implements ISpoutOutputCollector {
- //comma separated offsets
- StringBuilder emittedOffset;
-
- public SpoutOutputCollectorMock() {
- emittedOffset = new StringBuilder();
- }
-
- public String getOffsetSequenceAndReset() {
- String ret = null;
- if (emittedOffset.length() > 0) {
- emittedOffset.setLength(emittedOffset.length() - 1);
- ret = emittedOffset.toString();
- emittedOffset.setLength(0);
- }
- return ret;
- }
-
- @Override
- public List<Integer> emit(String streamId, List<Object> tuple, Object messageId) {
- MessageId mid = (MessageId) messageId;
- String pid = mid.getPartitionId();
- String offset = mid.getOffset();
- emittedOffset.append(pid + "_" + offset + ",");
- return null;
- }
-
- @Override
- public void emitDirect(int arg0, String arg1, List<Object> arg2, Object arg3) {
- }
-
- @Override
- public void flush() {
- // NO-OP
- }
-
- @Override
- public void reportError(Throwable arg0) {
- }
-
- @Override
- public long getPendingCount() {
- return 0;
- }
-}
diff --git a/external/storm-eventhubs/src/test/java/org/apache/storm/eventhubs/spout/StateStoreMock.java b/external/storm-eventhubs/src/test/java/org/apache/storm/eventhubs/spout/StateStoreMock.java
deleted file mode 100755
index 7bfcd93d9..000000000
--- a/external/storm-eventhubs/src/test/java/org/apache/storm/eventhubs/spout/StateStoreMock.java
+++ /dev/null
@@ -1,54 +0,0 @@
-/*******************************************************************************
- * 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
- * with the License. You may obtain a copy of the License at
- *
- * 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.storm.eventhubs.spout;
-
-import java.util.HashMap;
-import java.util.Map;
-
-/**
- * A state store mocker
- */
-public class StateStoreMock implements IStateStore {
- Map<String, String> myDataMap;
-
- @Override
- public void open() {
- myDataMap = new HashMap<String, String>();
- }
-
- @Override
- public void close() {
- myDataMap = null;
- }
-
- @Override
- public void saveData(String path, String data) {
- if (myDataMap != null) {
- myDataMap.put(path, data);
- }
- }
-
- @Override
- public String readData(String path) {
- if (myDataMap != null) {
- return myDataMap.get(path);
- }
- return null;
- }
-}
diff --git a/external/storm-eventhubs/src/test/java/org/apache/storm/eventhubs/spout/TestEventData.java b/external/storm-eventhubs/src/test/java/org/apache/storm/eventhubs/spout/TestEventData.java
deleted file mode 100755
index 0ea3ac023..000000000
--- a/external/storm-eventhubs/src/test/java/org/apache/storm/eventhubs/spout/TestEventData.java
+++ /dev/null
@@ -1,48 +0,0 @@
-/*
- * 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
- * with the License. You may obtain a copy of the License at
- *
- * 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.storm.eventhubs.spout;
-
-import org.junit.jupiter.api.AfterEach;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
-
-import static org.junit.jupiter.api.Assertions.assertTrue;
-
-public class TestEventData {
-
- @BeforeEach
- public void setUp() {
- }
-
- @AfterEach
- public void tearDown() {
- }
-
- @Test
- public void testEventDataComparison() {
-
- MessageId messageId1 = MessageId.create(null, "3", 1);
- EventDataWrap eventData1 = EventDataWrap.create(null, messageId1);
-
- MessageId messageId2 = MessageId.create(null, "13", 2);
- EventDataWrap eventData2 = EventDataWrap.create(null, messageId2);
-
- assertTrue(eventData2.compareTo(eventData1) > 0);
- }
-}
diff --git a/external/storm-eventhubs/src/test/java/org/apache/storm/eventhubs/spout/TestEventHubSpout.java b/external/storm-eventhubs/src/test/java/org/apache/storm/eventhubs/spout/TestEventHubSpout.java
deleted file mode 100755
index 9bb8ed2a2..000000000
--- a/external/storm-eventhubs/src/test/java/org/apache/storm/eventhubs/spout/TestEventHubSpout.java
+++ /dev/null
@@ -1,75 +0,0 @@
-/*******************************************************************************
- * 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
- * with the License. You may obtain a copy of the License at
- *
- * 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.storm.eventhubs.spout;
-
-import org.junit.jupiter.api.AfterEach;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
-
-import static org.junit.jupiter.api.Assertions.assertEquals;
-
-
-public class TestEventHubSpout {
-
- @BeforeEach
- public void setUp() {
- }
-
- @AfterEach
- public void tearDown() {
- }
-
- @Test
- public void testSpoutConfig() {
- EventHubSpoutConfig conf = new EventHubSpoutConfig("username", "pas\\s+w/ord",
- "namespace", "entityname", 16);
- conf.setZkConnectionString("zookeeper");
- conf.setCheckpointIntervalInSeconds(1);
- assertEquals(conf.getConnectionString(),
- "Endpoint=amqps://namespace.servicebus.windows.net;EntityPath=entityname;SharedAccessKeyName=username;" +
- "SharedAccessKey=pas\\s+w/ord;OperationTimeout=PT1M;RetryPolicy=Default");
- }
-
- @Test
- public void testSpoutBasic() {
- //This spout owns 2 partitions: 6 and 14
- EventHubSpoutCallerMock mock = new EventHubSpoutCallerMock(16, 8, 6, 10);
- String result = mock.execute("r6,f6_0,a6_1,a6_2,a14_0,a14_2,r4,f14_1,r2");
- assertEquals("6_0,14_0,6_1,14_1,6_2,14_2,6_0,14_3,6_3,14_4,6_4,14_1", result);
- }
-
- @Test
- public void testSpoutCheckpoint() {
- //Make sure that even though nextTuple() doesn't receive valid data,
- //the offset will be checkpointed after checkpointInterval seconds.
-
- //This spout owns 1 partitions: 6
- EventHubSpoutCallerMock mock = new EventHubSpoutCallerMock(8, 8, 6, 1);
- String result = mock.execute("r6,a6_0,a6_1,a6_2");
- try {
- Thread.sleep(2000);
- } catch (InterruptedException ex) {
- }
- EventHubReceiverMock.setPause(true);
- result = mock.execute("r3");
- EventHubReceiverMock.setPause(false);
- assertEquals("3", mock.getCheckpoint(6));
- }
-
-}
diff --git a/external/storm-eventhubs/src/test/java/org/apache/storm/eventhubs/spout/TestPartitionManager.java b/external/storm-eventhubs/src/test/java/org/apache/storm/eventhubs/spout/TestPartitionManager.java
deleted file mode 100755
index beb5199cd..000000000
--- a/external/storm-eventhubs/src/test/java/org/apache/storm/eventhubs/spout/TestPartitionManager.java
+++ /dev/null
@@ -1,119 +0,0 @@
-/*******************************************************************************
- * 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
- * with the License. You may obtain a copy of the License at
- *
- * 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.storm.eventhubs.spout;
-
-import org.junit.jupiter.api.AfterEach;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
-
-import static org.junit.jupiter.api.Assertions.assertEquals;
-
-public class TestPartitionManager {
-
- @BeforeEach
- public void setUp() throws Exception {
- }
-
- @AfterEach
- public void tearDown() throws Exception {
- }
-
- @Test
- public void testPartitionManagerNoFail() {
- PartitionManagerCallerMock mock
- = new PartitionManagerCallerMock("1");
- String result = mock.execute("r,r,r,a0,a1,a2,r");
- assertEquals("0,1,2,3", result);
- }
-
- @Test
- public void testPartitionManagerResend() {
- PartitionManagerCallerMock mock
- = new PartitionManagerCallerMock("1");
- String result = mock.execute("r,a0,r,r,r,f3,r,f2,f1,r,r,a1,a2,a3,r");
- assertEquals("0,1,2,3,3,1,2,4", result);
- }
-
- @Test
- public void testPMCheckpointWithPending() {
- PartitionManagerCallerMock mock
- = new PartitionManagerCallerMock("1");
- mock.execute("r,r,r");
- //no ack, so return the first of pending list
- assertEquals("0", mock.checkpoint());
- mock.execute("a0,a2");
- //still need to return the first of pending list
- assertEquals("1", mock.checkpoint());
- }
-
- @Test
- public void testPMCheckpointWithResend() {
- PartitionManagerCallerMock mock
- = new PartitionManagerCallerMock("1");
- mock.execute("r,r,r,f2,f1,f0");
- //pending is empty, return the smallest in toResend
- assertEquals("0", mock.checkpoint());
- mock.execute("r,a0");
- //pending is still empty
- assertEquals("1", mock.checkpoint());
- }
-
- @Test
- public void testPMCheckpointWithPendingAndResend() {
- PartitionManagerCallerMock mock
- = new PartitionManagerCallerMock("1");
- mock.execute("r,r,r,f2,f1");
- //return the smaller of pending and toResend
- assertEquals("0", mock.checkpoint());
- mock.execute("a0,r");
- //now pending: [3], toResend: [1,2]
- assertEquals("1", mock.checkpoint());
- }
-
- @Test
- public void testPMCheckpointWithNoPendingAndNoResend() {
- PartitionManagerCallerMock mock
- = new PartitionManagerCallerMock("1");
- //if no event sent, no checkpoint shall be created
- assertEquals(null, mock.checkpoint());
- mock.execute("r,r,r,f2,f1,r,r,a2,a1,a0");
- //all events are sent successfully, return last sent offset
- assertEquals("2", mock.checkpoint());
- }
-
- @Test
- public void testPartitionManagerMaxPendingMessages() {
- PartitionManagerCallerMock mock
- = new PartitionManagerCallerMock("1");
- String result = mock.execute("r1024");
- //any receive call after exceeding max pending messages results in null
- result = mock.execute("r2");
- assertEquals("null,null", result);
- result = mock.execute("a0,a1,r2");
- assertEquals("1024,1025", result);
- }
-
- @Test
- public void testPartitionManagerEnqueueTimeFilter() {
- PartitionManagerCallerMock mock
- = new PartitionManagerCallerMock("1", 123456);
- String result = mock.execute("r2");
- assertEquals("123457,123458", result);
- }
-}
diff --git a/external/storm-eventhubs/src/test/java/org/apache/storm/eventhubs/trident/TestTransactionalTridentEmitter.java b/external/storm-eventhubs/src/test/java/org/apache/storm/eventhubs/trident/TestTransactionalTridentEmitter.java
deleted file mode 100755
index f1ea226d8..000000000
--- a/external/storm-eventhubs/src/test/java/org/apache/storm/eventhubs/trident/TestTransactionalTridentEmitter.java
+++ /dev/null
@@ -1,92 +0,0 @@
-/*******************************************************************************
- * 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
- * with the License. You may obtain a copy of the License at
- *
- * 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.storm.eventhubs.trident;
-
-import java.util.Map;
-import org.apache.storm.eventhubs.spout.EventHubReceiverMock;
-import org.apache.storm.eventhubs.spout.EventHubSpoutConfig;
-import org.apache.storm.eventhubs.spout.IEventHubReceiver;
-import org.apache.storm.eventhubs.spout.IEventHubReceiverFactory;
-import org.junit.jupiter.api.AfterEach;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
-
-import static org.junit.jupiter.api.Assertions.assertTrue;
-
-public class TestTransactionalTridentEmitter {
- private final int batchSize = 32;
- private TransactionalTridentEventHubEmitter emitter;
- private Partition partition;
- private TridentCollectorMock collectorMock;
-
- @BeforeEach
- public void setUp() throws Exception {
- EventHubSpoutConfig conf = new EventHubSpoutConfig("username", "password",
- "namespace", "entityname", 16, "zookeeper");
- conf.setTopologyName("TestTopo");
- IEventHubReceiverFactory recvFactory = new IEventHubReceiverFactory() {
- @Override
- public IEventHubReceiver create(EventHubSpoutConfig config,
- String partitionId) {
- return new EventHubReceiverMock(partitionId);
- }
- };
- partition = new Partition(conf, "0");
- emitter = new TransactionalTridentEventHubEmitter(conf, batchSize, null, recvFactory);
- collectorMock = new TridentCollectorMock();
- }
-
- @AfterEach
- public void tearDown() throws Exception {
- emitter.close();
- emitter = null;
- }
-
- @Test
- public void testEmitInSequence() {
- //test the happy path, emit batches in sequence
- Map<String, Object> meta = emitter.emitPartitionBatchNew(null, collectorMock, partition, null);
- String collected = collectorMock.getBuffer();
- assertTrue(collected.startsWith("message" + 0));
- //System.out.println("collected: " + collected);
- collectorMock.clear();
-
- emitter.emitPartitionBatchNew(null, collectorMock, partition, meta);
- collected = collectorMock.getBuffer();
- //System.out.println("collected: " + collected);
- assertTrue(collected.startsWith("message" + batchSize));
- }
-
- @Test
- public void testReEmit() {
- //test we can re-emit the second batch
- Map<String, Object> meta = emitter.emitPartitionBatchNew(null, collectorMock, partition, null);
- collectorMock.clear();
-
- //emit second batch
- Map<String, Object> meta1 = emitter.emitPartitionBatchNew(null, collectorMock, partition, meta);
- String collected0 = collectorMock.getBuffer();
- collectorMock.clear();
-
- //re-emit second batch
- emitter.emitPartitionBatch(null, collectorMock, partition, meta1);
- String collected1 = collectorMock.getBuffer();
- assertTrue(collected0.equals(collected1));
- }
-}
diff --git a/external/storm-eventhubs/src/test/java/org/apache/storm/eventhubs/trident/TridentCollectorMock.java b/external/storm-eventhubs/src/test/java/org/apache/storm/eventhubs/trident/TridentCollectorMock.java
deleted file mode 100755
index 59f70025e..000000000
--- a/external/storm-eventhubs/src/test/java/org/apache/storm/eventhubs/trident/TridentCollectorMock.java
+++ /dev/null
@@ -1,57 +0,0 @@
-/*******************************************************************************
- * 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
- * with the License. You may obtain a copy of the License at
- *
- * 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.storm.eventhubs.trident;
-
-import java.util.List;
-import org.apache.storm.trident.operation.TridentCollector;
-
-/**
- * A mock of TridentCollector
- */
-public class TridentCollectorMock implements TridentCollector {
- StringBuilder buffer;
-
- public TridentCollectorMock() {
- buffer = new StringBuilder();
- }
-
- @Override
- public void emit(List<Object> tuples) {
- for (Object o : tuples) {
- buffer.append(o.toString());
- }
- }
-
- @Override
- public void flush() {
- // NO-OP
- }
-
- @Override
- public void reportError(Throwable arg0) {
- }
-
- public void clear() {
- buffer.setLength(0);
- }
-
- public String getBuffer() {
- return buffer.toString();
- }
-}
diff --git a/pom.xml b/pom.xml
index e9a7ea309..2e5799a08 100644
--- a/pom.xml
+++ b/pom.xml
@@ -493,7 +493,6 @@
<module>external/storm-hive</module>
<module>external/storm-jdbc</module>
<module>external/storm-redis</module>
- <module>external/storm-eventhubs</module>
<module>external/storm-elasticsearch</module>
<module>external/storm-solr</module>
<module>external/storm-metrics</module>