metron-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From ceste...@apache.org
Subject metron git commit: METRON-1588 Migrate storm-kafka-client to 1.2.1 closes apache/incubator-metron#1039
Date Thu, 14 Jun 2018 16:07:15 GMT
Repository: metron
Updated Branches:
  refs/heads/master 77570464b -> 828ab7134


METRON-1588 Migrate storm-kafka-client to 1.2.1 closes apache/incubator-metron#1039


Project: http://git-wip-us.apache.org/repos/asf/metron/repo
Commit: http://git-wip-us.apache.org/repos/asf/metron/commit/828ab713
Tree: http://git-wip-us.apache.org/repos/asf/metron/tree/828ab713
Diff: http://git-wip-us.apache.org/repos/asf/metron/diff/828ab713

Branch: refs/heads/master
Commit: 828ab71346b458b73cb14d42d68e7471ead7fa4e
Parents: 7757046
Author: Jungtaek Lim <kabhwan@gmail.com>
Authored: Thu Jun 14 12:07:05 2018 -0400
Committer: cstella <cestella@gmail.com>
Committed: Thu Jun 14 12:07:05 2018 -0400

----------------------------------------------------------------------
 NOTICE                                          |   6 +
 dependencies_with_url.csv                       |   4 +
 metron-platform/metron-api/pom.xml              |   4 -
 .../parsers/topology/ParserTopologyBuilder.java |   2 +-
 .../metron-storm-kafka-override/pom.xml         |  83 ++++-
 .../KafkaSpoutRetryExponentialBackoff.java      | 336 +++++++++++++++++++
 .../storm/kafka/spout/internal/Timer.java       |   9 +-
 metron-platform/metron-storm-kafka/pom.xml      |   5 -
 pom.xml                                         |   4 +-
 9 files changed, 435 insertions(+), 18 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/metron/blob/828ab713/NOTICE
----------------------------------------------------------------------
diff --git a/NOTICE b/NOTICE
index ff6550b..5227768 100644
--- a/NOTICE
+++ b/NOTICE
@@ -10,3 +10,9 @@
    This product includes software developed by Chef Software (https://www.chef.io)
    Copyright (c) 2012-2015, Chef Software, Inc.
 
+   This includes derived works from the Apache Storm (ASLv2 licensed) project (https://github.com/apache/storm):
+   Copyright 2015 The Apache Software Foundation
+   The derived work is adapted from
+     org/apache/storm/kafka/spout/KafkaSpoutRetryExponentialBackoff.java
+     org/apache/storm/kafka/spout/internal/Timer.java
+   and can be found in the org.apache.storm.kafka package

http://git-wip-us.apache.org/repos/asf/metron/blob/828ab713/dependencies_with_url.csv
----------------------------------------------------------------------
diff --git a/dependencies_with_url.csv b/dependencies_with_url.csv
index df3bcd2..438ce3e 100644
--- a/dependencies_with_url.csv
+++ b/dependencies_with_url.csv
@@ -119,15 +119,19 @@ com.fasterxml.jackson.core:jackson-annotations:jar:2.2.3:compile,ASLv2,http://wi
 com.fasterxml.jackson.core:jackson-annotations:jar:2.7.4:compile,ASLv2,http://github.com/FasterXML/jackson
 com.fasterxml.jackson.core:jackson-annotations:jar:2.8.3:compile,ASLv2,http://github.com/FasterXML/jackson
 com.fasterxml.jackson.core:jackson-annotations:jar:2.9.0:compile,ASLv2,http://github.com/FasterXML/jackson
+com.fasterxml.jackson.core:jackson-annotations:jar:2.9.4:compile,ASLv2,http://github.com/FasterXML/jackson
 com.fasterxml.jackson.core:jackson-core:jar:2.2.3:compile,ASLv2,http://wiki.fasterxml.com/JacksonHome
+com.fasterxml.jackson.core:jackson-core:jar:2.6.3:compile,ASLv2,https://github.com/FasterXML/jackson-core
 com.fasterxml.jackson.core:jackson-core:jar:2.6.6:compile,ASLv2,https://github.com/FasterXML/jackson-core
 com.fasterxml.jackson.core:jackson-core:jar:2.7.4:compile,ASLv2,https://github.com/FasterXML/jackson-core
 com.fasterxml.jackson.core:jackson-core:jar:2.8.3:compile,ASLv2,https://github.com/FasterXML/jackson-core
+com.fasterxml.jackson.core:jackson-core:jar:2.9.4:compile,ASLv2,https://github.com/FasterXML/jackson-core
 com.fasterxml.jackson.core:jackson-core:jar:2.9.5:compile,ASLv2,https://github.com/FasterXML/jackson-core
 com.fasterxml.jackson.core:jackson-databind:jar:2.2.3:compile,ASLv2,http://wiki.fasterxml.com/JacksonHome
 com.fasterxml.jackson.core:jackson-databind:jar:2.4.3:compile,ASLv2,http://github.com/FasterXML/jackson
 com.fasterxml.jackson.core:jackson-databind:jar:2.7.4:compile,ASLv2,http://github.com/FasterXML/jackson
 com.fasterxml.jackson.core:jackson-databind:jar:2.8.3:compile,ASLv2,http://github.com/FasterXML/jackson
+com.fasterxml.jackson.core:jackson-databind:jar:2.9.4:compile,ASLv2,http://github.com/FasterXML/jackson
 com.fasterxml.jackson.core:jackson-databind:jar:2.9.5:compile,ASLv2,http://github.com/FasterXML/jackson
 com.fasterxml.jackson.dataformat:jackson-dataformat-cbor:jar:2.6.6:compile,ASLv2,http://wiki.fasterxml.com/JacksonForCbor
 com.fasterxml.jackson.dataformat:jackson-dataformat-smile:jar:2.6.6:compile,ASLv2,http://wiki.fasterxml.com/JacksonForSmile

http://git-wip-us.apache.org/repos/asf/metron/blob/828ab713/metron-platform/metron-api/pom.xml
----------------------------------------------------------------------
diff --git a/metron-platform/metron-api/pom.xml b/metron-platform/metron-api/pom.xml
index e3bf12d..59b1622 100644
--- a/metron-platform/metron-api/pom.xml
+++ b/metron-platform/metron-api/pom.xml
@@ -31,10 +31,6 @@
 		<zookeeper.version>3.4.5.2.0.6.0-76</zookeeper.version>
 		<logger.version>1.2.15</logger.version>
 
-        <storm-kafka-client.version>1.0.1</storm-kafka-client.version>
-        <storm-hdfs.version>0.0.7-SNAPSHOT</storm-hdfs.version>
-        <storm-hbase.version>0.0.5-SNAPSHOT</storm-hbase.version>
-
         <spring.integration.version>3.0.0.RELEASE</spring.integration.version>
         <spring.version>3.2.6.RELEASE</spring.version>
         <commons-fileupload.version>1.2.2</commons-fileupload.version>

http://git-wip-us.apache.org/repos/asf/metron/blob/828ab713/metron-platform/metron-parsers/src/main/java/org/apache/metron/parsers/topology/ParserTopologyBuilder.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-parsers/src/main/java/org/apache/metron/parsers/topology/ParserTopologyBuilder.java
b/metron-platform/metron-parsers/src/main/java/org/apache/metron/parsers/topology/ParserTopologyBuilder.java
index cd4ad50..5b3e0d5 100644
--- a/metron-platform/metron-parsers/src/main/java/org/apache/metron/parsers/topology/ParserTopologyBuilder.java
+++ b/metron-platform/metron-parsers/src/main/java/org/apache/metron/parsers/topology/ParserTopologyBuilder.java
@@ -162,7 +162,7 @@ public class ParserTopologyBuilder {
     Map<String, Object> kafkaSpoutConfigOptions = kafkaConfigOptional.orElse(new HashMap<>());
     String inputTopic = parserConfig.getSensorTopic() != null ? parserConfig.getSensorTopic()
: sensorType;
     kafkaSpoutConfigOptions.putIfAbsent( SpoutConfiguration.FIRST_POLL_OFFSET_STRATEGY.key
-            , KafkaSpoutConfig.FirstPollOffsetStrategy.UNCOMMITTED_EARLIEST.toString()
+            , KafkaSpoutConfig.FirstPollOffsetStrategy.UNCOMMITTED_EARLIEST.name()
     );
     kafkaSpoutConfigOptions.putIfAbsent( ConsumerConfig.GROUP_ID_CONFIG
             , inputTopic + "_parser"

http://git-wip-us.apache.org/repos/asf/metron/blob/828ab713/metron-platform/metron-storm-kafka-override/pom.xml
----------------------------------------------------------------------
diff --git a/metron-platform/metron-storm-kafka-override/pom.xml b/metron-platform/metron-storm-kafka-override/pom.xml
index 9be733e..a38d3bf 100644
--- a/metron-platform/metron-storm-kafka-override/pom.xml
+++ b/metron-platform/metron-storm-kafka-override/pom.xml
@@ -92,14 +92,87 @@
                 </exclusion>
             </exclusions>
         </dependency>
-        <dependency>
-            <groupId>org.apache.metron</groupId>
-            <artifactId>metron-common</artifactId>
-            <version>${project.parent.version}</version>
-        </dependency>
     </dependencies>
 
     <build>
+        <plugins>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-shade-plugin</artifactId>
+                <version>${global_shade_version}</version>
+                <configuration>
+                    <createDependencyReducedPom>true</createDependencyReducedPom>
+                </configuration>
+                <executions>
+                    <execution>
+                        <phase>package</phase>
+                        <goals>
+                            <goal>shade</goal>
+                        </goals>
+                        <configuration>
+                          <filters>
+                            <filter>
+                              <artifact>*:*</artifact>
+                              <excludes>
+                                <exclude>META-INF/*.SF</exclude>
+                                <exclude>META-INF/*.DSA</exclude>
+                                <exclude>META-INF/*.RSA</exclude>
+                              </excludes>
+                            </filter>
+                          </filters>
+                          <relocations>
+                                <relocation>
+                                    <pattern>com.google.common</pattern>
+                                    <shadedPattern>org.apache.metron.storm.kafka.override.guava.common</shadedPattern>
+                                </relocation>
+                                <relocation>
+                                    <pattern>com.google.thirdparty</pattern>
+                                    <shadedPattern>org.apache.metron.storm.kafka.override.guava.thirdparty</shadedPattern>
+                                </relocation>
+                                <relocation>
+                                    <pattern>org.apache.commons.lang</pattern>
+                                    <shadedPattern>org.apache.metron.storm.kafka.override.commons.lang</shadedPattern>
+                                </relocation>
+                                <relocation>
+                                    <pattern>com.fasterxml</pattern>
+                                    <shadedPattern>org.apache.metron.storm.kafka.override.com.fasterxml</shadedPattern>
+                                </relocation>
+                            </relocations>
+                            <artifactSet>
+                                <excludes>
+                                    <exclude>storm:storm-core:*</exclude>
+                                    <exclude>storm:storm-lib:*</exclude>
+                                    <exclude>org.slf4j.impl*</exclude>
+                                    <exclude>org.slf4j:slf4j-log4j*</exclude>
+                                </excludes>
+                            </artifactSet>
+                            <transformers>
+                                <transformer
+                                  implementation="org.apache.maven.plugins.shade.resource.DontIncludeResourceTransformer">
+                                     <resources>
+                                        <resource>.yaml</resource>
+                                        <resource>LICENSE.txt</resource>
+                                        <resource>ASL2.0</resource>
+                                        <resource>NOTICE.txt</resource>
+                                      </resources>
+                                </transformer>
+                                <!-- UNCOMMENT THIS IF YOU NEED TO REGENERATE THE BEST
GUESS NOTICES FILE WHICH REQUIRES PRUNING EVERY RELEASE -->
+                                <!--transformer implementation="org.apache.maven.plugins.shade.resource.ApacheNoticeResourceTransformer">
+                                    <addHeader>false</addHeader>
+                                    <projectName>${project.name}</projectName>
+                                </transformer-->
+                                <transformer
+                                        implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/>
+                                <transformer
+                                        implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
+                                    <mainClass></mainClass>
+                                </transformer>
+                            </transformers>
+                        </configuration>
+                    </execution>
+                </executions>
+            </plugin>
+        </plugins>
         <resources>
             <resource>
                 <directory>src/main/resources</directory>

http://git-wip-us.apache.org/repos/asf/metron/blob/828ab713/metron-platform/metron-storm-kafka-override/src/main/java/org/apache/storm/kafka/spout/KafkaSpoutRetryExponentialBackoff.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-storm-kafka-override/src/main/java/org/apache/storm/kafka/spout/KafkaSpoutRetryExponentialBackoff.java
b/metron-platform/metron-storm-kafka-override/src/main/java/org/apache/storm/kafka/spout/KafkaSpoutRetryExponentialBackoff.java
new file mode 100644
index 0000000..439188b
--- /dev/null
+++ b/metron-platform/metron-storm-kafka-override/src/main/java/org/apache/storm/kafka/spout/KafkaSpoutRetryExponentialBackoff.java
@@ -0,0 +1,336 @@
+/*
+ * 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.spout;
+
+import org.apache.kafka.common.TopicPartition;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.Serializable;
+import java.util.Collection;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.Set;
+import java.util.TreeSet;
+import java.util.concurrent.TimeUnit;
+import org.apache.commons.lang.Validate;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+
+/*
+ This file is pulled from Apache Storm, with some modification to support lower version of
+ Apache Storm.
+
+ - Time.nanoTime() is introduced in Storm 1.1.0 so we changed to System.nanoTime()
+ -- Time.nanoTime() calls System.nanoTime() when it's not in time simulation mode.
+*/
+
+/**
+ * Implementation of {@link KafkaSpoutRetryService} using the exponential backoff formula.
The time of the nextRetry is set as follows:
+ * nextRetry = failCount == 1 ? currentTime + initialDelay : currentTime + delayPeriod*2^(failCount-1)
   where failCount = 1, 2, 3, ...
+ * nextRetry = Min(nextRetry, currentTime + maxDelay)
+ */
+public class KafkaSpoutRetryExponentialBackoff implements KafkaSpoutRetryService {
+    private static final Logger LOG = LoggerFactory.getLogger(KafkaSpoutRetryExponentialBackoff.class);
+    private static final RetryEntryTimeStampComparator RETRY_ENTRY_TIME_STAMP_COMPARATOR
= new RetryEntryTimeStampComparator();
+
+    private final TimeInterval initialDelay;
+    private final TimeInterval delayPeriod;
+    private final TimeInterval maxDelay;
+    private final int maxRetries;
+
+    //This class assumes that there is at most one retry schedule per message id in this
set at a time.
+    private final Set<RetrySchedule> retrySchedules = new TreeSet<>(RETRY_ENTRY_TIME_STAMP_COMPARATOR);
+    private final Set<KafkaSpoutMessageId> toRetryMsgs = new HashSet<>();   
  // Convenience data structure to speedup lookups
+
+    /**
+     * Comparator ordering by timestamp 
+     */
+    private static class RetryEntryTimeStampComparator implements Serializable, Comparator<RetrySchedule>
{
+        @Override
+        public int compare(RetrySchedule entry1, RetrySchedule entry2) {
+            int result = Long.valueOf(entry1.nextRetryTimeNanos()).compareTo(entry2.nextRetryTimeNanos());
+            
+            if(result == 0) {
+                //TreeSet uses compareTo instead of equals() for the Set contract
+                //Ensure that we can save two retry schedules with the same timestamp
+                result = entry1.hashCode() - entry2.hashCode();
+            }
+            return result;
+        }
+    }
+
+    private class RetrySchedule {
+        private final KafkaSpoutMessageId msgId;
+        private long nextRetryTimeNanos;
+
+        public RetrySchedule(KafkaSpoutMessageId msgId, long nextRetryTimeNanos) {
+            this.msgId = msgId;
+            this.nextRetryTimeNanos = nextRetryTimeNanos;
+            LOG.debug("Created {}", this);
+        }
+
+        public void setNextRetryTimeNanos() {
+            nextRetryTimeNanos = nextTime(msgId);
+            LOG.debug("Updated {}", this);
+        }
+
+        public boolean retry(long currentTimeNanos) {
+            return nextRetryTimeNanos <= currentTimeNanos;
+        }
+
+        @Override
+        public String toString() {
+            return "RetrySchedule{" +
+                    "msgId=" + msgId +
+                    ", nextRetryTimeNanos=" + nextRetryTimeNanos +
+                    '}';
+        }
+
+        public KafkaSpoutMessageId msgId() {
+            return msgId;
+        }
+
+        public long nextRetryTimeNanos() {
+            return nextRetryTimeNanos;
+        }
+    }
+
+    public static class TimeInterval implements Serializable {
+        private final long lengthNanos;
+        private final TimeUnit timeUnit;
+        private final long length;
+
+        /**
+         * @param length length of the time interval in the units specified by {@link TimeUnit}
+         * @param timeUnit unit used to specify a time interval on which to specify a time
unit
+         */
+        public TimeInterval(long length, TimeUnit timeUnit) {
+            this.lengthNanos = timeUnit.toNanos(length);
+            this.timeUnit = timeUnit;
+            this.length = length;
+        }
+        
+        public static TimeInterval seconds(long length) {
+            return new TimeInterval(length, TimeUnit.SECONDS);
+        }
+
+        public static TimeInterval milliSeconds(long length) {
+            return new TimeInterval(length, TimeUnit.MILLISECONDS);
+        }
+        
+        public static TimeInterval microSeconds(long length) {
+            return new TimeInterval(length, TimeUnit.MICROSECONDS);
+        }
+        
+        public long lengthNanos() {
+            return lengthNanos;
+        }
+        
+        public TimeUnit timeUnit() {
+            return timeUnit;
+        }
+
+        @Override
+        public String toString() {
+            return "TimeInterval{" +
+                    "length=" + length +
+                    ", timeUnit=" + timeUnit +
+                    '}';
+        }
+    }
+
+    /**
+     * The time stamp of the next retry is scheduled according to the exponential backoff
formula ( geometric progression):
+     * nextRetry = failCount == 1 ? currentTime + initialDelay : currentTime + delayPeriod^(failCount-1)
where failCount = 1, 2, 3, ...
+     * nextRetry = Min(nextRetry, currentTime + maxDelay).
+     * 
+     * By specifying a value for maxRetries lower than Integer.MAX_VALUE, the user decides
to sacrifice guarantee of delivery for the previous
+     * polled records in favor of processing more records.
+     *
+     * @param initialDelay      initial delay of the first retry
+     * @param delayPeriod       the time interval that is the ratio of the exponential backoff
formula (geometric progression)
+     * @param maxRetries        maximum number of times a tuple is retried before being acked
and scheduled for commit
+     * @param maxDelay          maximum amount of time waiting before retrying
+     *
+     */
+    public KafkaSpoutRetryExponentialBackoff(TimeInterval initialDelay, TimeInterval delayPeriod,
int maxRetries, TimeInterval maxDelay) {
+        this.initialDelay = initialDelay;
+        this.delayPeriod = delayPeriod;
+        this.maxRetries = maxRetries;
+        this.maxDelay = maxDelay;
+        LOG.debug("Instantiated {}", this.toStringImpl());
+    }
+
+    @Override
+    public Map<TopicPartition, Long> earliestRetriableOffsets() {
+        final Map<TopicPartition, Long> tpToEarliestRetriableOffset = new HashMap<>();
+        final long currentTimeNanos = System.nanoTime();
+        for (RetrySchedule retrySchedule : retrySchedules) {
+            if (retrySchedule.retry(currentTimeNanos)) {
+                final KafkaSpoutMessageId msgId = retrySchedule.msgId;
+                final TopicPartition tpForMessage = new TopicPartition(msgId.topic(), msgId.partition());
+                final Long currentLowestOffset = tpToEarliestRetriableOffset.get(tpForMessage);
+                if(currentLowestOffset != null) {
+                    tpToEarliestRetriableOffset.put(tpForMessage, Math.min(currentLowestOffset,
msgId.offset()));
+                } else {
+                    tpToEarliestRetriableOffset.put(tpForMessage, msgId.offset());
+                }
+            } else {
+                break;  // Stop searching as soon as passed current time
+            }
+        }
+        LOG.debug("Topic partitions with entries ready to be retried [{}] ", tpToEarliestRetriableOffset);
+        return tpToEarliestRetriableOffset;
+    }
+
+    @Override
+    public boolean isReady(KafkaSpoutMessageId msgId) {
+        boolean retry = false;
+        if (isScheduled(msgId)) {
+            final long currentTimeNanos = System.nanoTime();
+            for (RetrySchedule retrySchedule : retrySchedules) {
+                if (retrySchedule.retry(currentTimeNanos)) {
+                    if (retrySchedule.msgId.equals(msgId)) {
+                        retry = true;
+                        LOG.debug("Found entry to retry {}", retrySchedule);
+                        break; //Stop searching if the message is known to be ready for retry
+                    }
+                } else {
+                    LOG.debug("Entry to retry not found {}", retrySchedule);
+                    break;  // Stop searching as soon as passed current time
+                }
+            }
+        }
+        return retry;
+    }
+
+    @Override
+    public boolean isScheduled(KafkaSpoutMessageId msgId) {
+        return toRetryMsgs.contains(msgId);
+    }
+
+    @Override
+    public boolean remove(KafkaSpoutMessageId msgId) {
+        boolean removed = false;
+        if (isScheduled(msgId)) {
+            toRetryMsgs.remove(msgId);
+            for (Iterator<RetrySchedule> iterator = retrySchedules.iterator(); iterator.hasNext();
) {
+                final RetrySchedule retrySchedule = iterator.next();
+                if (retrySchedule.msgId().equals(msgId)) {
+                    iterator.remove();
+                    removed = true;
+                    break; //There is at most one schedule per message id
+                }
+            }
+        }
+        LOG.debug(removed ? "Removed {} " : "Not removed {}", msgId);
+        LOG.trace("Current state {}", retrySchedules);
+        return removed;
+    }
+
+    @Override
+    public boolean retainAll(Collection<TopicPartition> topicPartitions) {
+        boolean result = false;
+        for (Iterator<RetrySchedule> rsIterator = retrySchedules.iterator(); rsIterator.hasNext();
) {
+            final RetrySchedule retrySchedule = rsIterator.next();
+            final KafkaSpoutMessageId msgId = retrySchedule.msgId;
+            final TopicPartition tpRetry= new TopicPartition(msgId.topic(), msgId.partition());
+            if (!topicPartitions.contains(tpRetry)) {
+                rsIterator.remove();
+                toRetryMsgs.remove(msgId);
+                LOG.debug("Removed {}", retrySchedule);
+                LOG.trace("Current state {}", retrySchedules);
+                result = true;
+            }
+        }
+        return result;
+    }
+
+    @Override
+    public boolean schedule(KafkaSpoutMessageId msgId) {
+        if (msgId.numFails() > maxRetries) {
+            LOG.debug("Not scheduling [{}] because reached maximum number of retries [{}].",
msgId, maxRetries);
+            return false;
+        } else {
+            //Remove existing schedule for the message id
+            remove(msgId);
+            final RetrySchedule retrySchedule = new RetrySchedule(msgId, nextTime(msgId));
+            retrySchedules.add(retrySchedule);
+            toRetryMsgs.add(msgId);
+            LOG.debug("Scheduled. {}", retrySchedule);
+            LOG.trace("Current state {}", retrySchedules);
+            return true;
+        }
+    }
+    
+    @Override
+    public int readyMessageCount() {
+        int count = 0;
+        final long currentTimeNanos = System.nanoTime();
+        for (RetrySchedule retrySchedule : retrySchedules) {
+            if (retrySchedule.retry(currentTimeNanos)) {
+                ++count;
+            } else {
+                break; //Stop counting when past current time
+            }
+        }
+        return count;
+    }
+
+    @Override
+    public KafkaSpoutMessageId getMessageId(ConsumerRecord<?, ?> record) {
+        KafkaSpoutMessageId msgId = new KafkaSpoutMessageId(record);
+        if (isScheduled(msgId)) {
+            for (KafkaSpoutMessageId originalMsgId : toRetryMsgs) {
+                if (originalMsgId.equals(msgId)) {
+                    return originalMsgId;
+                }
+            }
+        }
+        return msgId;
+    }
+
+    // if value is greater than Long.MAX_VALUE it truncates to Long.MAX_VALUE
+    private long nextTime(KafkaSpoutMessageId msgId) {
+        Validate.isTrue(msgId.numFails() > 0, "nextTime assumes the message has failed
at least once");
+        final long currentTimeNanos = System.nanoTime();
+        final long nextTimeNanos = msgId.numFails() == 1                // numFails = 1,
2, 3, ...
+                ? currentTimeNanos + initialDelay.lengthNanos
+                : currentTimeNanos + delayPeriod.lengthNanos * (long)(Math.pow(2, msgId.numFails()-1));
+        return Math.min(nextTimeNanos, currentTimeNanos + maxDelay.lengthNanos);
+    }
+
+    @Override
+    public String toString() {
+        return toStringImpl();
+    }
+    
+    private String toStringImpl() {
+        //This is here to avoid an overridable call in the constructor
+        return "KafkaSpoutRetryExponentialBackoff{" +
+                "delay=" + initialDelay +
+                ", ratio=" + delayPeriod +
+                ", maxRetries=" + maxRetries +
+                ", maxRetryDelay=" + maxDelay +
+                '}';
+    }
+}

http://git-wip-us.apache.org/repos/asf/metron/blob/828ab713/metron-platform/metron-storm-kafka-override/src/main/java/org/apache/storm/kafka/spout/internal/Timer.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-storm-kafka-override/src/main/java/org/apache/storm/kafka/spout/internal/Timer.java
b/metron-platform/metron-storm-kafka-override/src/main/java/org/apache/storm/kafka/spout/internal/Timer.java
index f9782ab..0b045c0 100644
--- a/metron-platform/metron-storm-kafka-override/src/main/java/org/apache/storm/kafka/spout/internal/Timer.java
+++ b/metron-platform/metron-storm-kafka-override/src/main/java/org/apache/storm/kafka/spout/internal/Timer.java
@@ -18,7 +18,14 @@
 package org.apache.storm.kafka.spout.internal;
 
 import java.util.concurrent.TimeUnit;
-import org.apache.storm.utils.Time;
+
+/*
+ This file is pulled from Apache Storm, with some modification to support lower version of
+ Apache Storm.
+
+ - Time.nanoTime() is introduced in Storm 1.1.0 so we changed to System.nanoTime()
+ -- Time.nanoTime() calls System.nanoTime() when it's not in time simulation mode.
+*/
 
 public class Timer {
   private final long delay;

http://git-wip-us.apache.org/repos/asf/metron/blob/828ab713/metron-platform/metron-storm-kafka/pom.xml
----------------------------------------------------------------------
diff --git a/metron-platform/metron-storm-kafka/pom.xml b/metron-platform/metron-storm-kafka/pom.xml
index 5ccc0d9..fdcde4f 100644
--- a/metron-platform/metron-storm-kafka/pom.xml
+++ b/metron-platform/metron-storm-kafka/pom.xml
@@ -36,11 +36,6 @@
             <version>${project.parent.version}</version>
         </dependency>
         <dependency>
-            <groupId>org.apache.storm</groupId>
-            <artifactId>storm-kafka-client</artifactId>
-            <version>${global_storm_kafka_version}</version>
-        </dependency>
-        <dependency>
             <groupId>org.apache.kafka</groupId>
             <artifactId>kafka-clients</artifactId>
             <version>${global_kafka_version}</version>

http://git-wip-us.apache.org/repos/asf/metron/blob/828ab713/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index 75c9a31..a5689f5 100644
--- a/pom.xml
+++ b/pom.xml
@@ -91,7 +91,7 @@
         <global_curator_version>2.7.1</global_curator_version>
         <global_classindex_version>3.3</global_classindex_version>
         <global_storm_version>1.0.3</global_storm_version>
-        <global_storm_kafka_version>1.1.0</global_storm_kafka_version>
+        <global_storm_kafka_version>1.2.2</global_storm_kafka_version>
         <global_flux_version>${base_flux_version}</global_flux_version>
         <global_pcap_version>1.7.1</global_pcap_version>
         <global_kafka_version>0.10.0.1</global_kafka_version>
@@ -129,7 +129,7 @@
             <properties>
                 <hdp_version>2.5.0.0</hdp_version>
                 <build_number>1245</build_number>
-                <global_storm_kafka_version>1.1.0</global_storm_kafka_version>
+                <global_storm_kafka_version>1.2.2</global_storm_kafka_version>
                 <global_storm_version>${base_storm_version}.${hdp_version}-${build_number}</global_storm_version>
                 <global_kafka_version>${base_kafka_version}.${hdp_version}-${build_number}</global_kafka_version>
             </properties>


Mime
View raw message