This is an automated email from the ASF dual-hosted git repository.
sekikn pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/bigtop.git
The following commit(s) were added to refs/heads/master by this push:
new 85218c9 BIGTOP-3305. Bump Flume to 1.9.0. (#599)
85218c9 is described below
commit 85218c9dfdb5b45812c1bb4bbbea4b751591e987
Author: Masatake Iwasaki <[email protected]>
AuthorDate: Sun Mar 15 22:47:29 2020 +0900
BIGTOP-3305. Bump Flume to 1.9.0. (#599)
---
.../src/common/flume/do-component-build | 3 +-
.../src/common/flume/patch0-FLUME-2662.diff | 6 +-
.../common/flume/patch1-FLUME-3026_rebased.diff | 99 ----------------------
.../src/common/flume/patch2-scala-symbol.diff | 96 ---------------------
.../src/common/flume/patch3-FLUME-3354.diff | 13 ---
.../src/common/flume/patch4-FLUME-3355.diff | 61 +++++++++++++
bigtop.bom | 2 +-
7 files changed, 67 insertions(+), 213 deletions(-)
diff --git a/bigtop-packages/src/common/flume/do-component-build
b/bigtop-packages/src/common/flume/do-component-build
index 6578085..85506dd 100755
--- a/bigtop-packages/src/common/flume/do-component-build
+++ b/bigtop-packages/src/common/flume/do-component-build
@@ -28,8 +28,9 @@ sed -i -e \
# This profile will activate hadoop2.version as the relevant property
# and will also make us specify hbase version with -hadoop2 prefix
mvn -DskipTests \
- -Dhadoop2.version=$HADOOP_VERSION \
+ -Dhadoop.version=${HADOOP_VERSION} \
-Dhbase.version=${HBASE_VERSION} \
-Dhive.version=${HIVE_VERSION} \
-Dkafka.version=${KAFKA_VERSION} \
+ -Dzookeeper.version=${ZOOKEEPER_VERSION} \
clean install "$@"
diff --git a/bigtop-packages/src/common/flume/patch0-FLUME-2662.diff
b/bigtop-packages/src/common/flume/patch0-FLUME-2662.diff
index 61497b9..bc060e9 100644
--- a/bigtop-packages/src/common/flume/patch0-FLUME-2662.diff
+++ b/bigtop-packages/src/common/flume/patch0-FLUME-2662.diff
@@ -1,8 +1,8 @@
diff --git a/pom.xml b/pom.xml
-index 3c82a47..bdd998d 100644
+index b756c0c..a06f819 100644
--- a/pom.xml
+++ b/pom.xml
-@@ -58,7 +58,7 @@ limitations under the License.
+@@ -57,7 +57,7 @@ limitations under the License.
<commons-collections.version>3.2.2</commons-collections.version>
<commons-compress.version>1.4.1</commons-compress.version>
<commons-dbcp.version>1.4</commons-dbcp.version>
@@ -10,4 +10,4 @@ index 3c82a47..bdd998d 100644
+ <commons-io.version>2.4</commons-io.version>
<commons-lang.version>2.5</commons-lang.version>
<curator.version>2.6.0</curator.version>
- <derby.version>10.11.1.1</derby.version>
+ <derby.version>10.14.1.0</derby.version>
diff --git a/bigtop-packages/src/common/flume/patch1-FLUME-3026_rebased.diff
b/bigtop-packages/src/common/flume/patch1-FLUME-3026_rebased.diff
deleted file mode 100644
index 20a2ddd..0000000
--- a/bigtop-packages/src/common/flume/patch1-FLUME-3026_rebased.diff
+++ /dev/null
@@ -1,99 +0,0 @@
-diff --git
a/flume-ng-channels/flume-kafka-channel/src/test/java/org/apache/flume/channel/kafka/TestKafkaChannel.java
b/flume-ng-channels/flume-kafka-channel/src/test/java/org/apache/flume/channel/kafka/TestKafkaChannel.java
-index 5e5f2d0..63607f7 100644
----
a/flume-ng-channels/flume-kafka-channel/src/test/java/org/apache/flume/channel/kafka/TestKafkaChannel.java
-+++
b/flume-ng-channels/flume-kafka-channel/src/test/java/org/apache/flume/channel/kafka/TestKafkaChannel.java
-@@ -20,6 +20,7 @@ package org.apache.flume.channel.kafka;
-
- import com.google.common.collect.Lists;
- import kafka.admin.AdminUtils;
-+import kafka.admin.RackAwareMode;
- import kafka.utils.ZKGroupTopicDirs;
- import kafka.utils.ZkUtils;
- import org.apache.commons.lang.RandomStringUtils;
-@@ -883,7 +884,8 @@ public class TestKafkaChannel {
- ZkUtils.apply(testUtil.getZkUrl(), sessionTimeoutMs,
connectionTimeoutMs, false);
- int replicationFactor = 1;
- Properties topicConfig = new Properties();
-- AdminUtils.createTopic(zkUtils, topicName, numPartitions,
replicationFactor, topicConfig);
-+ AdminUtils.createTopic(zkUtils, topicName, numPartitions,
replicationFactor, topicConfig,
-+ RackAwareMode.Disabled$.MODULE$);
- }
-
- public static void deleteTopic(String topicName) {
-diff --git
a/flume-ng-sinks/flume-ng-kafka-sink/src/test/java/org/apache/flume/sink/kafka/TestKafkaSink.java
b/flume-ng-sinks/flume-ng-kafka-sink/src/test/java/org/apache/flume/sink/kafka/TestKafkaSink.java
-index d92c71f..66c6fe3 100644
----
a/flume-ng-sinks/flume-ng-kafka-sink/src/test/java/org/apache/flume/sink/kafka/TestKafkaSink.java
-+++
b/flume-ng-sinks/flume-ng-kafka-sink/src/test/java/org/apache/flume/sink/kafka/TestKafkaSink.java
-@@ -21,6 +21,7 @@ package org.apache.flume.sink.kafka;
- import com.google.common.base.Charsets;
-
- import kafka.admin.AdminUtils;
-+import kafka.admin.RackAwareMode;
- import kafka.message.MessageAndMetadata;
- import kafka.utils.ZkUtils;
-
-@@ -674,7 +675,8 @@ public class TestKafkaSink {
- ZkUtils.apply(testUtil.getZkUrl(), sessionTimeoutMs,
connectionTimeoutMs, false);
- int replicationFactor = 1;
- Properties topicConfig = new Properties();
-- AdminUtils.createTopic(zkUtils, topicName, numPartitions,
replicationFactor, topicConfig);
-+ AdminUtils.createTopic(zkUtils, topicName, numPartitions,
replicationFactor, topicConfig,
-+ RackAwareMode.Disabled$.MODULE$);
- }
-
- public static void deleteTopic(String topicName) {
-@@ -698,4 +700,4 @@ public class TestKafkaSink {
- return newTopic;
- }
-
--}
-\ No newline at end of file
-+}
-diff --git
a/flume-ng-sources/flume-kafka-source/src/test/java/org/apache/flume/source/kafka/KafkaSourceEmbeddedKafka.java
b/flume-ng-sources/flume-kafka-source/src/test/java/org/apache/flume/source/kafka/KafkaSourceEmbeddedKafka.java
-index 53bd65c..ba75623 100644
----
a/flume-ng-sources/flume-kafka-source/src/test/java/org/apache/flume/source/kafka/KafkaSourceEmbeddedKafka.java
-+++
b/flume-ng-sources/flume-kafka-source/src/test/java/org/apache/flume/source/kafka/KafkaSourceEmbeddedKafka.java
-@@ -17,6 +17,7 @@
- package org.apache.flume.source.kafka;
-
- import kafka.admin.AdminUtils;
-+import kafka.admin.RackAwareMode;
- import kafka.server.KafkaConfig;
- import kafka.server.KafkaServerStartable;
- import kafka.utils.ZkUtils;
-@@ -131,7 +132,8 @@ public class KafkaSourceEmbeddedKafka {
- ZkUtils zkUtils = ZkUtils.apply(zkClient, false);
- int replicationFactor = 1;
- Properties topicConfig = new Properties();
-- AdminUtils.createTopic(zkUtils, topicName, numPartitions,
replicationFactor, topicConfig);
-+ AdminUtils.createTopic(zkUtils, topicName, numPartitions,
replicationFactor, topicConfig,
-+ RackAwareMode.Disabled$.MODULE$);
- }
-
- }
-diff --git
a/flume-ng-sources/flume-kafka-source/src/test/java/org/apache/flume/source/kafka/TestKafkaSource.java
b/flume-ng-sources/flume-kafka-source/src/test/java/org/apache/flume/source/kafka/TestKafkaSource.java
-index 7804fa2..2d5bbf8 100644
----
a/flume-ng-sources/flume-kafka-source/src/test/java/org/apache/flume/source/kafka/TestKafkaSource.java
-+++
b/flume-ng-sources/flume-kafka-source/src/test/java/org/apache/flume/source/kafka/TestKafkaSource.java
-@@ -20,7 +20,7 @@ package org.apache.flume.source.kafka;
- import com.google.common.base.Charsets;
- import com.google.common.collect.Lists;
- import junit.framework.Assert;
--import kafka.common.TopicExistsException;
-+import org.apache.kafka.common.errors.TopicExistsException;
- import kafka.utils.ZKGroupTopicDirs;
- import kafka.utils.ZkUtils;
- import org.apache.avro.io.BinaryEncoder;
-diff --git a/pom.xml b/pom.xml
-index 3c82a47..2276355 100644
---- a/pom.xml
-+++ b/pom.xml
-@@ -77,7 +77,7 @@ limitations under the License.
- <jetty.version>6.1.26</jetty.version>
- <joda-time.version>2.9.9</joda-time.version>
- <junit.version>4.10</junit.version>
-- <kafka.version>0.9.0.1</kafka.version>
-+ <kafka.version>0.10.2.2</kafka.version>
- <kite.version>1.0.0</kite.version>
- <hive.version>1.0.0</hive.version>
- <lifecycle-mapping.version>1.0.0</lifecycle-mapping.version>
diff --git a/bigtop-packages/src/common/flume/patch2-scala-symbol.diff
b/bigtop-packages/src/common/flume/patch2-scala-symbol.diff
deleted file mode 100644
index 17833d4..0000000
--- a/bigtop-packages/src/common/flume/patch2-scala-symbol.diff
+++ /dev/null
@@ -1,96 +0,0 @@
-From f809342685fcf1e1a2dc0fc227de84ccb26dad10 Mon Sep 17 00:00:00 2001
-From: Anton Chevychalov <[email protected]>
-Date: Wed, 25 Oct 2017 15:47:48 +0300
-Subject: [PATCH] Fix kafka and Scala 2.11 trouble
-
----
- .../org/apache/flume/channel/kafka/KafkaChannel.java | 4 ++--
- .../org/apache/flume/source/kafka/KafkaSource.java | 18 ++++++++++++++----
- 2 files changed, 16 insertions(+), 6 deletions(-)
-
-diff --git
a/flume-ng-channels/flume-kafka-channel/src/main/java/org/apache/flume/channel/kafka/KafkaChannel.java
b/flume-ng-channels/flume-kafka-channel/src/main/java/org/apache/flume/channel/kafka/KafkaChannel.java
-index 5bd9be0..46494fd 100644
----
a/flume-ng-channels/flume-kafka-channel/src/main/java/org/apache/flume/channel/kafka/KafkaChannel.java
-+++
b/flume-ng-channels/flume-kafka-channel/src/main/java/org/apache/flume/channel/kafka/KafkaChannel.java
-@@ -77,7 +77,7 @@ import java.util.concurrent.atomic.AtomicBoolean;
- import java.util.concurrent.atomic.AtomicReference;
-
- import static org.apache.flume.channel.kafka.KafkaChannelConfiguration.*;
--import static scala.collection.JavaConverters.asJavaListConverter;
-+import scala.collection.JavaConverters;
-
- public class KafkaChannel extends BasicChannelSemantics {
-
-@@ -357,7 +357,7 @@ public class KafkaChannel extends BasicChannelSemantics {
- private Map<TopicPartition, OffsetAndMetadata> getZookeeperOffsets(ZkUtils
client) {
- Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
- ZKGroupTopicDirs topicDirs = new ZKGroupTopicDirs(groupId, topicStr);
-- List<String> partitions = asJavaListConverter(
-+ List<String> partitions = JavaConverters.seqAsJavaListConverter(
-
client.getChildrenParentMayNotExist(topicDirs.consumerOffsetDir())).asJava();
- for (String partition : partitions) {
- TopicPartition key = new TopicPartition(topicStr,
Integer.valueOf(partition));
-diff --git
a/flume-ng-sources/flume-kafka-source/src/main/java/org/apache/flume/source/kafka/KafkaSource.java
b/flume-ng-sources/flume-kafka-source/src/main/java/org/apache/flume/source/kafka/KafkaSource.java
-index ffdc96e..960e9e8 100644
----
a/flume-ng-sources/flume-kafka-source/src/main/java/org/apache/flume/source/kafka/KafkaSource.java
-+++
b/flume-ng-sources/flume-kafka-source/src/main/java/org/apache/flume/source/kafka/KafkaSource.java
-@@ -28,8 +28,10 @@ import java.util.Properties;
- import java.util.UUID;
- import java.util.concurrent.atomic.AtomicBoolean;
- import java.util.regex.Pattern;
-+import java.util.stream.Collectors;
-
- import com.google.common.annotations.VisibleForTesting;
-+import kafka.cluster.Broker;
- import kafka.cluster.BrokerEndPoint;
- import kafka.utils.ZKGroupTopicDirs;
- import kafka.utils.ZkUtils;
-@@ -57,6 +59,7 @@ import org.apache.kafka.clients.consumer.KafkaConsumer;
- import org.apache.kafka.clients.consumer.OffsetAndMetadata;
- import org.apache.kafka.common.PartitionInfo;
- import org.apache.kafka.common.TopicPartition;
-+import org.apache.kafka.common.network.ListenerName;
- import org.apache.kafka.common.protocol.SecurityProtocol;
- import org.apache.kafka.common.security.JaasUtils;
- import org.slf4j.Logger;
-@@ -64,9 +67,10 @@ import org.slf4j.LoggerFactory;
-
- import com.google.common.base.Optional;
- import scala.Option;
-+import scala.collection.Seq;
-
- import static org.apache.flume.source.kafka.KafkaSourceConstants.*;
--import static scala.collection.JavaConverters.asJavaListConverter;
-+import scala.collection.JavaConverters;
-
- /**
- * A Source for Kafka which reads messages from kafka topics.
-@@ -464,8 +468,14 @@ public class KafkaSource extends AbstractPollableSource
- ZkUtils zkUtils = ZkUtils.apply(zookeeperConnect, ZK_SESSION_TIMEOUT,
ZK_CONNECTION_TIMEOUT,
- JaasUtils.isZkSecurityEnabled());
- try {
-- List<BrokerEndPoint> endPoints =
--
asJavaListConverter(zkUtils.getAllBrokerEndPointsForChannel(securityProtocol)).asJava();
-+ Seq<Broker> allBrokersInCluster = zkUtils.getAllBrokersInCluster();
-+ List<Broker> brokerList = JavaConverters.seqAsJavaListConverter(
-+ zkUtils.getAllBrokersInCluster()).asJava();
-+ List<BrokerEndPoint> endPoints = brokerList.stream()
-+ .map(broker -> broker.getBrokerEndPoint(
-+ ListenerName.forSecurityProtocol(securityProtocol))
-+ )
-+ .collect(Collectors.toList());
- List<String> connections = new ArrayList<>();
- for (BrokerEndPoint endPoint : endPoints) {
- connections.add(endPoint.connectionString());
-@@ -597,7 +607,7 @@ public class KafkaSource extends AbstractPollableSource
- String
topicStr) {
- Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
- ZKGroupTopicDirs topicDirs = new ZKGroupTopicDirs(groupId, topicStr);
-- List<String> partitions = asJavaListConverter(
-+ List<String> partitions = JavaConverters.seqAsJavaListConverter(
-
client.getChildrenParentMayNotExist(topicDirs.consumerOffsetDir())).asJava();
- for (String partition : partitions) {
- TopicPartition key = new TopicPartition(topicStr,
Integer.valueOf(partition));
---
-1.9.1
-
diff --git a/bigtop-packages/src/common/flume/patch3-FLUME-3354.diff
b/bigtop-packages/src/common/flume/patch3-FLUME-3354.diff
index 3a0e912..3ad7274 100644
--- a/bigtop-packages/src/common/flume/patch3-FLUME-3354.diff
+++ b/bigtop-packages/src/common/flume/patch3-FLUME-3354.diff
@@ -52,16 +52,3 @@ index 1fcb4eb..db724fa 100644
}
public static void createDbAndTable(Driver driver, String databaseName,
-diff --git a/pom.xml b/pom.xml
-index bdd998d..2304421 100644
---- a/pom.xml
-+++ b/pom.xml
-@@ -79,7 +79,7 @@ limitations under the License.
- <junit.version>4.10</junit.version>
- <kafka.version>0.10.2.2</kafka.version>
- <kite.version>1.0.0</kite.version>
-- <hive.version>1.0.0</hive.version>
-+ <hive.version>2.3.6</hive.version>
- <lifecycle-mapping.version>1.0.0</lifecycle-mapping.version>
- <log4j-extras.version>1.1</log4j-extras.version>
- <log4j-log4j.version>1.2.17</log4j-log4j.version>
diff --git a/bigtop-packages/src/common/flume/patch4-FLUME-3355.diff
b/bigtop-packages/src/common/flume/patch4-FLUME-3355.diff
new file mode 100644
index 0000000..f066772
--- /dev/null
+++ b/bigtop-packages/src/common/flume/patch4-FLUME-3355.diff
@@ -0,0 +1,61 @@
+diff --git
a/flume-ng-channels/flume-kafka-channel/src/main/java/org/apache/flume/channel/kafka/KafkaChannel.java
b/flume-ng-channels/flume-kafka-channel/src/main/java/org/apache/flume/channel/kafka/KafkaChannel.java
+index 40494d4..cbfbc9d 100644
+---
a/flume-ng-channels/flume-kafka-channel/src/main/java/org/apache/flume/channel/kafka/KafkaChannel.java
++++
b/flume-ng-channels/flume-kafka-channel/src/main/java/org/apache/flume/channel/kafka/KafkaChannel.java
+@@ -315,7 +315,7 @@ public class KafkaChannel extends BasicChannelSemantics {
+ private void migrateOffsets() {
+ try (KafkaZkClient zkClient = KafkaZkClient.apply(zookeeperConnect,
+ JaasUtils.isZkSecurityEnabled(), ZK_SESSION_TIMEOUT,
ZK_CONNECTION_TIMEOUT, 10,
+- Time.SYSTEM, "kafka.server", "SessionExpireListener");
++ Time.SYSTEM, "kafka.server", "SessionExpireListener",
scala.Option.empty());
+ KafkaConsumer<String, byte[]> consumer = new
KafkaConsumer<>(consumerProps)) {
+ Map<TopicPartition, OffsetAndMetadata> kafkaOffsets =
getKafkaOffsets(consumer);
+ if (!kafkaOffsets.isEmpty()) {
+diff --git
a/flume-ng-channels/flume-kafka-channel/src/test/java/org/apache/flume/channel/kafka/TestOffsetsAndMigration.java
b/flume-ng-channels/flume-kafka-channel/src/test/java/org/apache/flume/channel/kafka/TestOffsetsAndMigration.java
+index 2362c0d..cf98a91 100644
+---
a/flume-ng-channels/flume-kafka-channel/src/test/java/org/apache/flume/channel/kafka/TestOffsetsAndMigration.java
++++
b/flume-ng-channels/flume-kafka-channel/src/test/java/org/apache/flume/channel/kafka/TestOffsetsAndMigration.java
+@@ -144,7 +144,7 @@ public class TestOffsetsAndMigration extends
TestKafkaChannelBase {
+ if (hasZookeeperOffsets) {
+ KafkaZkClient zkClient = KafkaZkClient.apply(testUtil.getZkUrl(),
+ JaasUtils.isZkSecurityEnabled(), 30000, 30000, 10, Time.SYSTEM,
+- "kafka.server", "SessionExpireListener");
++ "kafka.server", "SessionExpireListener", scala.Option.empty());
+ zkClient.getConsumerOffset(group, new TopicPartition(topic, 0));
+ Long offset = tenthOffset + 1;
+ zkClient.setOrCreateConsumerOffset(group, new TopicPartition(topic, 0),
offset);
+diff --git
a/flume-ng-sources/flume-kafka-source/src/main/java/org/apache/flume/source/kafka/KafkaSource.java
b/flume-ng-sources/flume-kafka-source/src/main/java/org/apache/flume/source/kafka/KafkaSource.java
+index 20f7c7d..6f00b48 100644
+---
a/flume-ng-sources/flume-kafka-source/src/main/java/org/apache/flume/source/kafka/KafkaSource.java
++++
b/flume-ng-sources/flume-kafka-source/src/main/java/org/apache/flume/source/kafka/KafkaSource.java
+@@ -479,7 +479,7 @@ public class KafkaSource extends AbstractPollableSource
+ private String lookupBootstrap(String zookeeperConnect, SecurityProtocol
securityProtocol) {
+ try (KafkaZkClient zkClient = KafkaZkClient.apply(zookeeperConnect,
+ JaasUtils.isZkSecurityEnabled(), ZK_SESSION_TIMEOUT,
ZK_CONNECTION_TIMEOUT, 10,
+- Time.SYSTEM, "kafka.server", "SessionExpireListener")) {
++ Time.SYSTEM, "kafka.server", "SessionExpireListener",
scala.Option.empty())) {
+ List<Broker> brokerList =
+
JavaConverters.seqAsJavaListConverter(zkClient.getAllBrokersInCluster()).asJava();
+ List<BrokerEndPoint> endPoints = brokerList.stream()
+@@ -563,7 +563,7 @@ public class KafkaSource extends AbstractPollableSource
+ private void migrateOffsets(String topicStr) {
+ try (KafkaZkClient zkClient = KafkaZkClient.apply(zookeeperConnect,
+ JaasUtils.isZkSecurityEnabled(), ZK_SESSION_TIMEOUT,
ZK_CONNECTION_TIMEOUT, 10,
+- Time.SYSTEM, "kafka.server", "SessionExpireListener");
++ Time.SYSTEM, "kafka.server", "SessionExpireListener",
scala.Option.empty());
+ KafkaConsumer<String, byte[]> consumer = new
KafkaConsumer<>(kafkaProps)) {
+ Map<TopicPartition, OffsetAndMetadata> kafkaOffsets =
+ getKafkaOffsets(consumer, topicStr);
+diff --git
a/flume-ng-sources/flume-kafka-source/src/test/java/org/apache/flume/source/kafka/TestKafkaSource.java
b/flume-ng-sources/flume-kafka-source/src/test/java/org/apache/flume/source/kafka/TestKafkaSource.java
+index a82c972..b2643d8 100644
+---
a/flume-ng-sources/flume-kafka-source/src/test/java/org/apache/flume/source/kafka/TestKafkaSource.java
++++
b/flume-ng-sources/flume-kafka-source/src/test/java/org/apache/flume/source/kafka/TestKafkaSource.java
+@@ -878,7 +878,7 @@ public class TestKafkaSource {
+ if (hasZookeeperOffsets) {
+ KafkaZkClient zkClient =
KafkaZkClient.apply(kafkaServer.getZkConnectString(),
+ JaasUtils.isZkSecurityEnabled(), 30000, 30000, 10, Time.SYSTEM,
+- "kafka.server", "SessionExpireListener");
++ "kafka.server", "SessionExpireListener", scala.Option.empty());
+ zkClient.getConsumerOffset(group, new TopicPartition(topic, 0));
+ Long offset = tenthOffset + 1;
+ zkClient.setOrCreateConsumerOffset(group, new TopicPartition(topic, 0),
offset);
diff --git a/bigtop.bom b/bigtop.bom
index 0b59263..025fdfe 100644
--- a/bigtop.bom
+++ b/bigtop.bom
@@ -235,7 +235,7 @@ bigtop {
'flume' {
name = 'flume'
relNotes = 'Apache Flume'
- version { base = '1.8.0'; pkg = base; release = 1 }
+ version { base = '1.9.0'; pkg = base; release = 1 }
tarball { destination = "apache-$name-${version.base}-src.tar.gz"
source = destination }
url { download_path = "/$name/${version.base}/"
