You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@samza.apache.org by ca...@apache.org on 2020/03/19 21:01:41 UTC
[samza-hello-samza] branch master updated: Merge latest branch into
master and set version to 1.4.0 (#77)
This is an automated email from the ASF dual-hosted git repository.
cameronlee pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/samza-hello-samza.git
The following commit(s) were added to refs/heads/master by this push:
new d962f50 Merge latest branch into master and set version to 1.4.0 (#77)
d962f50 is described below
commit d962f503b2e5c1c1832cbb8eb51c0914d7034149
Author: Cameron Lee <ca...@linkedin.com>
AuthorDate: Thu Mar 19 14:01:35 2020 -0700
Merge latest branch into master and set version to 1.4.0 (#77)
* Latest changes for 1.1.0
* Bump version
* Bump guava version
* Updated latest branch to work with samza master (1.1.1-SNAPSHOT)
* [CHERRY-PICK] Kafka Download URL correction and version bump for ZK & Kafka
* Making script work on RHEL-7/Linux because of https://arfore.com/2015/10/05/rhel7-and-ncat-changes/
* Updating after Bharath's scala 2.11 change, upgrading YARN version to match samza's yarn version
* Fixed build failure due to changes in Samza project (#63)
* SAMZA-2218: Add a Couchbase example to samza-hello-samza (#64)
* SAMZA-2223: Update Couchbase example to use NoOpTableReadFunction
* Update dependencies to match OSS documentation
* Update POM to 1.4.0-SNAPSHOT as samza 1.3 has been published
* Upgrade gradle.properties
* SAMZA-2433: Use log4j2 in samza-hello-samza (#70)
* SAMZA-2437: Sample for producing to Azure Blob Storage (#71)
* add license for PageViewAvroRecord (#75)
* Sync latest branch with master (#76)
* fix couple of inconsistencies after merge
* remove SNAPSHOT from version
* updating bin/deploy.sh
---
.gitignore | 3 +
bin/deploy.sh | 2 +-
bin/grid | 5 +-
build.gradle | 27 ++-
gradle.properties | 4 +-
pom.xml | 132 +++++++++--
src/main/assembly/src.xml | 2 +-
src/main/config/azure-blob-application.properties | 39 ++++
.../main/config/couchbase-table-example.properties | 15 +-
.../samza/examples/azure/AzureBlobApplication.java | 141 ++++++++++++
.../examples/azure/data/PageViewAvroRecord.java | 67 ++++++
.../examples/cookbook/CouchbaseTableExample.java | 245 +++++++++++++++++++++
.../examples/cookbook/RemoteTableJoinExample.java | 5 +-
.../src/main/resources/log4j.xml | 34 ---
src/main/resources/log4j.xml | 52 -----
src/main/resources/log4j2.xml | 52 +++++
src/test/resources/log4j2-test.xml | 37 ++++
17 files changed, 724 insertions(+), 138 deletions(-)
diff --git a/.gitignore b/.gitignore
index f31af00..a80b94e 100644
--- a/.gitignore
+++ b/.gitignore
@@ -31,3 +31,6 @@ deploy
build/
.gradle/
state
+manifest.txt
+pathing.jar
+out/
diff --git a/bin/deploy.sh b/bin/deploy.sh
index a03d5bb..63a326c 100755
--- a/bin/deploy.sh
+++ b/bin/deploy.sh
@@ -23,4 +23,4 @@ base_dir=`pwd`
mvn clean package
mkdir -p $base_dir/deploy/samza
-tar -xvf $base_dir/target/hello-samza-1.2.0-dist.tar.gz -C $base_dir/deploy/samza
+tar -xvf $base_dir/target/hello-samza-1.4.0-dist.tar.gz -C $base_dir/deploy/samza
diff --git a/bin/grid b/bin/grid
index f937e53..5c590e4 100755
--- a/bin/grid
+++ b/bin/grid
@@ -36,7 +36,7 @@ COMMAND=$1
SYSTEM=$2
DOWNLOAD_KAFKA=https://archive.apache.org/dist/kafka/2.1.1/kafka_2.11-2.1.1.tgz
-DOWNLOAD_YARN=https://archive.apache.org/dist/hadoop/common/hadoop-2.7.1/hadoop-2.7.1.tar.gz
+DOWNLOAD_YARN=https://archive.apache.org/dist/hadoop/common/hadoop-2.9.2/hadoop-2.9.2.tar.gz
DOWNLOAD_ZOOKEEPER=https://archive.apache.org/dist/zookeeper/zookeeper-3.4.14/zookeeper-3.4.14.tar.gz
SERVICE_WAIT_TIMEOUT_SEC=20
@@ -79,6 +79,7 @@ install_all_without_yarn() {
}
install_samza() {
+ echo "Building samza from master..."
mkdir -p "$DEPLOY_ROOT_DIR"
if [ -d "$DOWNLOAD_CACHE_DIR/samza/.git" ]; then
pushd "$DOWNLOAD_CACHE_DIR/samza"
@@ -102,7 +103,7 @@ install_zookeeper() {
install_yarn() {
mkdir -p "$DEPLOY_ROOT_DIR"
- install yarn $DOWNLOAD_YARN hadoop-2.7.1
+ install yarn $DOWNLOAD_YARN hadoop-2.9.2
cp "$BASE_DIR/conf/yarn-site.xml" "$DEPLOY_ROOT_DIR/yarn/etc/hadoop/yarn-site.xml"
if [ ! -f "$HOME/.samza/conf/yarn-site.xml" ]; then
mkdir -p "$HOME/.samza/conf"
diff --git a/build.gradle b/build.gradle
index 7aba4c8..fa3a063 100644
--- a/build.gradle
+++ b/build.gradle
@@ -49,28 +49,33 @@ configurations {
explode
}
+configurations.all {
+ // using log4j1 to log4j2 bridge so need to exclude log4j1
+ exclude group: 'log4j', module: 'log4j'
+ // exclude all other slf4j bindings that are transitively pulled in
+ exclude group: 'org.slf4j', module: 'slf4j-log4j12'
+}
+
dependencies {
compile(group: 'org.codehaus.jackson', name: 'jackson-jaxrs', version: '1.9.13')
compile(group: 'org.slf4j', name: 'slf4j-api', version: "$SLF4J_VERSION")
- compile(group: 'org.slf4j', name: 'slf4j-log4j12', version: "$SLF4J_VERSION")
compile(group: 'org.schwering', name: 'irclib', version: '1.10')
compile(group: 'org.apache.samza', name: 'samza-api', version: "$SAMZA_VERSION")
- compile(group: 'org.apache.samza', name: 'samza-kv_2.11', version: "$SAMZA_VERSION")
-
- compile(group: 'org.apache.samza', name: 'samza-test_2.11', version: "$SAMZA_VERSION")
+ compile(group: 'org.apache.samza', name: 'samza-aws_2.11', version: "$SAMZA_VERSION")
+ compile(group: 'org.apache.samza', name: 'samza-azure_2.11', version: "$SAMZA_VERSION")
compile(group: 'org.apache.samza', name: 'samza-kafka_2.11', version: "$SAMZA_VERSION")
+ compile(group: 'org.apache.samza', name: 'samza-kv_2.11', version: "$SAMZA_VERSION")
+ compile(group: 'org.apache.samza', name: 'samza-kv-couchbase_2.11', version: "$SAMZA_VERSION")
compile(group: 'org.apache.samza', name: 'samza-kv-rocksdb_2.11', version: "$SAMZA_VERSION")
- compile(group: 'org.apache.samza', name: 'samza-azure_2.11', version: "$SAMZA_VERSION")
- compile(group: 'org.apache.samza', name: 'samza-aws_2.11', version: "$SAMZA_VERSION")
- testCompile(group: 'junit', name: 'junit', version: "4.12")
- explode (group: 'org.apache.samza', name: 'samza-shell', ext: 'tgz', classifier: 'dist', version: "$SAMZA_VERSION")
+ explode(group: 'org.apache.samza', name: 'samza-shell', ext: 'tgz', classifier: 'dist', version: "$SAMZA_VERSION")
runtime(group: 'org.apache.samza', name: 'samza-core_2.11', version: "$SAMZA_VERSION")
- runtime(group: 'org.apache.samza', name: 'samza-log4j_2.11', version: "$SAMZA_VERSION")
-
+ runtime(group: 'org.apache.samza', name: 'samza-log4j2_2.11', version: "$SAMZA_VERSION")
runtime(group: 'org.apache.samza', name: 'samza-shell', version: "$SAMZA_VERSION")
runtime(group: 'org.apache.samza', name: 'samza-yarn_2.11', version: "$SAMZA_VERSION")
runtime(group: 'org.apache.kafka', name: 'kafka_2.11', version: "$KAFKA_VERSION")
runtime(group: 'org.apache.hadoop', name: 'hadoop-hdfs', version: "$HADOOP_VERSION")
+ testCompile(group: 'org.apache.samza', name: 'samza-test_2.11', version: "$SAMZA_VERSION")
+ testCompile(group: 'junit', name: 'junit', version: "4.12")
}
// make the samza distribution .tgz file
@@ -103,7 +108,7 @@ task distTar(dependsOn: build, type: Tar) {
from configurations.runtime
from configurations.runtime.artifacts.files
from("src/main/resources/") {
- include "log4j.xml"
+ include "log4j2.xml"
}
}
}
diff --git a/gradle.properties b/gradle.properties
index 283cc4a..f6a27cc 100644
--- a/gradle.properties
+++ b/gradle.properties
@@ -17,9 +17,9 @@
* under the License.
*/
-SAMZA_VERSION=1.2.0
+SAMZA_VERSION=1.4.0
KAFKA_VERSION=0.11.0.2
-HADOOP_VERSION=2.6.1
+HADOOP_VERSION=2.7.1
SLF4J_VERSION = 1.7.7
diff --git a/pom.xml b/pom.xml
index bef75f0..0ec204f 100644
--- a/pom.xml
+++ b/pom.xml
@@ -27,7 +27,7 @@ under the License.
<groupId>org.apache.samza</groupId>
<artifactId>hello-samza</artifactId>
- <version>1.2.0</version>
+ <version>1.4.0</version>
<packaging>jar</packaging>
<name>Samza Example</name>
<description>
@@ -61,6 +61,17 @@ under the License.
<groupId>org.apache.samza</groupId>
<artifactId>samza-core_2.11</artifactId>
<version>${samza.version}</version>
+ <exclusions>
+ <!-- zookeeper dependency pulls in log4j1; exclude log4j1 components since we want to use log4j2 components -->
+ <exclusion>
+ <groupId>log4j</groupId>
+ <artifactId>log4j</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-log4j12</artifactId>
+ </exclusion>
+ </exclusions>
</dependency>
<dependency>
<groupId>org.apache.samza</groupId>
@@ -95,14 +106,8 @@ under the License.
</dependency>
<dependency>
<groupId>org.apache.samza</groupId>
- <artifactId>samza-log4j_2.11</artifactId>
+ <artifactId>samza-log4j2_2.11</artifactId>
<version>${samza.version}</version>
- <exclusions>
- <exclusion>
- <groupId>org.apache.samza</groupId>
- <artifactId>samza-core_2.12</artifactId>
- </exclusion>
- </exclusions>
</dependency>
<dependency>
<groupId>org.apache.samza</groupId>
@@ -115,6 +120,17 @@ under the License.
<groupId>org.apache.samza</groupId>
<artifactId>samza-yarn_2.11</artifactId>
<version>${samza.version}</version>
+ <exclusions>
+ <!-- zookeeper dependency pulls in log4j1; exclude log4j1 components since we want to use log4j2 components -->
+ <exclusion>
+ <groupId>log4j</groupId>
+ <artifactId>log4j</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-log4j12</artifactId>
+ </exclusion>
+ </exclusions>
</dependency>
<dependency>
<groupId>org.apache.samza</groupId>
@@ -137,11 +153,6 @@ under the License.
<version>${samza.version}</version>
</dependency>
<dependency>
- <groupId>org.apache.kafka</groupId>
- <artifactId>kafka_2.11</artifactId>
- <version>0.11.0.2</version>
- </dependency>
- <dependency>
<groupId>org.schwering</groupId>
<artifactId>irclib</artifactId>
<version>1.10</version>
@@ -152,54 +163,126 @@ under the License.
<version>1.6.2</version>
</dependency>
<dependency>
- <groupId>org.slf4j</groupId>
- <artifactId>slf4j-log4j12</artifactId>
- <version>1.6.2</version>
- </dependency>
- <dependency>
<groupId>org.codehaus.jackson</groupId>
<artifactId>jackson-jaxrs</artifactId>
<version>1.9.13</version>
</dependency>
<dependency>
+ <groupId>org.apache.httpcomponents</groupId>
+ <artifactId>httpcore</artifactId>
+ <version>4.4.1</version>
+ </dependency>
+ <dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-annotations</artifactId>
<version>${hadoop.version}</version>
+ <exclusions>
+ <!-- exclude log4j1 components since we want to use log4j2 components -->
+ <exclusion>
+ <groupId>log4j</groupId>
+ <artifactId>log4j</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-log4j12</artifactId>
+ </exclusion>
+ </exclusions>
</dependency>
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-auth</artifactId>
<version>${hadoop.version}</version>
+ <exclusions>
+ <!-- exclude log4j1 components since we want to use log4j2 components -->
+ <exclusion>
+ <groupId>log4j</groupId>
+ <artifactId>log4j</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-log4j12</artifactId>
+ </exclusion>
+ </exclusions>
</dependency>
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-common</artifactId>
<version>${hadoop.version}</version>
- </dependency>
- <dependency>
- <groupId>org.apache.httpcomponents</groupId>
- <artifactId>httpcore</artifactId>
- <version>4.4.1</version>
+ <exclusions>
+ <!-- exclude log4j1 components since we want to use log4j2 components -->
+ <exclusion>
+ <groupId>log4j</groupId>
+ <artifactId>log4j</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-log4j12</artifactId>
+ </exclusion>
+ </exclusions>
</dependency>
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-hdfs</artifactId>
<version>${hadoop.version}</version>
+ <exclusions>
+ <!-- exclude log4j1 components since we want to use log4j2 components -->
+ <exclusion>
+ <groupId>log4j</groupId>
+ <artifactId>log4j</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-log4j12</artifactId>
+ </exclusion>
+ </exclusions>
</dependency>
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-yarn-api</artifactId>
<version>${hadoop.version}</version>
+ <exclusions>
+ <!-- exclude log4j1 components since we want to use log4j2 components -->
+ <exclusion>
+ <groupId>log4j</groupId>
+ <artifactId>log4j</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-log4j12</artifactId>
+ </exclusion>
+ </exclusions>
</dependency>
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-yarn-client</artifactId>
<version>${hadoop.version}</version>
+ <exclusions>
+ <!-- exclude log4j1 components since we want to use log4j2 components -->
+ <exclusion>
+ <groupId>log4j</groupId>
+ <artifactId>log4j</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-log4j12</artifactId>
+ </exclusion>
+ </exclusions>
</dependency>
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-yarn-common</artifactId>
<version>${hadoop.version}</version>
+ <exclusions>
+ <!-- exclude log4j1 components since we want to use log4j2 components -->
+ <exclusion>
+ <groupId>log4j</groupId>
+ <artifactId>log4j</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-log4j12</artifactId>
+ </exclusion>
+ </exclusions>
</dependency>
<dependency>
<groupId>com.google.guava</groupId>
@@ -211,7 +294,7 @@ under the License.
<properties>
<!-- maven specific properties -->
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
- <samza.version>1.2.0</samza.version>
+ <samza.version>1.4.0</samza.version>
<hadoop.version>2.6.1</hadoop.version>
</properties>
@@ -289,6 +372,7 @@ under the License.
<exclude>.git/**</exclude>
<exclude>*.md</exclude>
<exclude>docs/**</exclude>
+ <exclude>out/**</exclude>
<exclude>config/**</exclude>
<exclude>bin/**</exclude>
<exclude>.gitignore</exclude>
diff --git a/src/main/assembly/src.xml b/src/main/assembly/src.xml
index 1614aaf..5fac44a 100644
--- a/src/main/assembly/src.xml
+++ b/src/main/assembly/src.xml
@@ -41,7 +41,7 @@
</fileSets>
<files>
<file>
- <source>${basedir}/src/main/resources/log4j.xml</source>
+ <source>${basedir}/src/main/resources/log4j2.xml</source>
<outputDirectory>lib</outputDirectory>
</file>
<file>
diff --git a/src/main/config/azure-blob-application.properties b/src/main/config/azure-blob-application.properties
new file mode 100644
index 0000000..a13535f
--- /dev/null
+++ b/src/main/config/azure-blob-application.properties
@@ -0,0 +1,39 @@
+# 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.
+
+# Instructions to run this samza job are given in {@link samza.examples.azure.AzureBlobApplication}
+
+# "azure-blob-container" creates an Azure Container (if it doesnt already exist) in Azure Storage Account.
+# For valid container names follow guidance in https://docs.microsoft.com/en-us/rest/api/storageservices/naming-and-referencing-containers--blobs--and-metadata#container-names
+
+# Job
+job.factory.class=org.apache.samza.job.yarn.YarnJobFactory
+job.name=azure-blob
+
+# YARN package path
+yarn.package.path=file://${basedir}/target/${project.artifactId}-${pom.version}-dist.tar.gz
+
+# StreamApplication class
+app.class=samza.examples.azure.AzureBlobApplication
+
+#Azure blob essential configs
+systems.azure-blob-container.samza.factory=org.apache.samza.system.azureblob.AzureBlobSystemFactory
+sensitive.systems.azure-blob-container.azureblob.account.name=your-azure-storage-account-name
+sensitive.systems.azure-blob-container.azureblob.account.key=your-azure-storage-account-key
+
+#Azure blob config - to created a blob per 2 input kafka messages
+systems.azure-blob-container.azureblob.maxMessagesPerBlob=2
diff --git a/bin/deploy.sh b/src/main/config/couchbase-table-example.properties
old mode 100755
new mode 100644
similarity index 73%
copy from bin/deploy.sh
copy to src/main/config/couchbase-table-example.properties
index a03d5bb..0bbfdea
--- a/bin/deploy.sh
+++ b/src/main/config/couchbase-table-example.properties
@@ -1,4 +1,3 @@
-#!/bin/bash
# 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
@@ -16,11 +15,11 @@
# specific language governing permissions and limitations
# under the License.
-home_dir=`pwd`
-base_dir=$(dirname $0)/..
-cd $base_dir
-base_dir=`pwd`
+# Application / Job
+app.class=samza.examples.cookbook.CouchbaseTableExample
+job.factory.class=org.apache.samza.job.yarn.YarnJobFactory
+job.name=couchbase-table-example
+job.container.count=2
-mvn clean package
-mkdir -p $base_dir/deploy/samza
-tar -xvf $base_dir/target/hello-samza-1.2.0-dist.tar.gz -C $base_dir/deploy/samza
+# YARN
+yarn.package.path=file://${basedir}/target/${project.artifactId}-${pom.version}-dist.tar.gz
diff --git a/src/main/java/samza/examples/azure/AzureBlobApplication.java b/src/main/java/samza/examples/azure/AzureBlobApplication.java
new file mode 100644
index 0000000..15de822
--- /dev/null
+++ b/src/main/java/samza/examples/azure/AzureBlobApplication.java
@@ -0,0 +1,141 @@
+/*
+ * 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 samza.examples.azure;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import java.util.List;
+import java.util.Map;
+import org.apache.samza.application.StreamApplication;
+import org.apache.samza.application.descriptors.StreamApplicationDescriptor;
+import org.apache.samza.operators.MessageStream;
+import org.apache.samza.operators.OutputStream;
+import org.apache.samza.serializers.JsonSerdeV2;
+import org.apache.samza.serializers.NoOpSerde;
+import org.apache.samza.system.descriptors.GenericOutputDescriptor;
+import org.apache.samza.system.descriptors.GenericSystemDescriptor;
+import org.apache.samza.system.kafka.descriptors.KafkaInputDescriptor;
+import org.apache.samza.system.kafka.descriptors.KafkaSystemDescriptor;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import samza.examples.azure.data.PageViewAvroRecord;
+import samza.examples.cookbook.data.PageView;
+import samza.examples.wikipedia.application.WikipediaApplication;
+
+
+/**
+ * In this example, we demonstrate sending blobs to Azure Blob Storage.
+ * This Samza job reads from Kafka topic "page-view-azure-blob-input" and produces blobs to Azure-Container "azure-blob-container" in your Azure Storage account.
+ *
+ * Currently, Samza supports sending Avro files are blobs.
+ * Hence the incoming messages into the Samza job have to be converted to an Avro record.
+ * For this job, we use input message as {@link samza.examples.cookbook.data.PageView} and
+ * covert it to an Avro record defined as {@link samza.examples.azure.data.PageViewAvroRecord}.
+ *
+ * To run the below example:
+ *
+ * <ol>
+ * <li>
+ * Replace your-azure-storage-account-name and your-azure-storage-account-key with details of your Azure Storage Account.
+ * </li>
+ * <li>
+ * Ensure that the topic "page-view-azure-blob-input" is created <br/>
+ * ./deploy/kafka/bin/kafka-topics.sh --zookeeper localhost:2181 --create --topic page-view-azure-blob-input --partitions 1 --replication-factor 1
+ * </li>
+ * <li>
+ * Run the application using the run-app.sh script <br/>
+ * ./deploy/samza/bin/run-app.sh --config-factory=org.apache.samza.config.factories.PropertiesConfigFactory --config-path=file://$PWD/deploy/samza/config/azure-blob-application.properties
+ * </li>
+ * <li>
+ * Produce some messages to the "page-view-azure-blob-input" topic <br/>
+ * ./deploy/kafka/bin/kafka-console-producer.sh --topic page-view-azure-blob-input --broker-list localhost:9092 <br/>
+ * {"userId": "user1", "country": "india", "pageId":"google.com"} <br/>
+ * {"userId": "user2", "country": "france", "pageId":"facebook.com"} <br/>
+ * {"userId": "user3", "country": "china", "pageId":"yahoo.com"} <br/>
+ * {"userId": "user4", "country": "italy", "pageId":"linkedin.com"} <br/>
+ * {"userId": "user5", "country": "germany", "pageId":"amazon.com"} <br/>
+ * {"userId": "user6", "country": "denmark", "pageId":"apple.com"} <br/>
+ * </li>
+ * <li>
+ * Seeing Output:
+ * <ol>
+ * <li>
+ * See blobs in your Azure portal at https://<azure-storage-account-name>.blob.core.windows.net/azure-blob-container/PageViewEventStream/<time-stamp>.avro
+ * </li>
+ * <li>
+ * system-name "azure-blob-container" in configs and code below maps to Azure-Container in Azure Storage account.
+ * </li>
+ * <li>
+ * <time-stamp> is of the format yyyy/MM/dd/HH/mm-ss-randomString.avro. Hence navigate through the virtual folders on the portal to see your blobs.
+ * </li>
+ * <li>
+ * Due to network calls, allow a few minutes for blobs to appear on the portal.
+ * </li>
+ * <li>
+ * Config "maxMessagesPerBlob=2" ensures that a blob is created per 2 input messages. Adjust input or config accordingly.
+ * </li>
+ * </ol>
+ * </li>
+ * </ol>
+ */
+public class AzureBlobApplication implements StreamApplication {
+ private static final Logger LOG = LoggerFactory.getLogger(AzureBlobApplication.class);
+
+ private static final List<String> KAFKA_CONSUMER_ZK_CONNECT = ImmutableList.of("localhost:2181");
+ private static final List<String> KAFKA_PRODUCER_BOOTSTRAP_SERVERS = ImmutableList.of("localhost:9092");
+ private static final Map<String, String> KAFKA_DEFAULT_STREAM_CONFIGS = ImmutableMap.of("replication.factor", "1");
+ private static final String INPUT_PAGEVIEW_STREAM_ID = "page-view-azure-blob-input";
+ private static final String OUTPUT_SYSTEM = "azure-blob-container";
+ private static final String OUTPUT_STREAM = "PageViewEventStream";
+
+ @Override
+ public void describe(StreamApplicationDescriptor appDescriptor) {
+ // Define a system descriptor for Kafka
+ KafkaSystemDescriptor kafkaSystemDescriptor =
+ new KafkaSystemDescriptor("kafka").withConsumerZkConnect(KAFKA_CONSUMER_ZK_CONNECT)
+ .withProducerBootstrapServers(KAFKA_PRODUCER_BOOTSTRAP_SERVERS)
+ .withDefaultStreamConfigs(KAFKA_DEFAULT_STREAM_CONFIGS);
+
+ KafkaInputDescriptor<PageView> pageViewInputDescriptor =
+ kafkaSystemDescriptor.getInputDescriptor(INPUT_PAGEVIEW_STREAM_ID, new JsonSerdeV2<>(PageView.class));
+
+ // Define a system descriptor for Azure Blob Storage
+ GenericSystemDescriptor azureBlobSystemDescriptor =
+ new GenericSystemDescriptor(OUTPUT_SYSTEM, "org.apache.samza.system.azureblob.AzureBlobSystemFactory");
+
+ GenericOutputDescriptor<PageViewAvroRecord> azureBlobOuputDescriptor =
+ azureBlobSystemDescriptor.getOutputDescriptor(OUTPUT_STREAM, new NoOpSerde<>());
+
+ // Set Kafka as the default system for the job
+ appDescriptor.withDefaultSystem(kafkaSystemDescriptor);
+
+ // Define the input and output streams with descriptors
+ MessageStream<PageView> pageViewInput = appDescriptor.getInputStream(pageViewInputDescriptor);
+ OutputStream<PageViewAvroRecord> pageViewAvroRecordOutputStream = appDescriptor.getOutputStream(azureBlobOuputDescriptor);
+
+ // Define the execution flow with the high-level API
+ pageViewInput
+ .map((message) -> {
+ LOG.info("Sending: Received PageViewEvent with pageId: " + message.pageId);
+ return PageViewAvroRecord.buildPageViewRecord(message);
+ })
+ .sendTo(pageViewAvroRecordOutputStream);
+ }
+}
diff --git a/src/main/java/samza/examples/azure/data/PageViewAvroRecord.java b/src/main/java/samza/examples/azure/data/PageViewAvroRecord.java
new file mode 100644
index 0000000..73f16b2
--- /dev/null
+++ b/src/main/java/samza/examples/azure/data/PageViewAvroRecord.java
@@ -0,0 +1,67 @@
+/*
+ * 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 samza.examples.azure.data;
+
+import java.io.Serializable;
+import org.apache.avro.AvroRuntimeException;
+import samza.examples.cookbook.data.PageView;
+
+public class PageViewAvroRecord extends org.apache.avro.specific.SpecificRecordBase
+ implements org.apache.avro.specific.SpecificRecord, Serializable {
+ public final org.apache.avro.Schema SCHEMA = org.apache.avro.Schema.parse(
+ "{\"type\":\"record\",\"name\":\"PageViewAvroRecord\",\"namespace\":\"org.apache.samza.examples.events\", \"fields\":[{\"name\": \"userId\", \"type\": \"string\"}, {\"name\": \"country\", \"type\": \"string\"}, {\"name\": \"pageId\", \"type\": \"string\"}]}");
+
+ private String userId;
+ private String country;
+ private String pageId;
+
+ public static PageViewAvroRecord buildPageViewRecord(PageView pageView) {
+ PageViewAvroRecord record = new PageViewAvroRecord();
+ record.userId = pageView.userId;
+ record.country = pageView.country;
+ record.pageId = pageView.pageId;
+ return record;
+ }
+ public org.apache.avro.Schema getSchema() {
+ return SCHEMA;
+ }
+
+ public java.lang.Object get(int field) {
+ switch (field) {
+ case 0: return userId;
+ case 1: return country;
+ case 2: return pageId;
+ default: throw new AvroRuntimeException("bad index");
+ }
+ }
+
+ public void put(int field, Object value) {
+ switch (field) {
+ case 0:
+ userId = (String) value; break;
+ case 1:
+ country = (String) value; break;
+ case 2:
+ pageId = (String) value; break;
+ default:
+ throw new AvroRuntimeException("bad index");
+ }
+ }
+}
diff --git a/src/main/java/samza/examples/cookbook/CouchbaseTableExample.java b/src/main/java/samza/examples/cookbook/CouchbaseTableExample.java
new file mode 100644
index 0000000..a6b2dee
--- /dev/null
+++ b/src/main/java/samza/examples/cookbook/CouchbaseTableExample.java
@@ -0,0 +1,245 @@
+/*
+ * 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 samza.examples.cookbook;
+
+import com.couchbase.client.java.document.json.JsonObject;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import java.text.SimpleDateFormat;
+import java.time.Duration;
+import java.util.Arrays;
+import java.util.Date;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import org.apache.samza.SamzaException;
+import org.apache.samza.application.StreamApplication;
+import org.apache.samza.application.descriptors.StreamApplicationDescriptor;
+import org.apache.samza.context.Context;
+import org.apache.samza.operators.MessageStream;
+import org.apache.samza.operators.OutputStream;
+import org.apache.samza.operators.functions.MapFunction;
+import org.apache.samza.serializers.StringSerde;
+import org.apache.samza.system.kafka.descriptors.KafkaInputDescriptor;
+import org.apache.samza.system.kafka.descriptors.KafkaOutputDescriptor;
+import org.apache.samza.system.kafka.descriptors.KafkaSystemDescriptor;
+import org.apache.samza.table.descriptors.RemoteTableDescriptor;
+import org.apache.samza.table.remote.NoOpTableReadFunction;
+import org.apache.samza.table.remote.RemoteTable;
+import org.apache.samza.table.remote.couchbase.CouchbaseTableWriteFunction;
+import org.apache.samza.table.retry.TableRetryPolicy;
+
+
+/**
+ * This is a simple word count example using a remote store.
+ *
+ * In this example, we use Couchbase to demonstrate how to invoke API's on a remote store other than get, put or delete
+ * as defined in {@link org.apache.samza.table.remote.AsyncRemoteTable}. Input messages are collected from user through
+ * a Kafka console producer, and tokenized using space. For each word, we increment a counter for this word
+ * as well as a counter for all words on Couchbase. We also output the current value of both counters to Kafka console
+ * consumer.
+ *
+ * A rate limit of 4 requests/second to Couchbase is set of the entire job, internally Samza uses an embedded
+ * rate limiter, which evenly distributes the total rate limit among tasks. As we invoke 2 calls on Couchbase
+ * for each word, you should see roughly 2 messages per second in the Kafka console consumer
+ * window.
+ *
+ * A retry policy with 1 second fixed backoff time and max 3 retries is attached to the remote table.
+ *
+ * <p> Concepts covered: remote table, rate limiter, retry, arbitrary operation on remote store.
+ *
+ * To run the below example:
+ *
+ * <ol>
+ * <li>
+ * Create a Couchbase instance using docker; Log into the admin UI at http://localhost:8091 (Administrator/password) <br/>
+ * create a bucket called "my-bucket" <br/>
+ * Under Security tab, create a user with the same name, set 123456 as the password, and give it "Data Reader"
+ * and "Data Writer" privilege for this bucket. <br/>
+ * More information can be found at https://docs.couchbase.com/server/current/getting-started/do-a-quick-install.html
+ * </li>
+ * <li>
+ * Create Kafka topics "word-input" and "count-output" <br/>
+ * ./deploy/kafka/bin/kafka-topics.sh --zookeeper localhost:2181 --create --topic word-input --partitions 2 --replication-factor 1
+ * ./deploy/kafka/bin/kafka-topics.sh --zookeeper localhost:2181 --create --topic count-output --partitions 2 --replication-factor 1
+ * </li>
+ * <li>
+ * Run the application using the run-app.sh script <br/>
+ * ./deploy/samza/bin/run-app.sh --config-factory=org.apache.samza.config.factories.PropertiesConfigFactory --config-path=file://$PWD/deploy/samza/config/couchbase-table-example.properties
+ * </li>
+ * <li>
+ * Consume messages from the output topic <br/>
+ * ./deploy/kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic count-output
+ * </li>
+ * <li>
+ * Produce some messages to the input topic <br/>
+ * ./deploy/kafka/bin/kafka-console-producer.sh --topic word-input --broker-list localhost:9092
+ *
+ * After the console producer is started, type
+ * 1
+ * 2
+ * 3
+ * 4
+ * 5
+ * 4
+ * 3
+ * 2
+ * 1
+ *
+ * You should see messages like below from the console consumer window
+ *
+ * 2019-05-23 21:18:07 2019-05-23 21:18:07 word=2, count=1, total-count=1
+ * 2019-05-23 21:18:07 2019-05-23 21:18:07 word=1, count=1, total-count=2
+ * 2019-05-23 21:18:07 2019-05-23 21:18:07 word=4, count=1, total-count=3
+ * 2019-05-23 21:18:07 2019-05-23 21:18:07 word=3, count=1, total-count=4
+ * 2019-05-23 21:18:08 2019-05-23 21:18:08 word=4, count=2, total-count=5
+ * 2019-05-23 21:18:08 2019-05-23 21:18:08 word=5, count=1, total-count=6
+ * 2019-05-23 21:18:09 2019-05-23 21:18:09 word=2, count=2, total-count=7
+ * 2019-05-23 21:18:09 2019-05-23 21:18:09 word=3, count=2, total-count=8
+ * 2019-05-23 21:18:10 2019-05-23 21:18:10 word=1, count=2, total-count=9
+ *
+ * You can examine the result on Couchbase Admin GUI as well.
+ *
+ * Note:
+ * - If you enter "1 2 3 4 5 4 3 2 1", you should see roughly 1 QPS as
+ * the input is processed by only one task
+ *
+ *
+ * </li>
+ * </ol>
+ *
+ */
+public class CouchbaseTableExample implements StreamApplication {
+
+ private static final SimpleDateFormat DATE_FORMAT = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
+
+ private static final String KAFKA_SYSTEM_NAME = "kafka";
+ private static final List<String> KAFKA_CONSUMER_ZK_CONNECT = ImmutableList.of("localhost:2181");
+ private static final List<String> KAFKA_PRODUCER_BOOTSTRAP_SERVERS = ImmutableList.of("localhost:9092");
+ private static final Map<String, String> KAFKA_DEFAULT_STREAM_CONFIGS = ImmutableMap.of("replication.factor", "1");
+
+ private static final String INPUT_STREAM_ID = "word-input";
+ private static final String OUTPUT_STREAM_ID = "count-output";
+
+ private static final String CLUSTER_NODES = "couchbase://127.0.0.1";
+ private static final int COUCHBASE_PORT = 11210;
+ private static final String BUCKET_NAME = "my-bucket";
+ private static final String BUCKET_PASSWORD = "123456";
+ private static final String TOTAL_COUNT_ID = "total-count";
+
+ @Override
+ public void describe(StreamApplicationDescriptor app) {
+
+ KafkaSystemDescriptor kafkaSystemDescriptor = new KafkaSystemDescriptor(KAFKA_SYSTEM_NAME)
+ .withConsumerZkConnect(KAFKA_CONSUMER_ZK_CONNECT)
+ .withProducerBootstrapServers(KAFKA_PRODUCER_BOOTSTRAP_SERVERS)
+ .withDefaultStreamConfigs(KAFKA_DEFAULT_STREAM_CONFIGS);
+
+ KafkaInputDescriptor<String> wordInputDescriptor =
+ kafkaSystemDescriptor.getInputDescriptor(INPUT_STREAM_ID, new StringSerde());
+
+ KafkaOutputDescriptor<String> countOutputDescriptor =
+ kafkaSystemDescriptor.getOutputDescriptor(OUTPUT_STREAM_ID, new StringSerde());
+
+ MyCouchbaseTableWriteFunction writeFn = new MyCouchbaseTableWriteFunction(BUCKET_NAME, CLUSTER_NODES)
+ .withBootstrapCarrierDirectPort(COUCHBASE_PORT)
+ .withUsernameAndPassword(BUCKET_NAME, BUCKET_PASSWORD)
+ .withTimeout(Duration.ofSeconds(5));
+
+ TableRetryPolicy retryPolicy = new TableRetryPolicy()
+ .withFixedBackoff(Duration.ofSeconds(1))
+ .withStopAfterAttempts(3);
+
+ RemoteTableDescriptor couchbaseTableDescriptor = new RemoteTableDescriptor("couchbase-table")
+ .withReadFunction(new NoOpTableReadFunction())
+ .withReadRateLimiterDisabled()
+ .withWriteFunction(writeFn)
+ .withWriteRetryPolicy(retryPolicy)
+ .withWriteRateLimit(4);
+
+ app.withDefaultSystem(kafkaSystemDescriptor);
+ MessageStream<String> wordStream = app.getInputStream(wordInputDescriptor);
+ OutputStream<String> countStream = app.getOutputStream(countOutputDescriptor);
+ app.getTable(couchbaseTableDescriptor);
+
+ wordStream
+ .flatMap(m -> Arrays.asList(m.split(" ")))
+ .filter(word -> word != null && word.length() > 0)
+ .map(new MyCountFunction())
+ .map(countString -> currentTime() + " " + countString)
+ .sendTo(countStream);
+ }
+
+ static class MyCountFunction implements MapFunction<String, String> {
+
+ private MyCouchbaseTableWriteFunction writeFn;
+
+ @Override
+ public void init(Context context) {
+ RemoteTable table = (RemoteTable) context.getTaskContext().getTable("couchbase-table");
+ writeFn = (MyCouchbaseTableWriteFunction) table.getWriteFunction();
+ }
+
+ @Override
+ public String apply(String word) {
+ CompletableFuture<Long> countFuture = writeFn.incCounter(word);
+ CompletableFuture<Long> totalCountFuture = writeFn.incCounter(TOTAL_COUNT_ID);
+ return String.format("%s word=%s, count=%d, total-count=%d",
+ currentTime(), word, countFuture.join(), totalCountFuture.join());
+ }
+ }
+
+ static class MyCouchbaseTableWriteFunction extends CouchbaseTableWriteFunction<JsonObject> {
+
+ private final static int OP_COUNTER = 1;
+
+ public MyCouchbaseTableWriteFunction(String bucketName, String... clusterNodes) {
+ super(bucketName, JsonObject.class, clusterNodes);
+ }
+
+ @Override
+ public <T> CompletableFuture<T> writeAsync(int opId, Object... args) {
+ switch (opId) {
+ case OP_COUNTER:
+ Preconditions.checkArgument(2 == args.length,
+ String.format("Two arguments (String and int) are expected for counter operation (opId=%d)", opId));
+ String id = (String) args[0];
+ int delta = (int) args[1];
+ return asyncWriteHelper(
+ bucket.async().counter(id, delta, 1, timeout.toMillis(), TimeUnit.MILLISECONDS),
+ String.format("Failed to invoke counter with Id %s from bucket %s.", id, bucketName),
+ false);
+ default:
+ throw new SamzaException("Unknown opId: " + opId);
+ }
+ }
+
+ public CompletableFuture<Long> incCounter(String id) {
+ return table.writeAsync(OP_COUNTER, id, 1);
+ }
+
+ }
+
+ private static String currentTime() {
+ return DATE_FORMAT.format(new Date());
+ }
+
+}
diff --git a/src/main/java/samza/examples/cookbook/RemoteTableJoinExample.java b/src/main/java/samza/examples/cookbook/RemoteTableJoinExample.java
index 066f37c..b21a48a 100644
--- a/src/main/java/samza/examples/cookbook/RemoteTableJoinExample.java
+++ b/src/main/java/samza/examples/cookbook/RemoteTableJoinExample.java
@@ -30,7 +30,6 @@ import java.util.concurrent.CompletableFuture;
import org.apache.samza.SamzaException;
import org.apache.samza.application.StreamApplication;
import org.apache.samza.application.descriptors.StreamApplicationDescriptor;
-import org.apache.samza.context.Context;
import org.apache.samza.operators.KV;
import org.apache.samza.operators.MessageStream;
import org.apache.samza.operators.OutputStream;
@@ -40,7 +39,6 @@ import org.apache.samza.serializers.StringSerde;
import org.apache.samza.system.kafka.descriptors.KafkaInputDescriptor;
import org.apache.samza.system.kafka.descriptors.KafkaOutputDescriptor;
import org.apache.samza.system.kafka.descriptors.KafkaSystemDescriptor;
-import org.apache.samza.table.AsyncReadWriteTable;
import org.apache.samza.table.Table;
import org.apache.samza.table.descriptors.CachingTableDescriptor;
import org.apache.samza.table.remote.BaseTableFunction;
@@ -161,7 +159,8 @@ public class RemoteTableJoinExample implements StreamApplication {
}
}
- static class StockPriceReadFunction extends BaseTableFunction implements TableReadFunction<String, Double> {
+ static class StockPriceReadFunction extends BaseTableFunction
+ implements TableReadFunction<String, Double> {
@Override
public CompletableFuture<Double> getAsync(String symbol) {
return CompletableFuture.supplyAsync(() -> {
diff --git a/src/main/java/samza/examples/sql/samza-sql-stream-table-join/src/main/resources/log4j.xml b/src/main/java/samza/examples/sql/samza-sql-stream-table-join/src/main/resources/log4j.xml
deleted file mode 100644
index 119fefb..0000000
--- a/src/main/java/samza/examples/sql/samza-sql-stream-table-join/src/main/resources/log4j.xml
+++ /dev/null
@@ -1,34 +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.
--->
-<!DOCTYPE log4j:configuration SYSTEM "log4j.dtd">
-<log4j:configuration xmlns:log4j="http://jakarta.apache.org/log4j/">
- <appender name="RollingAppender" class="org.apache.log4j.RollingFileAppender">
- <param name="File" value="${samza.log.dir}/${samza.container.name}.log" />
- <param name="MaxFileSize" value="25MB" />
- <param name="MaxBackupIndex" value="100" />
- <layout class="org.apache.log4j.PatternLayout">
- <param name="ConversionPattern" value="%d{yyyy-MM-dd HH:mm:ss.SSS} [%t] %c{1} [%p] %m%n" />
- </layout>
- </appender>
- <root>
- <priority value="info" />
- <appender-ref ref="RollingAppender"/>
- </root>
-</log4j:configuration>
\ No newline at end of file
diff --git a/src/main/resources/log4j.xml b/src/main/resources/log4j.xml
deleted file mode 100644
index 805d5ca..0000000
--- a/src/main/resources/log4j.xml
+++ /dev/null
@@ -1,52 +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.
-
--->
-
-<!DOCTYPE log4j:configuration SYSTEM "log4j.dtd">
-<log4j:configuration xmlns:log4j="http://jakarta.apache.org/log4j/">
- <appender name="jmx" class="org.apache.samza.logging.log4j.JmxAppender" />
-
- <appender name="RollingAppender" class="org.apache.log4j.RollingFileAppender">
- <param name="File" value="${samza.log.dir}/${samza.container.name}.log" />
- <param name="MaxFileSize" value="256MB" />
- <param name="MaxBackupIndex" value="20" />
- <layout class="org.apache.log4j.PatternLayout">
- <param name="ConversionPattern" value="%d{yyyy-MM-dd HH:mm:ss.SSS} [%t] %c{1} [%p] %m%n" />
- </layout>
- </appender>
- <appender name="StartupAppender" class="org.apache.log4j.RollingFileAppender">
- <param name="File" value="${samza.log.dir}/${samza.container.name}-startup.log" />
- <param name="MaxFileSize" value="256MB" />
- <param name="MaxBackupIndex" value="1" />
- <layout class="org.apache.log4j.PatternLayout">
- <param name="ConversionPattern" value="%d{yyyy-MM-dd HH:mm:ss.SSS} [%t] %c{1} [%p] %m%n" />
- </layout>
- </appender>
- <logger name="STARTUP_LOGGER" additivity="false">
- <level value="info" />
- <appender-ref ref="StartupAppender"/>
- </logger>
- <root>
- <priority value="info" />
- <appender-ref ref="RollingAppender"/>
- <appender-ref ref="jmx" />
- </root>
-</log4j:configuration>
diff --git a/src/main/resources/log4j2.xml b/src/main/resources/log4j2.xml
new file mode 100644
index 0000000..adcdadb
--- /dev/null
+++ b/src/main/resources/log4j2.xml
@@ -0,0 +1,52 @@
+<?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.
+
+-->
+
+<Configuration>
+ <Appenders>
+ <RollingFile name="RollingFile" fileName="${sys:samza.log.dir}/${sys:samza.container.name}.log"
+ filePattern="${sys:samza.log.dir}/${sys:samza.container.name}-%d{yyyy-MM-dd}-%i.log">
+ <PatternLayout pattern="%d{yyyy-MM-dd HH:mm:ss.SSS} [%t] %c{1} [%p] %m%n"/>
+ <Policies>
+ <SizeBasedTriggeringPolicy size="256MB" />
+ </Policies>
+ <DefaultRolloverStrategy max="20"/>
+ </RollingFile>
+
+ <RollingFile name="StartupAppender" fileName="${sys:samza.log.dir}/${sys:samza.container.name}-startup.log"
+ filePattern="${sys:samza.log.dir}/${sys:samza.container.name}-startup-%d{MM-dd-yyyy}-%i.log">
+ <PatternLayout pattern="%d{yyyy-MM-dd HH:mm:ss.SSS} [%t] %c{1} [%p] %m%n"/>
+ <Policies>
+ <SizeBasedTriggeringPolicy size="256MB" />
+ </Policies>
+ <DefaultRolloverStrategy max="1"/>
+ </RollingFile>
+ </Appenders>
+
+ <Loggers>
+ <Logger name="STARTUP_LOGGER" level="info" additivity="false">
+ <AppenderRef ref="StartupAppender"/>
+ </Logger>
+ <Root level="info">
+ <AppenderRef ref="RollingFile"/>
+ </Root>
+ </Loggers>
+</Configuration>
diff --git a/src/test/resources/log4j2-test.xml b/src/test/resources/log4j2-test.xml
new file mode 100644
index 0000000..c095c00
--- /dev/null
+++ b/src/test/resources/log4j2-test.xml
@@ -0,0 +1,37 @@
+<?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.
+
+-->
+
+<!-- this is needed to avoid using src/main/resources/log4j2.xml for tests -->
+
+<Configuration>
+ <Appenders>
+ <Console name="STDOUT" target="SYSTEM_OUT">
+ <PatternLayout pattern="%d{yyyy-MM-dd HH:mm:ss.SSS} [%t] %c{1} [%p] %m%n"/>
+ </Console>
+ </Appenders>
+
+ <Loggers>
+ <Root level="info">
+ <AppenderRef ref="STDOUT"/>
+ </Root>
+ </Loggers>
+</Configuration>