You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@storm.apache.org by ka...@apache.org on 2018/07/19 03:35:01 UTC
[5/7] storm git commit: STORM-2953: Remove storm-kafka
http://git-wip-us.apache.org/repos/asf/storm/blob/e58ac3e0/external/storm-kafka/README.md
----------------------------------------------------------------------
diff --git a/external/storm-kafka/README.md b/external/storm-kafka/README.md
deleted file mode 100644
index d25bf29..0000000
--- a/external/storm-kafka/README.md
+++ /dev/null
@@ -1,382 +0,0 @@
-Storm Kafka
-====================
-
-Provides core Storm and Trident spout implementations for consuming data from Apache Kafka 0.8.x.
-
-## Spouts
-We support both Trident and core Storm spouts. For both spout implementations, we use a BrokerHost interface that
-tracks Kafka broker host to partition mapping and kafkaConfig that controls some Kafka related parameters.
-
-### BrokerHosts
-In order to initialize your Kafka spout/emitter you need to construct an instance of the marker interface BrokerHosts.
-Currently, we support the following two implementations:
-
-#### ZkHosts
-ZkHosts is what you should use if you want to dynamically track Kafka broker to partition mapping. This class uses
-Kafka's ZooKeeper entries to track brokerHost -> partition mapping. You can instantiate an object by calling
-```java
- public ZkHosts(String brokerZkStr, String brokerZkPath)
- public ZkHosts(String brokerZkStr)
-```
-Where brokerZkStr is just ip:port (e.g. localhost:2181). brokerZkPath is the root directory under which all the topics and
-partition information is stored. By default this is /brokers which is what the default Kafka implementation uses.
-
-By default, the broker-partition mapping is refreshed every 60 seconds from ZooKeeper. If you want to change it, you
-should set host.refreshFreqSecs to your chosen value.
-
-#### StaticHosts
-This is an alternative implementation where broker -> partition information is static. In order to construct an instance
-of this class, you need to first construct an instance of GlobalPartitionInformation.
-
-```java
- Broker brokerForPartition0 = new Broker("localhost");//localhost:9092
- Broker brokerForPartition1 = new Broker("localhost", 9092);//localhost:9092 but we specified the port explicitly
- Broker brokerForPartition2 = new Broker("localhost:9092");//localhost:9092 specified as one string.
- GlobalPartitionInformation partitionInfo = new GlobalPartitionInformation();
- partitionInfo.addPartition(0, brokerForPartition0);//mapping from partition 0 to brokerForPartition0
- partitionInfo.addPartition(1, brokerForPartition1);//mapping from partition 1 to brokerForPartition1
- partitionInfo.addPartition(2, brokerForPartition2);//mapping from partition 2 to brokerForPartition2
- StaticHosts hosts = new StaticHosts(partitionInfo);
-```
-
-### KafkaConfig
-The second thing needed for constructing a kafkaSpout is an instance of KafkaConfig.
-```java
- public KafkaConfig(BrokerHosts hosts, String topic)
- public KafkaConfig(BrokerHosts hosts, String topic, String clientId)
-```
-
-The BrokerHosts can be any implementation of BrokerHosts interface as described above. The topic is name of Kafka topic.
-The optional ClientId is used as a part of the ZooKeeper path where the spout's current consumption offset is stored.
-
-There are 2 extensions of KafkaConfig currently in use.
-
-Spoutconfig is an extension of KafkaConfig that supports additional fields with ZooKeeper connection info and for controlling
-behavior specific to KafkaSpout. The Zkroot will be used as root to store your consumer's offset. The id should uniquely
-identify your spout.
-```java
-public SpoutConfig(BrokerHosts hosts, String topic, String zkRoot, String id);
-```
-In addition to these parameters, SpoutConfig contains the following fields that control how KafkaSpout behaves:
-```java
- // setting for how often to save the current Kafka offset to ZooKeeper
- public long stateUpdateIntervalMs = 2000;
-
- // Retry strategy for failed messages
- public String failedMsgRetryManagerClass = ExponentialBackoffMsgRetryManager.class.getName();
-
- // Exponential back-off retry settings. These are used by ExponentialBackoffMsgRetryManager for retrying messages after a bolt
- // calls OutputCollector.fail(). These come into effect only if ExponentialBackoffMsgRetryManager is being used.
- // Initial delay between successive retries
- public long retryInitialDelayMs = 0;
- public double retryDelayMultiplier = 1.0;
-
- // Maximum delay between successive retries
- public long retryDelayMaxMs = 60 * 1000;
- // Failed message will be retried infinitely if retryLimit is less than zero.
- public int retryLimit = -1;
-
-```
-Core KafkaSpout only accepts an instance of SpoutConfig.
-
-TridentKafkaConfig is another extension of KafkaConfig.
-TridentKafkaEmitter only accepts TridentKafkaConfig.
-
-The KafkaConfig class also has bunch of public variables that controls your application's behavior. Here are defaults:
-```java
- public int fetchSizeBytes = 1024 * 1024;
- public int socketTimeoutMs = 10000;
- public int fetchMaxWait = 10000;
- public int bufferSizeBytes = 1024 * 1024;
- public MultiScheme scheme = new RawMultiScheme();
- public boolean ignoreZkOffsets = false;
- public long startOffsetTime = kafka.api.OffsetRequest.EarliestTime();
- public long maxOffsetBehind = Long.MAX_VALUE;
- public boolean useStartOffsetTimeIfOffsetOutOfRange = true;
- public int metricsTimeBucketSizeInSecs = 60;
-```
-
-Most of them are self explanatory except MultiScheme.
-### MultiScheme
-MultiScheme is an interface that dictates how the ByteBuffer consumed from Kafka gets transformed into a storm tuple. It
-also controls the naming of your output field.
-
-```java
- public Iterable<List<Object>> deserialize(ByteBuffer ser);
- public Fields getOutputFields();
-```
-
-The default `RawMultiScheme` just takes the `ByteBuffer` and returns a tuple with the ByteBuffer converted to a `byte[]`. The name of the outputField is "bytes". There are alternative implementations like `SchemeAsMultiScheme` and `KeyValueSchemeAsMultiScheme` which can convert the `ByteBuffer` to `String`.
-
-There is also an extension of `SchemeAsMultiScheme`, `MessageMetadataSchemeAsMultiScheme`,
-which has an additional deserialize method that accepts the message `ByteBuffer` in addition to the `Partition` and `offset` associated with the message.
-
-```java
-public Iterable<List<Object>> deserializeMessageWithMetadata(ByteBuffer message, Partition partition, long offset)
-
-```
-
-This is useful for auditing/replaying messages from arbitrary points on a Kafka topic, saving the partition and offset of each message of a discrete stream instead of persisting the entire message.
-
-### Failed message retry
-FailedMsgRetryManager is an interface which defines the retry strategy for a failed message. Default implementation is ExponentialBackoffMsgRetryManager which retries with exponential delays
-between consecutive retries. To use a custom implementation, set SpoutConfig.failedMsgRetryManagerClass to the full classname
-of implementation. Here is the interface
-
-```java
- // Spout initialization can go here. This can be called multiple times during lifecycle of a worker.
- void prepare(SpoutConfig spoutConfig, Map stormConf);
-
- // Message corresponding to offset has failed. This method is called only if retryFurther returns true for offset.
- void failed(Long offset);
-
- // Message corresponding to offset has been acked.
- void acked(Long offset);
-
- // Message corresponding to the offset, has been re-emitted and under transit.
- void retryStarted(Long offset);
-
- /**
- * The offset of message, which is to be re-emitted. Spout will fetch messages starting from this offset
- * and resend them, except completed messages.
- */
- Long nextFailedMessageToRetry();
-
- /**
- * @return True if the message corresponding to the offset should be emitted NOW. False otherwise.
- */
- boolean shouldReEmitMsg(Long offset);
-
- /**
- * Spout will clean up the state for this offset if false is returned. If retryFurther is set to true,
- * spout will called failed(offset) in next call and acked(offset) otherwise
- */
- boolean retryFurther(Long offset);
-
- /**
- * Spout will call this method after retryFurther returns false.
- * This gives a chance for hooking up custom logic before all clean up.
- * @param partition,offset
- */
- void cleanOffsetAfterRetries(Partition partition, Long offset);
-
- /**
- * Clear any offsets before kafkaOffset. These offsets are no longer available in kafka.
- */
- Set<Long> clearOffsetsBefore(Long kafkaOffset);
-```
-
-#### Version incompatibility
-In Storm versions prior to 1.0, the MultiScheme methods accepted a `byte[]` instead of `ByteBuffer`. The `MultScheme` and the related
-Scheme apis were changed in version 1.0 to accept a ByteBuffer instead of a byte[].
-
-This means that pre 1.0 kafka spouts will not work with Storm versions 1.0 and higher. While running topologies in Storm version 1.0
-and higher, it must be ensured that the storm-kafka version is at least 1.0. Pre 1.0 shaded topology jars that bundles
-storm-kafka classes must be rebuilt with storm-kafka version 1.0 for running in clusters with storm 1.0 and higher.
-
-### Examples
-
-#### Core Spout
-
-```java
-BrokerHosts hosts = new ZkHosts(zkConnString);
-SpoutConfig spoutConfig = new SpoutConfig(hosts, topicName, "/" + topicName, UUID.randomUUID().toString());
-spoutConfig.scheme = new SchemeAsMultiScheme(new StringScheme());
-KafkaSpout kafkaSpout = new KafkaSpout(spoutConfig);
-```
-
-#### Trident Spout
-```java
-TridentTopology topology = new TridentTopology();
-BrokerHosts zk = new ZkHosts("localhost");
-TridentKafkaConfig spoutConf = new TridentKafkaConfig(zk, "test-topic");
-spoutConf.scheme = new SchemeAsMultiScheme(new StringScheme());
-OpaqueTridentKafkaSpout spout = new OpaqueTridentKafkaSpout(spoutConf);
-```
-
-
-### How KafkaSpout stores offsets of a Kafka topic and recovers in case of failures
-
-As shown in the above KafkaConfig properties, you can control from where in the Kafka topic the spout begins to read by
-setting `KafkaConfig.startOffsetTime` as follows:
-
-1. `kafka.api.OffsetRequest.EarliestTime()`: read from the beginning of the topic (i.e. from the oldest messages onwards)
-2. `kafka.api.OffsetRequest.LatestTime()`: read from the end of the topic (i.e. any new messsages that are being written to the topic)
-3. A Unix timestamp aka seconds since the epoch (e.g. via `System.currentTimeMillis()`):
- see [How do I accurately get offsets of messages for a certain timestamp using OffsetRequest?](https://cwiki.apache.org/confluence/display/KAFKA/FAQ#FAQ-HowdoIaccuratelygetoffsetsofmessagesforacertaintimestampusingOffsetRequest?) in the Kafka FAQ
-
-As the topology runs the Kafka spout keeps track of the offsets it has read and emitted by storing state information
-under the ZooKeeper path `SpoutConfig.zkRoot+ "/" + SpoutConfig.id`. In the case of failures it recovers from the last
-written offset in ZooKeeper.
-
-> **Important:** When re-deploying a topology make sure that the settings for `SpoutConfig.zkRoot` and `SpoutConfig.id`
-> were not modified, otherwise the spout will not be able to read its previous consumer state information (i.e. the
-> offsets) from ZooKeeper -- which may lead to unexpected behavior and/or to data loss, depending on your use case.
-
-This means that when a topology has run once the setting `KafkaConfig.startOffsetTime` will not have an effect for
-subsequent runs of the topology because now the topology will rely on the consumer state information (offsets) in
-ZooKeeper to determine from where it should begin (more precisely: resume) reading.
-If you want to force the spout to ignore any consumer state information stored in ZooKeeper, then you should
-set the parameter `KafkaConfig.ignoreZkOffsets` to `true`. If `true`, the spout will always begin reading from the
-offset defined by `KafkaConfig.startOffsetTime` as described above.
-
-
-## Using storm-kafka with different versions of kafka
-
-Storm-kafka's Kafka dependency is defined as `provided` scope in maven, meaning it will not be pulled in
-as a transitive dependency. This allows you to use a version of Kafka dependency compatible with your kafka cluster.
-
-When building a project with storm-kafka, you must explicitly add the Kafka dependency. For example, to
-use Kafka 0.8.1.1 built against Scala 2.10, you would use the following dependency in your `pom.xml`:
-
-```xml
- <dependency>
- <groupId>org.apache.kafka</groupId>
- <artifactId>kafka_2.10</artifactId>
- <version>0.8.1.1</version>
- <exclusions>
- <exclusion>
- <groupId>org.apache.zookeeper</groupId>
- <artifactId>zookeeper</artifactId>
- </exclusion>
- <exclusion>
- <groupId>log4j</groupId>
- <artifactId>log4j</artifactId>
- </exclusion>
- </exclusions>
- </dependency>
-```
-
-Note that the ZooKeeper and log4j dependencies are excluded to prevent version conflicts with Storm's dependencies.
-
-You can also override the kafka dependency version while building from maven, with parameter `kafka.version` and `kafka.artifact.id`
-e.g. `mvn clean install -Dkafka.artifact.id=kafka_2.11 -Dkafka.version=0.9.0.1`
-
-When selecting a kafka dependency version, you should ensure -
- 1. kafka api is compatible with storm-kafka. Currently, only 0.9.x and 0.8.x client API is supported by storm-kafka
- module. If you want to use a higher version, storm-kafka-client module should be used instead.
- 2. The kafka client selected by you should be wire compatible with the broker. e.g. 0.9.x client will not work with
- 0.8.x broker.
-
-
-## Writing to Kafka as part of your topology
-You can create an instance of org.apache.storm.kafka.bolt.KafkaBolt and attach it as a component to your topology or if you
-are using trident you can use org.apache.storm.kafka.trident.TridentState, org.apache.storm.kafka.trident.TridentStateFactory and
-org.apache.storm.kafka.trident.TridentKafkaUpdater.
-
-You need to provide implementation of following 2 interfaces
-
-### TupleToKafkaMapper and TridentTupleToKafkaMapper
-These interfaces have 2 methods defined:
-
-```java
- K getKeyFromTuple(Tuple/TridentTuple tuple);
- V getMessageFromTuple(Tuple/TridentTuple tuple);
-```
-
-As the name suggests, these methods are called to map a tuple to Kafka key and Kafka message. If you just want one field
-as key and one field as value, then you can use the provided FieldNameBasedTupleToKafkaMapper.java
-implementation. In the KafkaBolt, the implementation always looks for a field with field name "key" and "message" if you
-use the default constructor to construct FieldNameBasedTupleToKafkaMapper for backward compatibility
-reasons. Alternatively you could also specify a different key and message field by using the non default constructor.
-In the TridentKafkaState you must specify what is the field name for key and message as there is no default constructor.
-These should be specified while constructing and instance of FieldNameBasedTupleToKafkaMapper.
-
-### KafkaTopicSelector and trident KafkaTopicSelector
-This interface has only one method
-```java
-public interface KafkaTopicSelector {
- String getTopics(Tuple/TridentTuple tuple);
-}
-```
-The implementation of this interface should return the topic to which the tuple's key/message mapping needs to be published
-You can return a null and the message will be ignored. If you have one static topic name then you can use
-DefaultTopicSelector.java and set the name of the topic in the constructor.
-`FieldNameTopicSelector` and `FieldIndexTopicSelector` use to support decided which topic should to push message from tuple.
-User could specify the field name or field index in tuple ,selector will use this value as topic name which to publish message.
-When the topic name not found , `KafkaBolt` will write messages into default topic .
-Please make sure the default topic have created .
-
-### Specifying Kafka producer properties
-You can provide all the produce properties in your Storm topology by calling `KafkaBolt.withProducerProperties()` and `TridentKafkaStateFactory.withProducerProperties()`. Please see http://kafka.apache.org/documentation.html#newproducerconfigs
-Section "Important configuration properties for the producer" for more details.
-
-### Using wildcard kafka topic match
-You can do a wildcard topic match by adding the following config
-```
- Config config = new Config();
- config.put("kafka.topic.wildcard.match",true);
-
-```
-
-After this you can specify a wildcard topic for matching e.g. clickstream.*.log. This will match all streams matching clickstream.my.log, clickstream.cart.log etc
-
-
-### Putting it all together
-
-For the bolt :
-```java
- TopologyBuilder builder = new TopologyBuilder();
-
- Fields fields = new Fields("key", "message");
- FixedBatchSpout spout = new FixedBatchSpout(fields, 4,
- new Values("storm", "1"),
- new Values("trident", "1"),
- new Values("needs", "1"),
- new Values("javadoc", "1")
- );
- spout.setCycle(true);
- builder.setSpout("spout", spout, 5);
- //set producer properties.
- Properties props = new Properties();
- props.put("bootstrap.servers", "localhost:9092");
- props.put("acks", "1");
- props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
- props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
-
- KafkaBolt bolt = new KafkaBolt()
- .withProducerProperties(props)
- .withTopicSelector(new DefaultTopicSelector("test"))
- .withTupleToKafkaMapper(new FieldNameBasedTupleToKafkaMapper());
- builder.setBolt("forwardToKafka", bolt, 8).shuffleGrouping("spout");
-
- Config conf = new Config();
-
- StormSubmitter.submitTopology("kafkaboltTest", conf, builder.createTopology());
-```
-
-For Trident:
-
-```java
- Fields fields = new Fields("word", "count");
- FixedBatchSpout spout = new FixedBatchSpout(fields, 4,
- new Values("storm", "1"),
- new Values("trident", "1"),
- new Values("needs", "1"),
- new Values("javadoc", "1")
- );
- spout.setCycle(true);
-
- TridentTopology topology = new TridentTopology();
- Stream stream = topology.newStream("spout1", spout);
-
- //set producer properties.
- Properties props = new Properties();
- props.put("bootstrap.servers", "localhost:9092");
- props.put("acks", "1");
- props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
- props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
-
- TridentKafkaStateFactory stateFactory = new TridentKafkaStateFactory()
- .withProducerProperties(props)
- .withKafkaTopicSelector(new DefaultTopicSelector("test"))
- .withTridentTupleToKafkaMapper(new FieldNameBasedTupleToKafkaMapper("word", "count"));
- stream.partitionPersist(stateFactory, fields, new TridentKafkaUpdater(), new Fields());
-
- Config conf = new Config();
- StormSubmitter.submitTopology("kafkaTridentTest", conf, topology.build());
-```
-
-## Committer Sponsors
-
- * P. Taylor Goetz ([ptgoetz@apache.org](mailto:ptgoetz@apache.org))
http://git-wip-us.apache.org/repos/asf/storm/blob/e58ac3e0/external/storm-kafka/pom.xml
----------------------------------------------------------------------
diff --git a/external/storm-kafka/pom.xml b/external/storm-kafka/pom.xml
deleted file mode 100644
index 2e0f35b..0000000
--- a/external/storm-kafka/pom.xml
+++ /dev/null
@@ -1,125 +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.0.0-SNAPSHOT</version>
- <relativePath>../../pom.xml</relativePath>
- </parent>
-
- <packaging>jar</packaging>
- <artifactId>storm-kafka</artifactId>
- <name>storm-kafka</name>
- <description>Storm Spouts for Apache Kafka</description>
-
- <build>
- <plugins>
- <plugin>
- <groupId>org.jacoco</groupId>
- <artifactId>jacoco-maven-plugin</artifactId>
- <version>0.7.2.201409121644</version>
- <executions>
- <execution>
- <id>jacoco-initialize</id>
- <goals>
- <goal>prepare-agent</goal>
- </goals>
- </execution>
- <execution>
- <id>jacoco-report</id>
- <phase>test</phase>
- <goals>
- <goal>report</goal>
- </goals>
- </execution>
- </executions>
- </plugin>
- <plugin>
- <groupId>org.apache.maven.plugins</groupId>
- <artifactId>maven-checkstyle-plugin</artifactId>
- <!--Note - the version would be inherited-->
- <configuration>
- <maxAllowedViolations>180</maxAllowedViolations>
- </configuration>
- </plugin>
- </plugins>
- <sourceDirectory>src/jvm</sourceDirectory>
- <testSourceDirectory>src/test</testSourceDirectory>
- </build>
- <dependencies>
- <dependency>
- <groupId>commons-io</groupId>
- <artifactId>commons-io</artifactId>
- <scope>compile</scope>
- </dependency>
- <dependency>
- <groupId>org.mockito</groupId>
- <artifactId>mockito-core</artifactId>
- </dependency>
- <dependency>
- <groupId>org.hamcrest</groupId>
- <artifactId>java-hamcrest</artifactId>
- </dependency>
- <dependency>
- <groupId>org.apache.curator</groupId>
- <artifactId>curator-framework</artifactId>
- </dependency>
- <dependency>
- <groupId>org.apache.zookeeper</groupId>
- <artifactId>zookeeper</artifactId>
- </dependency>
- <dependency>
- <groupId>org.apache.curator</groupId>
- <artifactId>curator-recipes</artifactId>
- <scope>test</scope>
- </dependency>
- <dependency>
- <groupId>org.apache.curator</groupId>
- <artifactId>curator-test</artifactId>
- <scope>test</scope>
- </dependency>
- <dependency>
- <groupId>org.apache.kafka</groupId>
- <artifactId>${storm.kafka.artifact.id}</artifactId>
- <version>${storm.kafka.version}</version>
- <scope>provided</scope>
- </dependency>
- <dependency>
- <groupId>org.apache.kafka</groupId>
- <artifactId>kafka-clients</artifactId>
- <version>${storm.kafka.version}</version>
- </dependency>
- <dependency>
- <groupId>org.apache.storm</groupId>
- <artifactId>storm-client</artifactId>
- <version>${project.version}</version>
- <scope>${provided.scope}</scope>
- </dependency>
- <dependency>
- <groupId>commons-lang</groupId>
- <artifactId>commons-lang</artifactId>
- </dependency>
- <dependency>
- <groupId>com.googlecode.json-simple</groupId>
- <artifactId>json-simple</artifactId>
- </dependency>
- </dependencies>
-</project>
http://git-wip-us.apache.org/repos/asf/storm/blob/e58ac3e0/external/storm-kafka/src/jvm/org/apache/storm/kafka/Broker.java
----------------------------------------------------------------------
diff --git a/external/storm-kafka/src/jvm/org/apache/storm/kafka/Broker.java b/external/storm-kafka/src/jvm/org/apache/storm/kafka/Broker.java
deleted file mode 100644
index b33af99..0000000
--- a/external/storm-kafka/src/jvm/org/apache/storm/kafka/Broker.java
+++ /dev/null
@@ -1,79 +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.kafka;
-
-import com.google.common.base.Objects;
-import java.io.Serializable;
-
-public class Broker implements Serializable, Comparable<Broker> {
- public String host;
- public int port;
-
- // for kryo compatibility
- private Broker() {
-
- }
-
- public Broker(String host, int port) {
- this.host = host;
- this.port = port;
- }
-
- public Broker(String host) {
- this(host, 9092);
- }
-
- public static Broker fromString(String host) {
- Broker hp;
- String[] spec = host.split(":");
- if (spec.length == 1) {
- hp = new Broker(spec[0]);
- } else if (spec.length == 2) {
- hp = new Broker(spec[0], Integer.parseInt(spec[1]));
- } else {
- throw new IllegalArgumentException("Invalid host specification: " + host);
- }
- return hp;
- }
-
- @Override
- public int hashCode() {
- return Objects.hashCode(host, port);
- }
-
- @Override
- public boolean equals(Object obj) {
- if (this == obj) {
- return true;
- }
- if (obj == null || getClass() != obj.getClass()) {
- return false;
- }
- final Broker other = (Broker) obj;
- return Objects.equal(this.host, other.host) && Objects.equal(this.port, other.port);
- }
-
- @Override
- public String toString() {
- return host + ":" + port;
- }
-
- @Override
- public int compareTo(Broker o) {
- if (this.host.equals(o.host)) {
- return this.port - o.port;
- } else {
- return this.host.compareTo(o.host);
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/storm/blob/e58ac3e0/external/storm-kafka/src/jvm/org/apache/storm/kafka/BrokerHosts.java
----------------------------------------------------------------------
diff --git a/external/storm-kafka/src/jvm/org/apache/storm/kafka/BrokerHosts.java b/external/storm-kafka/src/jvm/org/apache/storm/kafka/BrokerHosts.java
deleted file mode 100644
index dbd6a10..0000000
--- a/external/storm-kafka/src/jvm/org/apache/storm/kafka/BrokerHosts.java
+++ /dev/null
@@ -1,20 +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.kafka;
-
-import java.io.Serializable;
-
-
-public interface BrokerHosts extends Serializable {
-
-}
http://git-wip-us.apache.org/repos/asf/storm/blob/e58ac3e0/external/storm-kafka/src/jvm/org/apache/storm/kafka/ByteBufferSerializer.java
----------------------------------------------------------------------
diff --git a/external/storm-kafka/src/jvm/org/apache/storm/kafka/ByteBufferSerializer.java b/external/storm-kafka/src/jvm/org/apache/storm/kafka/ByteBufferSerializer.java
deleted file mode 100644
index 37986a5..0000000
--- a/external/storm-kafka/src/jvm/org/apache/storm/kafka/ByteBufferSerializer.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
- * <p>
- * http://www.apache.org/licenses/LICENSE-2.0
- * <p>
- * 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.kafka;
-
-import java.nio.ByteBuffer;
-import java.util.Map;
-import org.apache.kafka.common.serialization.Serializer;
-import org.apache.storm.utils.Utils;
-
-public class ByteBufferSerializer implements Serializer<ByteBuffer> {
- @Override
- public void configure(Map<String, ?> map, boolean b) {
-
- }
-
- @Override
- public void close() {
-
- }
-
- @Override
- public byte[] serialize(String s, ByteBuffer b) {
- return Utils.toByteArray(b);
- }
-}
http://git-wip-us.apache.org/repos/asf/storm/blob/e58ac3e0/external/storm-kafka/src/jvm/org/apache/storm/kafka/DynamicBrokersReader.java
----------------------------------------------------------------------
diff --git a/external/storm-kafka/src/jvm/org/apache/storm/kafka/DynamicBrokersReader.java b/external/storm-kafka/src/jvm/org/apache/storm/kafka/DynamicBrokersReader.java
deleted file mode 100644
index 49ad530..0000000
--- a/external/storm-kafka/src/jvm/org/apache/storm/kafka/DynamicBrokersReader.java
+++ /dev/null
@@ -1,208 +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.kafka;
-
-import com.google.common.base.Preconditions;
-import java.net.SocketTimeoutException;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Map;
-import org.apache.curator.framework.CuratorFramework;
-import org.apache.curator.framework.CuratorFrameworkFactory;
-import org.apache.curator.retry.RetryNTimes;
-import org.apache.storm.Config;
-import org.apache.storm.kafka.trident.GlobalPartitionInformation;
-import org.apache.storm.utils.ObjectReader;
-import org.json.simple.JSONValue;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-
-public class DynamicBrokersReader {
-
- private static final Logger LOG = LoggerFactory.getLogger(DynamicBrokersReader.class);
-
- private CuratorFramework _curator;
- private String _zkPath;
- private String _topic;
- private Boolean _isWildcardTopic;
-
- public DynamicBrokersReader(Map<String, Object> conf, String zkStr, String zkPath, String topic) {
- // Check required parameters
- Preconditions.checkNotNull(conf, "conf cannot be null");
-
- validateConfig(conf);
-
- Preconditions.checkNotNull(zkStr, "zkString cannot be null");
- Preconditions.checkNotNull(zkPath, "zkPath cannot be null");
- Preconditions.checkNotNull(topic, "topic cannot be null");
-
- _zkPath = zkPath;
- _topic = topic;
- _isWildcardTopic = ObjectReader.getBoolean(conf.get("kafka.topic.wildcard.match"), false);
- try {
- _curator = CuratorFrameworkFactory.newClient(
- zkStr,
- ObjectReader.getInt(conf.get(Config.STORM_ZOOKEEPER_SESSION_TIMEOUT)),
- ObjectReader.getInt(conf.get(Config.STORM_ZOOKEEPER_CONNECTION_TIMEOUT)),
- new RetryNTimes(ObjectReader.getInt(conf.get(Config.STORM_ZOOKEEPER_RETRY_TIMES)),
- ObjectReader.getInt(conf.get(Config.STORM_ZOOKEEPER_RETRY_INTERVAL))));
- _curator.start();
- } catch (Exception ex) {
- LOG.error("Couldn't connect to zookeeper", ex);
- throw new RuntimeException(ex);
- }
- }
-
- /**
- * Get all partitions with their current leaders
- */
- public List<GlobalPartitionInformation> getBrokerInfo() throws SocketTimeoutException {
- List<String> topics = getTopics();
- List<GlobalPartitionInformation> partitions = new ArrayList<GlobalPartitionInformation>();
-
- for (String topic : topics) {
- GlobalPartitionInformation globalPartitionInformation = new GlobalPartitionInformation(topic, this._isWildcardTopic);
- try {
- int numPartitionsForTopic = getNumPartitions(topic);
- String brokerInfoPath = brokerPath();
- for (int partition = 0; partition < numPartitionsForTopic; partition++) {
- int leader = getLeaderFor(topic, partition);
- String path = brokerInfoPath + "/" + leader;
- try {
- byte[] brokerData = _curator.getData().forPath(path);
- Broker hp = getBrokerHost(brokerData);
- globalPartitionInformation.addPartition(partition, hp);
- } catch (org.apache.zookeeper.KeeperException.NoNodeException e) {
- LOG.error("Node {} does not exist ", path);
- }
- }
- } catch (SocketTimeoutException e) {
- throw e;
- } catch (Exception e) {
- throw new RuntimeException(e);
- }
- LOG.info("Read partition info from zookeeper: " + globalPartitionInformation);
- partitions.add(globalPartitionInformation);
- }
- return partitions;
- }
-
- private int getNumPartitions(String topic) {
- try {
- String topicBrokersPath = partitionPath(topic);
- List<String> children = _curator.getChildren().forPath(topicBrokersPath);
- return children.size();
- } catch (Exception e) {
- throw new RuntimeException(e);
- }
- }
-
- private List<String> getTopics() {
- List<String> topics = new ArrayList<String>();
- if (!_isWildcardTopic) {
- topics.add(_topic);
- return topics;
- } else {
- try {
- List<String> children = _curator.getChildren().forPath(topicsPath());
- for (String t : children) {
- if (t.matches(_topic)) {
- LOG.info(String.format("Found matching topic %s", t));
- topics.add(t);
- }
- }
- return topics;
- } catch (Exception e) {
- throw new RuntimeException(e);
- }
- }
- }
-
- public String topicsPath() {
- return _zkPath + "/topics";
- }
-
- public String partitionPath(String topic) {
- return topicsPath() + "/" + topic + "/partitions";
- }
-
- public String brokerPath() {
- return _zkPath + "/ids";
- }
-
-
- /**
- * get /brokers/topics/distributedTopic/partitions/1/state
- * { "controller_epoch":4, "isr":[ 1, 0 ], "leader":1, "leader_epoch":1, "version":1 }
- * @param topic
- * @param partition
- * @return
- */
- private int getLeaderFor(String topic, long partition) {
- try {
- String topicBrokersPath = partitionPath(topic);
- byte[] hostPortData = _curator.getData().forPath(topicBrokersPath + "/" + partition + "/state");
- Map<Object, Object> value = (Map<Object, Object>) JSONValue.parseWithException(new String(hostPortData, "UTF-8"));
- Integer leader = ((Number) value.get("leader")).intValue();
- if (leader == -1) {
- throw new RuntimeException("No leader found for partition " + partition);
- }
- return leader;
- } catch (RuntimeException e) {
- throw e;
- } catch (Exception e) {
- throw new RuntimeException(e);
- }
- }
-
- public void close() {
- _curator.close();
- }
-
- /**
- * [zk: localhost:2181(CONNECTED) 56] get /brokers/ids/0
- * { "host":"localhost", "jmx_port":9999, "port":9092, "version":1 }
- *
- * @param contents
- * @return
- */
- private Broker getBrokerHost(byte[] contents) {
- try {
- Map<Object, Object> value = (Map<Object, Object>) JSONValue.parseWithException(new String(contents, "UTF-8"));
- String host = (String) value.get("host");
- Integer port = ((Long) value.get("port")).intValue();
- return new Broker(host, port);
- } catch (RuntimeException e) {
- throw e;
- } catch (Exception e) {
- throw new RuntimeException(e);
- }
- }
-
- /**
- * Validate required parameters in the input configuration Map
- * @param conf
- */
- private void validateConfig(final Map<String, Object> conf) {
- Preconditions.checkNotNull(conf.get(Config.STORM_ZOOKEEPER_SESSION_TIMEOUT),
- "%s cannot be null", Config.STORM_ZOOKEEPER_SESSION_TIMEOUT);
- Preconditions.checkNotNull(conf.get(Config.STORM_ZOOKEEPER_CONNECTION_TIMEOUT),
- "%s cannot be null", Config.STORM_ZOOKEEPER_CONNECTION_TIMEOUT);
- Preconditions.checkNotNull(conf.get(Config.STORM_ZOOKEEPER_RETRY_TIMES),
- "%s cannot be null", Config.STORM_ZOOKEEPER_RETRY_TIMES);
- Preconditions.checkNotNull(conf.get(Config.STORM_ZOOKEEPER_RETRY_INTERVAL),
- "%s cannot be null", Config.STORM_ZOOKEEPER_RETRY_INTERVAL);
- }
-
-}
http://git-wip-us.apache.org/repos/asf/storm/blob/e58ac3e0/external/storm-kafka/src/jvm/org/apache/storm/kafka/DynamicPartitionConnections.java
----------------------------------------------------------------------
diff --git a/external/storm-kafka/src/jvm/org/apache/storm/kafka/DynamicPartitionConnections.java b/external/storm-kafka/src/jvm/org/apache/storm/kafka/DynamicPartitionConnections.java
deleted file mode 100644
index 1ca7144..0000000
--- a/external/storm-kafka/src/jvm/org/apache/storm/kafka/DynamicPartitionConnections.java
+++ /dev/null
@@ -1,91 +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.kafka;
-
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.Map;
-import java.util.Set;
-import kafka.javaapi.consumer.SimpleConsumer;
-import org.apache.storm.kafka.trident.IBrokerReader;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-
-public class DynamicPartitionConnections {
-
- private static final Logger LOG = LoggerFactory.getLogger(DynamicPartitionConnections.class);
- Map<Broker, ConnectionInfo> _connections = new HashMap<>();
- KafkaConfig _config;
- IBrokerReader _reader;
- public DynamicPartitionConnections(KafkaConfig config, IBrokerReader brokerReader) {
- _config = config;
- _reader = brokerReader;
- }
-
- public SimpleConsumer register(Partition partition) {
- Broker broker = _reader.getBrokerForTopic(partition.topic).getBrokerFor(partition.partition);
- return register(broker, partition.topic, partition.partition);
- }
-
- public SimpleConsumer register(Broker host, String topic, int partition) {
- if (!_connections.containsKey(host)) {
- _connections.put(host, new ConnectionInfo(
- new SimpleConsumer(host.host, host.port, _config.socketTimeoutMs, _config.bufferSizeBytes, _config.clientId)));
- }
- ConnectionInfo info = _connections.get(host);
- info.partitions.add(getHashKey(topic, partition));
- return info.consumer;
- }
-
- public SimpleConsumer getConnection(Partition partition) {
- ConnectionInfo info = _connections.get(partition.host);
- if (info != null) {
- return info.consumer;
- }
- return null;
- }
-
- public void unregister(Broker port, String topic, int partition) {
- ConnectionInfo info = _connections.get(port);
- info.partitions.remove(getHashKey(topic, partition));
- if (info.partitions.isEmpty()) {
- info.consumer.close();
- _connections.remove(port);
- }
- }
-
- public void unregister(Partition partition) {
- unregister(partition.host, partition.topic, partition.partition);
- }
-
- public void clear() {
- for (ConnectionInfo info : _connections.values()) {
- info.consumer.close();
- }
- _connections.clear();
- }
-
- private String getHashKey(String topic, int partition) {
- return topic + "_" + partition;
- }
-
- static class ConnectionInfo {
- SimpleConsumer consumer;
- Set<String> partitions = new HashSet<String>();
-
- public ConnectionInfo(SimpleConsumer consumer) {
- this.consumer = consumer;
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/storm/blob/e58ac3e0/external/storm-kafka/src/jvm/org/apache/storm/kafka/ExponentialBackoffMsgRetryManager.java
----------------------------------------------------------------------
diff --git a/external/storm-kafka/src/jvm/org/apache/storm/kafka/ExponentialBackoffMsgRetryManager.java b/external/storm-kafka/src/jvm/org/apache/storm/kafka/ExponentialBackoffMsgRetryManager.java
deleted file mode 100644
index 2651c30..0000000
--- a/external/storm-kafka/src/jvm/org/apache/storm/kafka/ExponentialBackoffMsgRetryManager.java
+++ /dev/null
@@ -1,201 +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.kafka;
-
-import java.util.Comparator;
-import java.util.HashSet;
-import java.util.Map;
-import java.util.PriorityQueue;
-import java.util.Queue;
-import java.util.Set;
-import java.util.concurrent.ConcurrentHashMap;
-import org.apache.storm.utils.Time;
-
-public class ExponentialBackoffMsgRetryManager implements FailedMsgRetryManager {
-
- private long retryInitialDelayMs;
- private double retryDelayMultiplier;
- private long retryDelayMaxMs;
- private int retryLimit;
-
- private Queue<MessageRetryRecord> waiting;
- private Map<Long, MessageRetryRecord> records;
-
- public ExponentialBackoffMsgRetryManager() {
-
- }
-
- public void prepare(SpoutConfig spoutConfig, Map<String, Object> topoConf) {
- this.retryInitialDelayMs = spoutConfig.retryInitialDelayMs;
- this.retryDelayMultiplier = spoutConfig.retryDelayMultiplier;
- this.retryDelayMaxMs = spoutConfig.retryDelayMaxMs;
- this.retryLimit = spoutConfig.retryLimit;
- this.waiting = new PriorityQueue<MessageRetryRecord>(11, new RetryTimeComparator());
- this.records = new ConcurrentHashMap<Long, MessageRetryRecord>();
- }
-
- @Override
- public void failed(Long offset) {
- MessageRetryRecord oldRecord = this.records.get(offset);
- MessageRetryRecord newRecord = oldRecord == null ?
- new MessageRetryRecord(offset) :
- oldRecord.createNextRetryRecord();
- this.records.put(offset, newRecord);
- this.waiting.add(newRecord);
- }
-
- @Override
- public void acked(Long offset) {
- MessageRetryRecord record = this.records.remove(offset);
- if (record != null) {
- this.waiting.remove(record);
- }
- }
-
- @Override
- public void retryStarted(Long offset) {
- MessageRetryRecord record = this.records.get(offset);
- if (record == null || !this.waiting.contains(record)) {
- throw new IllegalStateException("cannot retry a message that has not failed");
- } else {
- this.waiting.remove(record);
- }
- }
-
- @Override
- public Long nextFailedMessageToRetry() {
- if (this.waiting.size() > 0) {
- MessageRetryRecord first = this.waiting.peek();
- if (Time.currentTimeMillis() >= first.retryTimeUTC) {
- if (this.records.containsKey(first.offset)) {
- return first.offset;
- } else {
- // defensive programming - should be impossible
- this.waiting.remove(first);
- return nextFailedMessageToRetry();
- }
- }
- }
- return null;
- }
-
- @Override
- public boolean shouldReEmitMsg(Long offset) {
- MessageRetryRecord record = this.records.get(offset);
- return record != null &&
- this.waiting.contains(record) &&
- Time.currentTimeMillis() >= record.retryTimeUTC;
- }
-
- @Override
- public boolean retryFurther(Long offset) {
- MessageRetryRecord record = this.records.get(offset);
- return !(record != null &&
- this.retryLimit > 0 &&
- this.retryLimit <= record.retryNum);
- }
-
- @Override
- public void cleanOffsetAfterRetries(Partition partition, Long offset) {
- //Do nothing..
- }
-
- @Override
- public Set<Long> clearOffsetsBefore(Long kafkaOffset) {
- Set<Long> invalidOffsets = new HashSet<Long>();
- for (Long offset : records.keySet()) {
- if (offset < kafkaOffset) {
- MessageRetryRecord record = this.records.remove(offset);
- if (record != null) {
- this.waiting.remove(record);
- invalidOffsets.add(offset);
- }
- }
- }
- return invalidOffsets;
- }
-
- private static class RetryTimeComparator implements Comparator<MessageRetryRecord> {
-
- @Override
- public int compare(MessageRetryRecord record1, MessageRetryRecord record2) {
- return Long.valueOf(record1.retryTimeUTC).compareTo(Long.valueOf(record2.retryTimeUTC));
- }
-
- @Override
- public boolean equals(Object obj) {
- return false;
- }
- }
-
- /**
- * A MessageRetryRecord holds the data of how many times a message has
- * failed and been retried, and when the last failure occurred. It can
- * determine whether it is ready to be retried by employing an exponential
- * back-off calculation using config values stored in SpoutConfig:
- * <ul>
- * <li>retryInitialDelayMs - time to delay before the first retry</li>
- * <li>retryDelayMultiplier - multiplier by which to increase the delay for each subsequent retry</li>
- * <li>retryDelayMaxMs - maximum retry delay (once this delay time is reached, subsequent retries will
- * delay for this amount of time every time)
- * </li>
- * </ul>
- */
- private class MessageRetryRecord {
- private final long offset;
- private final int retryNum;
- private final long retryTimeUTC;
-
- public MessageRetryRecord(long offset) {
- this(offset, 1);
- }
-
- private MessageRetryRecord(long offset, int retryNum) {
- this.offset = offset;
- this.retryNum = retryNum;
- this.retryTimeUTC = Time.currentTimeMillis() + calculateRetryDelay();
- }
-
- /**
- * Create a MessageRetryRecord for the next retry that should occur after this one.
- * @return MessageRetryRecord with the next retry time, or null to indicate that another
- * retry should not be performed. The latter case can happen if we are about to
- * run into the org.apache.storm.Config.TOPOLOGY_MESSAGE_TIMEOUT_SECS in the Storm
- * configuration.
- */
- public MessageRetryRecord createNextRetryRecord() {
- return new MessageRetryRecord(this.offset, this.retryNum + 1);
- }
-
- private long calculateRetryDelay() {
- double delayMultiplier = Math.pow(retryDelayMultiplier, this.retryNum - 1);
- double delay = retryInitialDelayMs * delayMultiplier;
- Long maxLong = Long.MAX_VALUE;
- long delayThisRetryMs = delay >= maxLong.doubleValue()
- ? maxLong
- : (long) delay;
- return Math.min(delayThisRetryMs, retryDelayMaxMs);
- }
-
- @Override
- public boolean equals(Object other) {
- return (other instanceof MessageRetryRecord
- && this.offset == ((MessageRetryRecord) other).offset);
- }
-
- @Override
- public int hashCode() {
- return Long.valueOf(this.offset).hashCode();
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/storm/blob/e58ac3e0/external/storm-kafka/src/jvm/org/apache/storm/kafka/FailedFetchException.java
----------------------------------------------------------------------
diff --git a/external/storm-kafka/src/jvm/org/apache/storm/kafka/FailedFetchException.java b/external/storm-kafka/src/jvm/org/apache/storm/kafka/FailedFetchException.java
deleted file mode 100644
index a1f3fe5..0000000
--- a/external/storm-kafka/src/jvm/org/apache/storm/kafka/FailedFetchException.java
+++ /dev/null
@@ -1,24 +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.kafka;
-
-public class FailedFetchException extends RuntimeException {
-
- public FailedFetchException(String message) {
- super(message);
- }
-
- public FailedFetchException(Exception e) {
- super(e);
- }
-}
http://git-wip-us.apache.org/repos/asf/storm/blob/e58ac3e0/external/storm-kafka/src/jvm/org/apache/storm/kafka/FailedMsgRetryManager.java
----------------------------------------------------------------------
diff --git a/external/storm-kafka/src/jvm/org/apache/storm/kafka/FailedMsgRetryManager.java b/external/storm-kafka/src/jvm/org/apache/storm/kafka/FailedMsgRetryManager.java
deleted file mode 100644
index b7fafdc..0000000
--- a/external/storm-kafka/src/jvm/org/apache/storm/kafka/FailedMsgRetryManager.java
+++ /dev/null
@@ -1,77 +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.kafka;
-
-import java.io.Serializable;
-import java.util.Map;
-import java.util.Set;
-
-public interface FailedMsgRetryManager extends Serializable {
-
- /**
- * Initialization
- */
- void prepare(SpoutConfig spoutConfig, Map<String, Object> topoConf);
-
- /**
- * Message corresponding to the offset failed in kafka spout.
- * @param offset
- */
- void failed(Long offset);
-
- /**
- * Message corresponding to the offset, was acked to kafka spout.
- * @param offset
- */
- void acked(Long offset);
-
- /**
- * Message corresponding to the offset, has been re-emitted and under transit.
- * @param offset
- */
- void retryStarted(Long offset);
-
- /**
- * The offset of message, which is to be re-emitted. Spout will fetch messages starting from this offset
- * and resend them, except completed messages.
- * @return
- */
- Long nextFailedMessageToRetry();
-
- /**
- * @param offset
- * @return True if the message corresponding to the offset should be emitted NOW. False otherwise.
- */
- boolean shouldReEmitMsg(Long offset);
-
- /**
- * Spout will clean up the state for this offset if false is returned.
- * @param offset
- * @return True if the message will be retried again. False otherwise.
- */
- boolean retryFurther(Long offset);
-
- /**
- * Spout will call this method after retryFurther returns false.
- * This gives a chance for hooking up custom logic before all clean up.
- * @param partition,offset
- */
- void cleanOffsetAfterRetries(Partition partition, Long offset);
-
- /**
- * Clear any offsets before kafkaOffset. These offsets are no longer available in kafka.
- * @param kafkaOffset
- * @return Set of offsets removed.
- */
- Set<Long> clearOffsetsBefore(Long kafkaOffset);
-}
http://git-wip-us.apache.org/repos/asf/storm/blob/e58ac3e0/external/storm-kafka/src/jvm/org/apache/storm/kafka/IntSerializer.java
----------------------------------------------------------------------
diff --git a/external/storm-kafka/src/jvm/org/apache/storm/kafka/IntSerializer.java b/external/storm-kafka/src/jvm/org/apache/storm/kafka/IntSerializer.java
deleted file mode 100644
index fef8625..0000000
--- a/external/storm-kafka/src/jvm/org/apache/storm/kafka/IntSerializer.java
+++ /dev/null
@@ -1,42 +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
- * <p>
- * http://www.apache.org/licenses/LICENSE-2.0
- * <p>
- * 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.kafka;
-
-import java.util.Map;
-import org.apache.kafka.common.serialization.Serializer;
-
-public class IntSerializer implements Serializer<Integer> {
- @Override
- public void configure(Map<String, ?> map, boolean b) {
- }
-
- @Override
- public byte[] serialize(String topic, Integer val) {
- return new byte[]{
- (byte) (val >>> 24),
- (byte) (val >>> 16),
- (byte) (val >>> 8),
- val.byteValue()
- };
- }
-
- @Override
- public void close() {
- }
-}
http://git-wip-us.apache.org/repos/asf/storm/blob/e58ac3e0/external/storm-kafka/src/jvm/org/apache/storm/kafka/KafkaConfig.java
----------------------------------------------------------------------
diff --git a/external/storm-kafka/src/jvm/org/apache/storm/kafka/KafkaConfig.java b/external/storm-kafka/src/jvm/org/apache/storm/kafka/KafkaConfig.java
deleted file mode 100644
index a93f426..0000000
--- a/external/storm-kafka/src/jvm/org/apache/storm/kafka/KafkaConfig.java
+++ /dev/null
@@ -1,49 +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.kafka;
-
-import java.io.Serializable;
-import kafka.api.FetchRequest;
-import org.apache.storm.spout.MultiScheme;
-import org.apache.storm.spout.RawMultiScheme;
-
-public class KafkaConfig implements Serializable {
- private static final long serialVersionUID = 5276718734571623855L;
-
- public final BrokerHosts hosts;
- public final String topic;
- public final String clientId;
-
- public int fetchSizeBytes = 1024 * 1024;
- public int socketTimeoutMs = 10000;
- public int fetchMaxWait = 100;
- public int bufferSizeBytes = 1024 * 1024;
- public MultiScheme scheme = new RawMultiScheme();
- public boolean ignoreZkOffsets = false;
- public long startOffsetTime = kafka.api.OffsetRequest.EarliestTime();
- public long maxOffsetBehind = Long.MAX_VALUE;
- public boolean useStartOffsetTimeIfOffsetOutOfRange = true;
- public int metricsTimeBucketSizeInSecs = 60;
- public int minFetchByte = FetchRequest.DefaultMinBytes();
-
- public KafkaConfig(BrokerHosts hosts, String topic) {
- this(hosts, topic, kafka.api.OffsetRequest.DefaultClientId());
- }
-
- public KafkaConfig(BrokerHosts hosts, String topic, String clientId) {
- this.hosts = hosts;
- this.topic = topic;
- this.clientId = clientId;
- }
-
-}
http://git-wip-us.apache.org/repos/asf/storm/blob/e58ac3e0/external/storm-kafka/src/jvm/org/apache/storm/kafka/KafkaError.java
----------------------------------------------------------------------
diff --git a/external/storm-kafka/src/jvm/org/apache/storm/kafka/KafkaError.java b/external/storm-kafka/src/jvm/org/apache/storm/kafka/KafkaError.java
deleted file mode 100644
index 4bf2ed2..0000000
--- a/external/storm-kafka/src/jvm/org/apache/storm/kafka/KafkaError.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.kafka;
-
-public enum KafkaError {
- NO_ERROR,
- OFFSET_OUT_OF_RANGE,
- INVALID_MESSAGE,
- UNKNOWN_TOPIC_OR_PARTITION,
- INVALID_FETCH_SIZE,
- LEADER_NOT_AVAILABLE,
- NOT_LEADER_FOR_PARTITION,
- REQUEST_TIMED_OUT,
- BROKER_NOT_AVAILABLE,
- REPLICA_NOT_AVAILABLE,
- MESSAGE_SIZE_TOO_LARGE,
- STALE_CONTROLLER_EPOCH,
- OFFSET_METADATA_TOO_LARGE,
- UNKNOWN;
-
- public static KafkaError getError(int errorCode) {
- if (errorCode < 0 || errorCode >= UNKNOWN.ordinal()) {
- return UNKNOWN;
- } else {
- return values()[errorCode];
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/storm/blob/e58ac3e0/external/storm-kafka/src/jvm/org/apache/storm/kafka/KafkaSpout.java
----------------------------------------------------------------------
diff --git a/external/storm-kafka/src/jvm/org/apache/storm/kafka/KafkaSpout.java b/external/storm-kafka/src/jvm/org/apache/storm/kafka/KafkaSpout.java
deleted file mode 100644
index aed5986..0000000
--- a/external/storm-kafka/src/jvm/org/apache/storm/kafka/KafkaSpout.java
+++ /dev/null
@@ -1,255 +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
- * <p/>
- * http://www.apache.org/licenses/LICENSE-2.0
- * <p/>
- * 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.kafka;
-
-import com.google.common.base.Strings;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-import org.apache.storm.Config;
-import org.apache.storm.kafka.PartitionManager.KafkaMessageId;
-import org.apache.storm.kafka.trident.GlobalPartitionInformation;
-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;
-
-// TODO: need to add blacklisting
-// TODO: need to make a best effort to not re-emit messages if don't have to
-public class KafkaSpout extends BaseRichSpout {
- private static final Logger LOG = LoggerFactory.getLogger(KafkaSpout.class);
- SpoutConfig _spoutConfig;
- SpoutOutputCollector _collector;
- PartitionCoordinator _coordinator;
- DynamicPartitionConnections _connections;
- ZkState _state;
- long _lastUpdateMs = 0;
- int _currPartitionIndex = 0;
-
- public KafkaSpout(SpoutConfig spoutConf) {
- _spoutConfig = spoutConf;
- }
-
- @Override
- public void open(Map<String, Object> conf, final TopologyContext context, final SpoutOutputCollector collector) {
- _collector = collector;
- String topologyInstanceId = context.getStormId();
- Map<String, Object> stateConf = new HashMap<>(conf);
- List<String> zkServers = _spoutConfig.zkServers;
- if (zkServers == null) {
- zkServers = (List<String>) conf.get(Config.STORM_ZOOKEEPER_SERVERS);
- }
- Integer zkPort = _spoutConfig.zkPort;
- if (zkPort == null) {
- zkPort = ((Number) conf.get(Config.STORM_ZOOKEEPER_PORT)).intValue();
- }
- stateConf.put(Config.TRANSACTIONAL_ZOOKEEPER_SERVERS, zkServers);
- stateConf.put(Config.TRANSACTIONAL_ZOOKEEPER_PORT, zkPort);
- stateConf.put(Config.TRANSACTIONAL_ZOOKEEPER_ROOT, _spoutConfig.zkRoot);
- _state = new ZkState(stateConf);
-
- _connections = new DynamicPartitionConnections(_spoutConfig, KafkaUtils.makeBrokerReader(conf, _spoutConfig));
-
- // using TransactionalState like this is a hack
- int totalTasks = context.getComponentTasks(context.getThisComponentId()).size();
- if (_spoutConfig.hosts instanceof StaticHosts) {
- _coordinator = new StaticCoordinator(_connections, conf,
- _spoutConfig, _state, context.getThisTaskIndex(),
- totalTasks, context.getThisTaskId(), topologyInstanceId);
- } else {
- _coordinator = new ZkCoordinator(_connections, conf,
- _spoutConfig, _state, context.getThisTaskIndex(),
- totalTasks, context.getThisTaskId(), topologyInstanceId);
- }
-
- context.registerMetric("kafkaOffset", new IMetric() {
- KafkaUtils.KafkaOffsetMetric _kafkaOffsetMetric = new KafkaUtils.KafkaOffsetMetric(_connections);
-
- @Override
- public Object getValueAndReset() {
- List<PartitionManager> pms = _coordinator.getMyManagedPartitions();
- Set<Partition> latestPartitions = new HashSet();
- for (PartitionManager pm : pms) {
- latestPartitions.add(pm.getPartition());
- }
- _kafkaOffsetMetric.refreshPartitions(latestPartitions);
- for (PartitionManager pm : pms) {
- _kafkaOffsetMetric.setOffsetData(pm.getPartition(), pm.getOffsetData());
- }
- return _kafkaOffsetMetric.getValueAndReset();
- }
- }, _spoutConfig.metricsTimeBucketSizeInSecs);
-
- context.registerMetric("kafkaPartition", new IMetric() {
- @Override
- public Object getValueAndReset() {
- List<PartitionManager> pms = _coordinator.getMyManagedPartitions();
- Map<String, Object> concatMetricsDataMaps = new HashMap<>();
- for (PartitionManager pm : pms) {
- concatMetricsDataMaps.putAll(pm.getMetricsDataMap());
- }
- return concatMetricsDataMaps;
- }
- }, _spoutConfig.metricsTimeBucketSizeInSecs);
- }
-
- @Override
- public void close() {
- _state.close();
- }
-
- @Override
- public void nextTuple() {
- List<PartitionManager> managers = _coordinator.getMyManagedPartitions();
- for (int i = 0; i < managers.size(); i++) {
-
- try {
- // in case the number of managers decreased
- _currPartitionIndex = _currPartitionIndex % managers.size();
- EmitState state = managers.get(_currPartitionIndex).next(_collector);
- if (state != EmitState.EMITTED_MORE_LEFT) {
- _currPartitionIndex = (_currPartitionIndex + 1) % managers.size();
- }
- if (state != EmitState.NO_EMITTED) {
- break;
- }
- } catch (FailedFetchException e) {
- LOG.warn("Fetch failed", e);
- _coordinator.refresh();
- }
- }
-
- long diffWithNow = System.currentTimeMillis() - _lastUpdateMs;
-
- /*
- As far as the System.currentTimeMillis() is dependent on System clock,
- additional check on negative value of diffWithNow in case of external changes.
- */
- if (diffWithNow > _spoutConfig.stateUpdateIntervalMs || diffWithNow < 0) {
- commit();
- }
- }
-
- private PartitionManager getManagerForPartition(int partition) {
- for (PartitionManager partitionManager : _coordinator.getMyManagedPartitions()) {
- if (partitionManager.getPartition().partition == partition) {
- return partitionManager;
- }
- }
- return null;
- }
-
- @Override
- public void ack(Object msgId) {
- KafkaMessageId id = (KafkaMessageId) msgId;
- PartitionManager m = _coordinator.getManager(id.partition);
- if (m != null) {
- m.ack(id.offset);
- } else {
- // managers for partitions changed - try to find new manager responsible for that partition
- PartitionManager newManager = getManagerForPartition(id.partition.partition);
- if (newManager != null) {
- newManager.ack(id.offset);
- }
- }
- }
-
- @Override
- public void fail(Object msgId) {
- KafkaMessageId id = (KafkaMessageId) msgId;
- PartitionManager m = _coordinator.getManager(id.partition);
- if (m != null) {
- m.fail(id.offset);
- } else {
- // managers for partitions changed - try to find new manager responsible for that partition
- PartitionManager newManager = getManagerForPartition(id.partition.partition);
- if (newManager != null) {
- newManager.fail(id.offset);
- }
- }
- }
-
- @Override
- public void deactivate() {
- commit();
- }
-
- @Override
- public void declareOutputFields(OutputFieldsDeclarer declarer) {
- if (!Strings.isNullOrEmpty(_spoutConfig.outputStreamId)) {
- declarer.declareStream(_spoutConfig.outputStreamId, _spoutConfig.scheme.getOutputFields());
- } else {
- declarer.declare(_spoutConfig.scheme.getOutputFields());
- }
- }
-
- @Override
- public Map<String, Object> getComponentConfiguration() {
- Map<String, Object> configuration = super.getComponentConfiguration();
- if (configuration == null) {
- configuration = new HashMap<>();
- }
- String configKeyPrefix = "config.";
- configuration.put(configKeyPrefix + "topics", this._spoutConfig.topic);
- StringBuilder zkServers = new StringBuilder();
- if (_spoutConfig.zkServers != null && _spoutConfig.zkServers.size() > 0) {
- for (String zkServer : this._spoutConfig.zkServers) {
- zkServers.append(zkServer + ":" + this._spoutConfig.zkPort + ",");
- }
- configuration.put(configKeyPrefix + "zkServers", zkServers.toString());
- }
- BrokerHosts brokerHosts = this._spoutConfig.hosts;
- String zkRoot = this._spoutConfig.zkRoot + "/" + this._spoutConfig.id;
- if (brokerHosts instanceof ZkHosts) {
- ZkHosts zkHosts = (ZkHosts) brokerHosts;
- configuration.put(configKeyPrefix + "zkNodeBrokers", zkHosts.brokerZkPath);
- } else if (brokerHosts instanceof StaticHosts) {
- StaticHosts staticHosts = (StaticHosts) brokerHosts;
- GlobalPartitionInformation globalPartitionInformation = staticHosts.getPartitionInformation();
- boolean useTopicNameForPath = globalPartitionInformation.getbUseTopicNameForPartitionPathId();
- if (useTopicNameForPath) {
- zkRoot += ("/" + this._spoutConfig.topic);
- }
- List<Partition> partitions = globalPartitionInformation.getOrderedPartitions();
- StringBuilder staticPartitions = new StringBuilder();
- StringBuilder leaderHosts = new StringBuilder();
- for (Partition partition : partitions) {
- staticPartitions.append(partition.partition + ",");
- leaderHosts.append(partition.host.host + ":" + partition.host.port).append(",");
- }
- configuration.put(configKeyPrefix + "partitions", staticPartitions.toString());
- configuration.put(configKeyPrefix + "leaders", leaderHosts.toString());
- }
- configuration.put(configKeyPrefix + "zkRoot", zkRoot);
- return configuration;
- }
-
- private void commit() {
- _lastUpdateMs = System.currentTimeMillis();
- for (PartitionManager manager : _coordinator.getMyManagedPartitions()) {
- manager.commit();
- }
- }
-
- static enum EmitState {
- EMITTED_MORE_LEFT,
- EMITTED_END,
- NO_EMITTED
- }
-
-}