samza.git
28 hours agoSAMZA-1775: add some delay before renew under transient EH exception master
Hai Lu [Fri, 20 Jul 2018 03:23:30 +0000 (20:23 -0700)] 
SAMZA-1775: add some delay before renew under transient EH exception

There is no delay at all before we renew the partition. This sometimes lead to spam in the log for the following messages:

Received transient exception from EH client. Renew partition receiver for ssp ...

Author: Hai Lu <halu@linkedin.com>

Reviewers: Jagadish <jagadish@apache.org>

Closes #571 from lhaiesp/master

33 hours agoSAMZA-1774: Support table API in low level
Aditya Toomula [Thu, 19 Jul 2018 22:06:27 +0000 (15:06 -0700)] 
SAMZA-1774: Support table API in low level

Code changes to support table in low level API.

Author: Aditya Toomula <atoomula@linkedin.com>

Reviewers: Srini P<spunuru@linkedin.com>

Closes #556 from atoomula/table1

7 days agoSAMZA-1755: Fix JsonRelConverter to recursively convert relRecords.
Aditya Toomula [Fri, 13 Jul 2018 22:26:34 +0000 (15:26 -0700)] 
SAMZA-1755: Fix JsonRelConverter to recursively convert relRecords.

Author: Aditya Toomula <atoomula@linkedin.com>

Reviewers: Srini P<spunuru@linkedin.com>

Closes #558 from atoomula/sql1

10 days agoSAMZA-1730: Adding state valiations in StreamProcessor before any lifecycle operation...
Shanthoosh Venkataraman [Tue, 10 Jul 2018 23:34:58 +0000 (16:34 -0700)] 
SAMZA-1730: Adding state valiations in StreamProcessor before any lifecycle operation and group coordination.

Author: Shanthoosh Venkataraman <santhoshvenkat1988@gmail.com>

Reviewers: Jagadish<jagadish@apache.org>

Closes #535 from shanthoosh/abced

11 days agoSAMZA-1761: Reduce runTime of TestZkUtils test from 40s to 800ms.
Shanthoosh Venkataraman [Mon, 9 Jul 2018 19:29:14 +0000 (12:29 -0700)] 
SAMZA-1761: Reduce runTime of TestZkUtils test from 40s to 800ms.

Author: Shanthoosh Venkataraman <santhoshvenkat1988@gmail.com>

Reviewers: Prateek <pmaheshw@linkedin.com>

Closes #565 from shanthoosh/SAMZA-1761

3 weeks agoSAMZA-1735: Adding ListGaugeMBean, to enable MBean validation
Ray Matharu [Fri, 29 Jun 2018 23:40:21 +0000 (16:40 -0700)] 
SAMZA-1735: Adding ListGaugeMBean, to enable MBean validation

Adding ListGaugeMBean, to enable MBean validation.
Tested with LocalContainerRunner and YARN job.

JIRA SAMZA-1733/ SAMZA-1735.

Author: Ray Matharu <rmatharu@linkedin.com>

Reviewers: Cameron Lee <calee@linkedin.com>

Closes #567 from rmatharu/ListGaugeMBean

3 weeks agoSAMZA-1758: Configuring a timeout for TestRunner to execute the SamzaJob
sanil15 [Thu, 28 Jun 2018 22:48:28 +0000 (15:48 -0700)] 
SAMZA-1758: Configuring a timeout for TestRunner to execute the SamzaJob

Author: sanil15 <sanil.jain15@gmail.com>
Author: Sanil Jain <snjain@linkedin.com>

Reviewers: Bharath Kumarasubramanian <codin.martial@gmail.com>

Closes #563 from Sanil15/SAMZA-1758 and squashes the following commits:

5b198a0f [Sanil Jain] Fixing a bug for preconditions
6d2fd334 [Sanil Jain] Adressing Review
1eb3847c [Sanil Jain] Removing explicit handling of TimeoutException and adding more docs
0a9689c9 [sanil15] Addressing Review, moving tests from SamzaFailureTests, improving doc, adding validation
b79b5628 [sanil15] Using ExceptionUtils to get full stack trace, adding more docs
dd816ff8 [sanil15] Addressing review, using waitForFinish(timeout) to configure a timeout for TestRunner, adding some Failure tests
903c1162 [sanil15] Configuring a timeout for TestRunner to execute the SamzaJob

3 weeks agoSAMZA-1738: Merge in some minor additions from Linkedin branch
Cameron Lee [Thu, 28 Jun 2018 16:10:44 +0000 (09:10 -0700)] 
SAMZA-1738: Merge in some minor additions from Linkedin branch

Author: Cameron Lee <calee@linkedin.com>

Reviewers: Yi Pan <nickpan47@gmail.com>

Closes #549 from cameronlee314/sync_li_trunk

3 weeks agoSAMZA-1762: Fix Memory link in the Timer Registry Map
xinyuiscool [Tue, 26 Jun 2018 21:35:00 +0000 (14:35 -0700)] 
SAMZA-1762: Fix Memory link in the Timer Registry Map

Found a memory leak in the SystemTimerScheduler which does not remove the timers from scheduledFutures after the timers are fired. This caused memory problem for Samza jobs using TimerFn feature. This patch fixes this issue.

Author: xinyuiscool <xiliu@linkedin.com>

Reviewers: Boris S <boris@apache.org>

Closes #566 from xinyuiscool/SAMZA-1762

3 weeks agoSAMZA-1753: Added timestamp to Incoming message envelope.
Boris S [Mon, 25 Jun 2018 20:08:50 +0000 (13:08 -0700)] 
SAMZA-1753: Added timestamp to Incoming message envelope.

Author: Boris S <boryas@apache.org>
Author: Boris Shkolnik <bshkolni@linkedin.com>

Reviewers: xinyuiscool@apache.org

Closes #559 from sborya/kafkaTS

4 weeks agoMinor: Changing tests which use resource files to use Class.getResource instead of...
Cameron Lee [Sat, 23 Jun 2018 01:18:06 +0000 (18:18 -0700)] 
Minor: Changing tests which use resource files to use Class.getResource instead of File to get the path to the resource; without this change, these tests fail when run in the IDE

Author: Cameron Lee <calee@linkedin.com>

Reviewers: Jagadish<jagadish@apache.org>

Closes #562 from cameronlee314/get_resource

4 weeks agoSAMZA-1718: Simplify management of Zookeeper coordination state
Jagadish [Sat, 23 Jun 2018 01:06:51 +0000 (18:06 -0700)] 
SAMZA-1718: Simplify management of Zookeeper coordination state

1. Currently coordination related state is spread across several Zookeeper classes. There are also back-and-forth flows that exist between the ZkJobCoordinator, ZkControllerImpl, ZkControllerListener and ZkLeaderElector. This PR nukes un-necessary interfaces (and their implementation classes), simplifies state management and unifies state in the ZkJobCoordinator class.

2. Clearly defined life-cycle hooks on events:
- Protocol validations happen once during the lifecycle of a StreamProcessor (instead of each new session)
- New subscriptions to listeners happen at each a new Zk session

Author: Jagadish <jvenkatraman@linkedin.com>

Reviewers: Prateek M<pmaheshw@linkedin.com>

Closes #525 from vjagadish/zk-simplify

4 weeks agoFixed failed unit tests in TestCachingTable
Wei Song [Sat, 23 Jun 2018 00:29:29 +0000 (17:29 -0700)] 
Fixed failed unit tests in TestCachingTable

This is due to the recent refactoring of table metrics, for some reason running build locally didn't catch these failed tests.

Author: Wei Song <wsong@wsong-mn2.linkedin.biz>
Author: Cameron Lee <calee@linkedin.com>
Author: Jagadish <jvenkatraman@linkedin.com>
Author: Ray Matharu <rmatharu@linkedin.com>

Reviewers: Jagadish<jagadish@apache.org>, Cameron L<calee@linkedin.com>

Closes #560 from weisong44/table-metrics-fix

4 weeks agoSAMZA-1748: Standalone failure tests.
Shanthoosh Venkataraman [Fri, 22 Jun 2018 22:53:43 +0000 (15:53 -0700)] 
SAMZA-1748: Standalone failure tests.

In the standalone model, a processor can leave and join the group at any point in time.  This processor reshuffle is referred to as rebalancing which results in task(work) redistribution amongst other available, live processors in the group.

Processor rebalancing in existing standalone integration tests(junit tests) is accomplished through clean shutdown of the processors. However, in real production scenarios, processor rebalancing is triggered through unclean shutdown and full garbage collection(GC) of the processors.

As a part of this patch to cover those scenarios, the following integration tests are added.

1. Force killing the leader processor of the group.
2. Force killing a single follower in the group.
3. Force killing multiple followers in the group.
4. Force killing the leader and a follower in the  group.
5. Suspending and resuming the leader of the group.

Since existing standalone integration tests cover event consumption/production after the re-balancing phase, these new tests will just test the coordination. We'll iterate on this initial suite and add tests whenever necessary.

Author: Shanthoosh Venkataraman <santhoshvenkat1988@gmail.com>

Reviewers: Jagadish V <vjagadish1989@apache.org>

Closes #554 from shanthoosh/standalone_failure_tests

4 weeks agoSAMZA-1756: System exit calls in ApplicationRunnerMain break ProcessJob and cause...
Cameron Lee [Fri, 22 Jun 2018 22:46:14 +0000 (15:46 -0700)] 
SAMZA-1756: System exit calls in ApplicationRunnerMain break ProcessJob and cause unit tests to get skipped

Author: Cameron Lee <calee@linkedin.com>

Reviewers: Xinyu Liu <xinyu@apache.org>

Closes #561 from cameronlee314/app_runner_main_exit

4 weeks agoSAMZA-1648: Integration Test Framework & Collection Stream Impl
sanil15 [Fri, 22 Jun 2018 22:19:50 +0000 (15:19 -0700)] 
SAMZA-1648: Integration Test Framework & Collection Stream Impl

This patch provides the following:
- TestRunner: Tesing Wrapper to run Samza job
- CollectionStream: Acts as a stream descriptor for in memory collections
- CollectionStreamSystem: System associated with a Collection
- StreamUtils: Utilities over streams
- Sample example of tests

Link to SEP: https://cwiki.apache.org/confluence/display/SAMZA/SEP-12%3A+Integration+Test+Framework

Author: sanil15 <sanil.jain15@gmail.com>

Reviewers: Xinyu Liu <xinyu@apache.org>

Closes #501 from Sanil15/SAMZA-1648

4 weeks agoSAMZA-1726: Isolate InMemorySystemFactory to run separately per job
sanil15 [Fri, 22 Jun 2018 19:44:57 +0000 (12:44 -0700)] 
SAMZA-1726: Isolate InMemorySystemFactory to run separately per job

Tested by running the corresponding integration and unit tests

Author: sanil15 <sanil.jain15@gmail.com>

Reviewers: Xinyu Liu <xinyuliu.us@gmail.com>

Closes #532 from Sanil15/SAMZA-1726

5 weeks agoSAMZA-1751: Refactored metrics for table API
Wei Song [Fri, 15 Jun 2018 23:27:45 +0000 (16:27 -0700)] 
SAMZA-1751: Refactored metrics for table API

Refactored metrics for table API
 - Added TableMetricsUtil that encapsulates required parameters, maintains naming consistency and simplifies metrics creation API for tables.
 - Added metrics to local table
 - Maintained consistency between local, remote and caching table

Author: Wei Song <wsong@wsong-mn2.linkedin.biz>

Reviewers: Peng Du<pdu@linkedin.com>

Closes #555 from weisong44/table-metrics

5 weeks agoSAMZA-1747: Add metric to measure effectiveness of host-affinity
Jagadish [Thu, 14 Jun 2018 21:04:53 +0000 (14:04 -0700)] 
SAMZA-1747: Add metric to measure effectiveness of host-affinity

We require visibility into how effectively host-affinity performs. The goal is to help easily answer the following questions.
- How effectively is YARN matching my preferred-host requests
- When does Samza fallback to abandoning locality and issuing any-host requests?

design doc: https://docs.google.com/document/d/1oeNKDnG4JIGT2846us-jpnGW_RUjMjPIKDeEUlE_-jg/edit#

Author: Jagadish <jvenkatraman@linkedin.com>

Reviewers: Prateek M<pmaheshw@linkedin.com>

Closes #553 from vjagadish1989/hostaffinity-metrics

5 weeks agoAdd a new ListGauge metric-type
Ray Matharu [Thu, 14 Jun 2018 04:01:53 +0000 (21:01 -0700)] 
Add a new ListGauge metric-type

This PR introduces a ListGauge type,
A subsequent PR: https://github.com/apache/samza/pull/543 shows how this issued in conjunction with a diagnostics appender for error-tracking and dumpting to kafka via SnapshotReporter.

Author: Ray Matharu <rmatharu@linkedin.com>

Reviewers: Jagadish <jagadish@apache.org>, Cameron Lee <calee@linkedin.com>

Closes #541 from rayman7718/listgauge

5 weeks agoSAMZA-1709: Use lazy creation for SystemAdmins in ApplicationRunners
Cameron Lee [Wed, 13 Jun 2018 23:12:44 +0000 (16:12 -0700)] 
SAMZA-1709: Use lazy creation for SystemAdmins in ApplicationRunners

An instance of SystemAdmins is created when instantiating any AbstractApplicationRunner, but the SystemAdmins is only actually needed for some of the methods for some of the runners. For example, LocalApplicationRunner.kill does not need SystemAdmins, and LocalContainerRunner does not need SystemAdmins for anything.
Doing lazy instantiation allows us to more easily manage the SystemAdmins lifecycle, since it removes the need to add lifecycle hooks for the ApplicationRunner.
This also fixes the lifecycle management for SystemAdmins in ApplicationRunners.

Author: Cameron Lee <calee@linkedin.com>

Reviewers: Xinyu Liu <xinyu@apache.org>

Closes #512 from cameronlee314/runnner_system_admins

5 weeks agoSAMZA-1670: When fetching a newest offset for a partition, also prefetch and cache...
Cameron Lee [Wed, 13 Jun 2018 23:04:02 +0000 (16:04 -0700)] 
SAMZA-1670: When fetching a newest offset for a partition, also prefetch and cache the newest offsets for other partitions on the container

Author: Cameron Lee <calee@linkedin.com>

Reviewers: Jagadish <jagadish@apache.org>

Closes #520 from cameronlee314/partition_metadata

5 weeks agoSAMZA-1740: Moving SamzaSqlRelRecord to samza-api as it is needed to be used in UDFs
Aditya Toomula [Mon, 11 Jun 2018 17:49:45 +0000 (10:49 -0700)] 
SAMZA-1740: Moving SamzaSqlRelRecord to samza-api as it is needed to be used in UDFs

Please see description in the ticket. Also, implementing equals and hashCode methods for SamzaSqlRelRecord and SamzaSqlRelMessage.

Author: Aditya Toomula <atoomula@linkedin.com>

Reviewers: Srini P<spunuru@linkedin.com>, Jagadish <jagadish@apache.org>

Closes #545 from atoomula/sql

6 weeks agoFix loadDefaults error msg.
Boris S [Fri, 8 Jun 2018 21:25:27 +0000 (14:25 -0700)] 
Fix loadDefaults error msg.

SAMZA-1744

Author: Boris S <boryas@apache.org>
Author: Boris Shkolnik <bshkolni@linkedin.com>

Reviewers: Xinyu Liu <xinyu@apache.org>

Closes #551 from sborya/loadDefaultsErrorMsg and squashes the following commits:

c3003ad2 [Boris S] Fixed error message
0edf343b [Boris S] Merge branch 'master' of https://github.com/apache/samza
67e611ee [Boris S] Merge branch 'master' of https://github.com/apache/samza
dd39d089 [Boris S] Merge branch 'master' of https://github.com/apache/samza
1ad58d43 [Boris S] Merge branch 'master' of https://github.com/apache/samza
06b1ac36 [Boris Shkolnik] Merge branch 'master' of https://github.com/sborya/samza
5e6f5fb5 [Boris Shkolnik] Merge branch 'master' of https://github.com/apache/samza
010fa168 [Boris S] Merge branch 'master' of https://github.com/apache/samza
bbffb79b [Boris S] Merge branch 'master' of https://github.com/apache/samza
d4620d66 [Boris S] Merge branch 'master' of https://github.com/apache/samza
410ce78b [Boris S] Merge branch 'master' of https://github.com/apache/samza
a31a7aa2 [Boris Shkolnik] reduce debugging from info to debug in KafkaCheckpointManager.java

6 weeks agoSAMZA-1742: Add metrics reporter parameter to LocalApplicationRunner constructor.
Daniel Nishimura [Fri, 8 Jun 2018 21:02:53 +0000 (14:02 -0700)] 
SAMZA-1742: Add metrics reporter parameter to LocalApplicationRunner constructor.

vjagadish1989 this has already been reviewed and approved by you and cameronlee314 internally. Please approve here. Thanks!

Author: Daniel Nishimura <dnishimura@linkedin.com>

Reviewers: Jagadish <jagadish@apache.org>

Closes #550 from dnishimura/samza-1742-localapplicationrunner-custom-metrics

6 weeks agoSAMZA-1741: fix issue that EH consumer taking too long to shutdown
Hai Lu [Fri, 8 Jun 2018 17:05:36 +0000 (10:05 -0700)] 
SAMZA-1741: fix issue that EH consumer taking too long to shutdown

1.  lower the shutdown timeout from 1 min to 15 seconds
2. make sure EventHubManagers are shutdown in parallel
3. print a thread dump when we do fail during shutdown

Author: Hai Lu <halu@linkedin.com>

Reviewers: Jagadish <jagadish@apache.org>, Prateek <pmaheshw@linkedin.com>

Closes #548 from lhaiesp/master

6 weeks agoConvert a put to to a delete operation in ReadWriteTable and TableWriteFunction when...
Wei Song [Thu, 7 Jun 2018 23:51:30 +0000 (16:51 -0700)] 
Convert a put to to a delete operation in ReadWriteTable and TableWriteFunction when input value is null

Currently, the behavior of putting a null value is inconsistent: it is a delete for RocksDB, and not supported in in-memory store, and on a case-by-case basis for remote tables. It is desirable to unify the behavior. Furthermore, it eases the writing of a change captured stream to a table. A change captured stream contains typically 3 types of events: INSERT, UPDATE and DELETE, and they need to be applied properly when written to a table to produce a correct snapshot. In a change captured stream the payload of a DELETE event is typically is null, and this would result in a delete operation to a table in sendTo() operator.

Author: Wei Song <wsong@wsong-mn2.linkedin.biz>

Closes #547 from weisong44/table-fix

6 weeks agoMinor: KafkaConfig should treat empty changelog name as no changelog.
Prateek Maheshwari [Wed, 6 Jun 2018 18:00:06 +0000 (11:00 -0700)] 
Minor: KafkaConfig should treat empty changelog name as no changelog.

If a store changelog stream name is empty, treat is as a non-changelogged store instead of throwing an exception.

Author: Prateek Maheshwari <pmaheshwari@linkedin.com>

Reviewers: Jagadish Venkatraman <vjagadish1989@gmail.com>

Closes #546 from prateekm/kafka-changelog

6 weeks agoSAMZA-929: Set initialDelay in tokenRenewExecutor schedule to 0
Apoorva Sareen [Tue, 5 Jun 2018 18:44:49 +0000 (11:44 -0700)] 
SAMZA-929: Set initialDelay in tokenRenewExecutor schedule to 0

Changed initialDelay in tokenRenewExecutor scheduler to 0 so that it can re-login using the keytab as soon as the application master container starts.  This way even if application master restarts after the delegation token in launcher context has expired, it will be able to use the new token to launch other containers.

Author: Apoorva Sareen <asareen@MacBook-Pro-2.local>

Reviewers: Jagadish<jagadish@apache.org>

Closes #544 from apoorva121/master

7 weeks agoSAMZA-1736: Add counters and timers for batch get/put/delete operations in KeyValueSt...
Prateek Maheshwari [Fri, 1 Jun 2018 20:17:37 +0000 (13:17 -0700)] 
SAMZA-1736: Add counters and timers for batch get/put/delete operations in KeyValueStorageEngine

Author: Prateek Maheshwari <pmaheshwari@linkedin.com>

Reviewers: Cameron Lee <calee@linkedin.com>, Shanthoosh Venkatraman <svenkatr@linkedin.com>

Closes #539 from prateekm/store-metrics

7 weeks agoSAMZA-1719: Add caching support to table-api
Peng Du [Thu, 31 May 2018 17:43:30 +0000 (10:43 -0700)] 
SAMZA-1719: Add caching support to table-api

This change adds caching support for Samza tables. This is especially
useful for remote table where the accesses can have high latency for
applications that can tolerate staleness. Caching is added in the form
of a composite table that wraps the actual table and a cache. We reuse
the ReadWriteTable interface for the cache. A simple Guava cache-based
table is provided in this change.

Original PR was inadvertently closed: https://github.com/apache/samza/pull/524

Author: Peng Du <pdu@linkedin.com>

Reviewers: Jagadish <jagadish@apache.org>, Wei <wsong@linkedin.com>

Closes #531 from pdu-mn1/table-cache

7 weeks ago[MINOR] Add logging for EventHubs configs
Jagadish [Wed, 30 May 2018 22:51:08 +0000 (15:51 -0700)] 
[MINOR] Add logging for EventHubs configs

prateekm for review

Author: Jagadish <jvenkatraman@linkedin.com>

Reviewers: Prateek M<pmaheshw@linkedin.com>

Closes #540 from vjagadish1989/eh-logging

7 weeks agoSAMZA-1724: Guarantee exit from ApplicationRunnerMain during deploys
Prateek Maheshwari [Wed, 30 May 2018 17:37:23 +0000 (10:37 -0700)] 
SAMZA-1724: Guarantee exit from ApplicationRunnerMain during deploys

Author: Prateek Maheshwari <pmaheshw@LM-LSNSCDW6508.linkedin.biz>

Reviewers: Jagadish Venkatraman <vjagadish1989@gmail.com>

Closes #538 from prateekm/process-exit

7 weeks agoSAMZA-1731: Daemonize threads spawned from SamzaEventHubClientManager.
Shanthoosh Venkataraman [Wed, 30 May 2018 03:13:41 +0000 (20:13 -0700)] 
SAMZA-1731: Daemonize threads spawned from SamzaEventHubClientManager.

**Problem:**
Existing SamzaEventHubClientManager implementation spawns non-daemon threads when instantiating EventHubClient. These non-daemon threads stalls shutdown of samza processes even when the main thread has exited.

**Fix:**
Daemonize threads that are spawned from SamzaEventHubClientManager.

Author: Shanthoosh Venkataraman <santhoshvenkat1988@gmail.com>

Reviewers: Jagadish <jagadish@apache.org>

Closes #536 from shanthoosh/SAMZA-1731

7 weeks agoSAMZA-1728: BootstrappingChooser - Call checkOffset only for a lagging partition...
Aditya Toomula [Wed, 30 May 2018 01:48:02 +0000 (18:48 -0700)] 
SAMZA-1728: BootstrappingChooser - Call checkOffset only for a lagging partition while choosing.

Author: Aditya Toomula <atoomula@linkedin.com>

Reviewers: Jagadish<jagadish@apache.org>

Closes #533 from atoomula/chooser

7 weeks agoSAMZA-1732: Reduce the coordination timeouts in TestZkLocalApplicationRunner tests.
Shanthoosh Venkataraman [Tue, 29 May 2018 23:26:25 +0000 (16:26 -0700)] 
SAMZA-1732: Reduce the coordination timeouts in TestZkLocalApplicationRunner tests.

Currently all the tests in TestZkLocalApplicationRunner takes around 5 minutes to finish. Reducing the coordination timeout to reduce the test time.

Changes in TestZkLocalApplicationRunner test timeout values:
* Change debounce timeout from 20 seconds to 2 seconds.
* Change task.shutdown timeout from 30 seconds to 2 seconds.
* Change barrier timeout from 40 seconds to 2 seconds.

After this change, execution time of TestZkLocalApplicationRunner tests has reduced from `310` seconds to `55` seconds.

Author: Shanthoosh Venkataraman <svenkataraman@linkedin.com>

Reviewers: Jagadish <jagadish@apache.org>

Closes #537 from shanthoosh/reduce_zk_localAppRunnerTestTime

8 weeks agoSAMZA-1659: Serializable OperatorSpec
Yi Pan (Data Infrastructure) [Fri, 25 May 2018 16:37:55 +0000 (09:37 -0700)] 
SAMZA-1659: Serializable OperatorSpec

This change is to make the user supplied functions serializable. Hence, making the full user defined DAG serializable.

Author: Yi Pan (Data Infrastructure) <nickpan47@gmail.com>
Author: Yi Pan (Data Infrastructure) <yipan@yipan-mn1.linkedin.biz>
Author: Xinyu Liu <xiliu@xiliu-ld.linkedin.biz>

Reviewers: Jagadish <jvenkatraman@linkedin.com>, Prateek Maheshwari <pmaheshw@linkedin.com>

Closes #475 from nickpan47/serializable-opspec-only-Jan-24-18 and squashes the following commits:

db0dea73 [Yi Pan (Data Infrastructure)] SAMZA-1659: fix intermittent TestZkLocalApplicationRunner failure due to StreamProcessor#stop()
34716d42 [Yi Pan (Data Infrastructure)] SAMZA-1659: fix a comment on OperatorSpec#isClone()
37d4e6ae [Yi Pan (Data Infrastructure)] SAMZA-1659: addressing latest round of review comments
68674a14 [Yi Pan (Data Infrastructure)] Merge branch 'master' into serializable-opspec-only-Jan-24-18
d3a7826c [Yi Pan (Data Infrastructure)] SAMZA-1659: addressing review comments
f83e8dd0 [Yi Pan (Data Infrastructure)] Merge branch 'master' into serializable-opspec-only-Jan-24-18
acca418b [Yi Pan (Data Infrastructure)] Merge branch 'master' into serializable-opspec-only-Jan-24-18
842a73d6 [Yi Pan (Data Infrastructure)] SAMZA-1659: making user-defined functions in high-level API serializable
ad85a2cb [Yi Pan (Data Infrastructure)] Merge branch 'master' into serializable-opspec-only-Jan-24-18
c1567116 [Yi Pan (Data Infrastructure)] SAMZA-1659: Before re-merge with master. Still need to fix unit tests (moving OperatorSpec clone tests to OperatorSpecGraph.clone)
f2563f8e [Yi Pan (Data Infrastructure)] SAMZA-1659: serialize the whole DAG instead of each individual OperatorSpec.
24d33496 [Yi Pan (Data Infrastructure)] SAMZA-1659: updated according to review comments. Need to merge again with master.
3f643f8b [Yi Pan (Data Infrastructure)] SAMZA-1659: serialiable OperatorSpec
ed7d8c0e [Yi Pan (Data Infrastructure)] Fixed some javadoc and test files
94de218b [Yi Pan (Data Infrastructure)] Remove public access from StreamGraphImpl#getIntermediateStream(String, Serde)
8f4e9dd4 [Yi Pan (Data Infrastructure)] Serialization of StreamGraph in a wrapper class SerializedStreamGraph
f3bb1958 [Yi Pan (Data Infrastructure)] Fix some comments
c15246f5 [Yi Pan (Data Infrastructure)] Merge branch 'master' into serializable-opspec-only-Jan-24-18
e981967d [Yi Pan (Data Infrastructure)] WIP: fixing unit test for SamzaSQL translators w/ serialization of operator functions
40583051 [Yi Pan (Data Infrastructure)] WIP: update the serialization of user functions after the merge
18ba924f [Yi Pan (Data Infrastructure)] Merge branch 'master' into serializable-opspec-only-Jan-24-18
93951c5f [Yi Pan (Data Infrastructure)] Merge branch 'master' into serializable-opspec-only-Jan-24-18
54a28801 [Yi Pan (Data Infrastructure)] WIP: broadcast, sendtotable, and streamtotablejoin serialization and unit tests
45eb1fb0 [Yi Pan (Data Infrastructure)] Merge branch 'master' into serializable-opspec-only-Jan-24-18
7c8d1591 [Yi Pan (Data Infrastructure)] WIP: working on unit tests for trigger, broadcast, join, table, and SQL UDF function serialization
b973b105 [Yi Pan (Data Infrastructure)] Merge branch 'master' into serializable-opspec-only-Jan-24-18
aca42308 [Yi Pan (Data Infrastructure)] WIP: Serialize OperatorSpec only w/o StreamApplication interface change. Passed all build and tests.
0ebebfc3 [Yi Pan (Data Infrastructure)] WIP: serialization only change
1670aff0 [Yi Pan (Data Infrastructure)] WIP: class-loading of user program logic and main() method based user program logic are both included in ThreadJobFactory/ProcessJobFactory/YarnJobFactory. ThreadJobFactory test suite to be fixed.
4102aa8c [Yi Pan (Data Infrastructure)] WIP: continued working on potential offspring integration
dc7da87e [Yi Pan (Data Infrastructure)] WIP: unit tests for serialization
475a46bc [Yi Pan (Data Infrastructure)] WIP: fixed TestZkLocalApplicationRunner. Debugging issues w/ TestRepartitionWindowApp (i.e. missing changelog creation step when directly running LocalApplicationRunner)
6a14b2af [Yi Pan (Data Infrastructure)] WIP: fixed unit test failure for Windows
d4640329 [Yi Pan (Data Infrastructure)] WIP: fixing unit tests after merge
bf1ce907 [Yi Pan (Data Infrastructure)] WIP: removing StreamDescriptor first
50201728 [Yi Pan (Data Infrastructure)] Merge branch 'experiment-new-api-v2' into new-api-v2-0.14
dde1ab14 [Yi Pan (Data Infrastructure)] WIP: first end-to-end test
d7df6ed0 [Yi Pan (Data Infrastructure)] WIP: added all unit test for OperatorSpec#copy methods.
6fc6d4c0 [Yi Pan (Data Infrastructure)] WIP: experiment code to implement an end-to-end working example for new APIs
525d8bc1 [Yi Pan (Data Infrastructure)] Merge branch '0.14.0' into new-api-v2
e6fb96e5 [Yi Pan (Data Infrastructure)] WIP: merged all application types into StreamApplications
f227380f [Yi Pan (Data Infrastructure)] WIP: update the app runner classes
256155ad [Yi Pan (Data Infrastructure)] WIP: new API user code examples
4a6a58dc [Yi Pan (Data Infrastructure)] WIP: updated w/ low-level task API and global var ingestion/metrics reporter
3c50629e [Yi Pan (Data Infrastructure)] WIP: adding support for low-level task APIs
51541e13 [Yi Pan (Data Infrastructure)] WIP: cleanup StreamDescriptor
0bc7ee7b [Yi Pan (Data Infrastructure)] WIP: update the user code example on new APIs
cd528c1c [Yi Pan (Data Infrastructure)] WIP: updated spec and user DAG API
b898e6c0 [Yi Pan (Data Infrastructure)] WIP: new-api-v2
91f364f1 [Yi Pan (Data Infrastructure)] WIP: proto-type of input/output stream/system specs
ae3dc6ff [Yi Pan (Data Infrastructure)] WIP: new api revision
8bb97520 [Yi Pan (Data Infrastructure)] WIP: proto-type of input/output stream/system specs
5573a069 [Yi Pan (Data Infrastructure)] WIP: new api revision
aeb45730 [Xinyu Liu] SAMZA-1321: Propagate end-of-stream and watermark messages

8 weeks agoSAMZA-1725: Set travis build idle time out to 20 minutes.
Shanthoosh Venkataraman [Thu, 24 May 2018 02:27:48 +0000 (19:27 -0700)] 
SAMZA-1725: Set travis build idle time out to 20 minutes.

**Problem:**
Currently, average build time of samza codebase is 15 to 20 minutes. However, travis has a build idle timeout of 10 minutes and fails the build if the gradle build command doesn't log anything to console
for 10 minutes(occurs when running tests in samza-test module).

Sample travis build failure error:
```
No output has been received in the last 10m0s, this potentially indicates a stalled build or something wrong with the build itself.
Check the details on how to adjust your build configuration on: https://docs.travis-ci.com/user/common-build-problems/#Build-times-out-because-no-output-was-received
```

**Fix:**
Increasing the build idle-wait timeout value to 20 minutes.

Author: Shanthoosh Venkataraman <santhoshvenkat1988@gmail.com>

Reviewers: Daniel Nishimura <dnishimura@linkedin.com>

Closes #530 from shanthoosh/increase_travis_build_wait_time

8 weeks agoSAMZA-1540: SystemProducer instance for StreamAppender should have task.drop.producer...
Pawas Chhokra [Thu, 24 May 2018 00:59:51 +0000 (17:59 -0700)] 
SAMZA-1540: SystemProducer instance for StreamAppender should have task.drop.producer.errors==true

vjagadish1989 Kindly take a look, thanks.

Author: Pawas Chhokra <pchhokra@pchhokra-mn2.linkedin.biz>

Reviewers: Jagadish <jagadish@apache.org>

Closes #522 from PawasChhokra/SAMZA-1540_ChangesToStreamAppender

8 weeks agoSAMZA-1621: Delete the ephemeral processor node in StreamProcessor shutdown phase.
Shanthoosh Venkataraman [Tue, 22 May 2018 22:19:55 +0000 (15:19 -0700)] 
SAMZA-1621: Delete the ephemeral processor node in StreamProcessor shutdown phase.

Author: Shanthoosh Venkataraman <svenkataraman@linkedin.com>
Author: Shanthoosh Venkataraman <santhoshvenkat1988@gmail.com>

Reviewers: Jagadish <jagadish@apache.org>

Closes #451 from shanthoosh/SAMZA-1621

8 weeks agoSAMZA-1722: Upgrading avro from 1.7.0 to 1.7.1
sanil15 [Tue, 22 May 2018 00:15:31 +0000 (17:15 -0700)] 
SAMZA-1722: Upgrading avro from 1.7.0 to 1.7.1

Jira: SAMZA-1722: https://issues.apache.org/jira/browse/SAMZA-1722

Author: sanil15 <sanil.jain15@gmail.com>

Reviewers: Prateek Maheshwari <pmaheshwari@apache.org>

Closes #529 from Sanil15/SAMZA-1722

8 weeks agoSAMZA-1723: Schedule barrier changes on the debounce thread
Shanthoosh Venkataraman [Mon, 21 May 2018 18:15:51 +0000 (11:15 -0700)] 
SAMZA-1723: Schedule barrier changes on the debounce thread

In existing implementation, `ZkBarrierChangeHandler` is executed from the `ZkEventThread` and has following drawbacks:
* `ZkWatch` events are buffered into a in-memory queue(maintained by ZkClient) and delivered one at a time to ZkClient listener implementations. If the exeuction of a delivered `ZkWatch` event is in progress, then no other `ZkWatch` event will be delivered to the listeners. If `ZkBarrierChangeHandler` is executed from `ZkEventThread`,  any increase in processing latency will delay the delivery of other `ZkWatch` events(buffered in in-memory queue of ZkClient).
* During session expiration(zkConnection error scenario), buffering all events into `ScheduleAfterDebounceTime` helps us to garbage collect older generation events(to ensure correctness and not execute older generation `ZkWatch` events).

Author: Shanthoosh Venkataraman <santhoshvenkat1988@gmail.com>
Author: Shanthoosh Venkataraman <svenkataraman@linkedin.com>

Reviewers: Jagadish <jagadish@apache.org>

Closes #526 from shanthoosh/schedule_barrier_change_in_debounce_thread

8 weeks agoSAMZA-1647: Fix NPE in onJobModelExpired handler in StreamProcessor.
Shanthoosh Venkataraman [Mon, 21 May 2018 18:09:44 +0000 (11:09 -0700)] 
SAMZA-1647: Fix NPE in onJobModelExpired handler in StreamProcessor.

**Changes:**
* Switching to using explicit lock in StreamProcessor to make things simpler on state updation.
* Switch from using synchronized in ZkJobCoordinator to prevent any potential deadlocks
between two threads (where one thread holds the StreamProcessor and other thread has ZkJobCoordinator lock).
* Misc cleanups in StreamProcessor: Remove volatile qualifiers from state variables in StreamProcessor. Remove reinstantiating the
executorService in onNewJobModel.
* ZkJobCoordinator cleanups: Make some state variables as immutable.

**NOTE**: The classes in which these changes were made were aynonymous inner classes,
so to add proper unit tests we need to do big haul of refactor.

Author: Shanthoosh Venkataraman <santhoshvenkat1988@gmail.com>

Reviewers: Jagadish <jagadish@apache.org>

Closes #493 from shanthoosh/fix_npe_in_jobmodel_expired_handler

2 months agoSAMZA-1712: Tear down ZookeeperServer connections in ZkClient on interrupts.
Shanthoosh Venkataraman [Sat, 19 May 2018 00:29:00 +0000 (17:29 -0700)] 
SAMZA-1712: Tear down ZookeeperServer connections in ZkClient on interrupts.

**Problem:**

If a thread executing zkClient.close is interrupted, currently we swallow the ZkInterruptedException and proceed without closing the zookeeper connection.

This leads to ephemeral nodes of StreamProcessor lurking around in zookeeper after StreamProcessor shutdown.

Users had to wait till zookeeper server session timeout for the ephemeral nodes to get deleted.

**Change:**

Retry once on InterruptedException when closing the zkClient.

Misc changes:
* Remove unnecessary null checks.
* Remove unnecessary typecasts.

Author: Shanthoosh Venkataraman <santhoshvenkat1988@gmail.com>

Reviewers: Jagadish <jagadish@apache.org>

Closes #519 from shanthoosh/handle_interrupted_exception_in_zkclient_close

2 months agoSAMZA-1508: JobRunner should not return success until the job is healthy
Jacob Maes [Fri, 18 May 2018 21:23:07 +0000 (14:23 -0700)] 
SAMZA-1508: JobRunner should not return success until the job is healthy

Author: Jacob Maes <jmakes@apache.org>
Author: Jacob Maes <jmaes@linkedin.com>

Reviewers: Prateek Maheshwari <pmaheshw@linkedin.com>

Closes #367 from jmakes/samza-1508

2 months agoImplementing the fetchSinkInfo in ConfigBasedIOResolver
Srinivasulu Punuru [Fri, 18 May 2018 19:31:44 +0000 (12:31 -0700)] 
Implementing the fetchSinkInfo in ConfigBasedIOResolver

1. I think we missed implementing the fetchSinkInfo method in the ConfigBasedResolver when the API was introduced which is breaking the samza sql console tool. This fixes it.
2. latest release of mac removed realpath so the command line tools are broken. Removed the usage of realpath to fix these tools.

Thanks to nickpan47 for identifying these problems.

Author: Srinivasulu Punuru <spunuru@linkedin.com>

Reviewers: Yi Pan <nickpan47@gmail.com>

Closes #528 from srinipunuru/release-fix.1

2 months agoSAMZA-1720: Remove javafx.util dependency from samza-sql tests.
Shanthoosh Venkataraman [Fri, 18 May 2018 19:29:24 +0000 (12:29 -0700)] 
SAMZA-1720: Remove javafx.util dependency from samza-sql tests.

In samza-sql module, currently few test classes(`TestSamzaSqlRelMessageSerde` and `TestSamzaSqlRelRecordSerde`) are dependent upon `javafx.util.Pair` class(coming from `javafx` module). `javafx.util.Pair` is not supported by default in all JDK builds(example; open-jdk java-8 doesn't support `javafx` module) and it belongs to `javafx` package which is primarily used for developing GUI applications. This dependency is removed and replaced with `Pair` class from `apache-commons`.

Author: Shanthoosh Venkataraman <santhoshvenkat1988@gmail.com>

Reviewers: Jagadish V <vjagadish@gmail.com>

Closes #527 from shanthoosh/SAMZA-1720

2 months agoSAMZA-1715: Unit test for Kafka admin deletedMessagesCalled() fail
Yi Pan (Data Infrastructure) [Thu, 17 May 2018 05:19:37 +0000 (22:19 -0700)] 
SAMZA-1715: Unit test for Kafka admin deletedMessagesCalled() fail

Test locally and works.

Author: Yi Pan (Data Infrastructure) <yipan@yipan-mn1.linkedin.biz>

Reviewers: Jagadish <jagadish@apache.org>

Closes #523 from nickpan47/fix-unittest-deleted-messages

2 months agoSAMZA-1711: Re-enable existing standalone integration tests.
Shanthoosh Venkataraman [Mon, 14 May 2018 17:28:52 +0000 (10:28 -0700)] 
SAMZA-1711: Re-enable existing standalone integration tests.

**Changes:**

* Enable all existing standalone integration tests except `TestZkStreamProcessorSession`(`TestZkStreamProcessorSession` is flaky. It spawns `x` StreamProcessors and kills one StreamProcessor through zookeeper session expiration. Sleeps for 5 seconds and proceeds to do validation. If the rebalancing phase takes longer the sleep time, validation fails).
* Remove zookeeper unavailable unit test from LocalApplicationRunner(Race condition in zookeeper shutdown fails other tests). The deleted test will be added back in a separate test class.
* Increase zookeeper server minimum session timeout from 6 seconds to 120 seconds.
* Add assertions to validate if kafka topics setup were successful before the unit tests.

**Validation:**

Verified by running the following script on top of this patch in master branch.

```bash
i=0
while [ $i -lt 50 ]; do
    i=`expr $i + 1`
    echo "Run " +$i
    ./gradlew clean :samza-test:test -Dtest.single="TestZkLocalApplicationRunner" --debug --stacktrace >> ~/test-logs_10
    ./gradlew clean :samza-test:test -Dtest.single="TestZkStreamProcessor" --debug --stacktrace >> ~/test-logs_10
    ./gradlew clean :samza-test:test -Dtest.single="TestStreamProcessor" --debug --stacktrace >> ~/test-logs_10
    ./gradlew clean :samza-test:test -Dtest.single="TestZkStreamProcessorFailures" --debug --stacktrace >> ~/test-logs_10
done;
```

**Result:**

```bash
[svenkatasvenkata-ld2 samza]$ grep 'BUILD SUCCESS' ~/test-logs_10  | wc -l
200
[svenkatasvenkata-ld2 samza]$ grep 'BUILD FAIL' ~/test-logs_10  | wc -l
0
```

Author: Shanthoosh Venkataraman <santhoshvenkat1988@gmail.com>
Author: Shanthoosh Venkataraman <svenkataraman@linkedin.com>

Reviewers: Xinyu Liu <xinyu@apache.org>

Closes #515 from shanthoosh/turn_all_integration_tests_on

2 months agoSAMZA-1696: Delete TestKeyValueStores flaky test
Ahmed Abdul Hamid [Fri, 11 May 2018 17:55:02 +0000 (10:55 -0700)] 
SAMZA-1696: Delete TestKeyValueStores flaky test

`testParallelReadWriteSameKey` is a flaky test that was failing due to a race condition between the main test thread and the thread manipulating the key-value store under test. Specifically, the main test thread could assert the store has received a value before another thread gets to set it. The hard-coded wait durations did not guarantee the main test thread would wait for all other threads to complete, causing it to assert prematurely.

This change deletes this test since the underlying `RocksDB` store wrapped within `RocksDbKeyValueStore` is thread-safe by design, which renders this test redundant and unnecessary.

Author: Ahmed Abdul Hamid <ahabdulh@ahabdulh-ld1.linkedin.biz>

Reviewers: Shanthoosh Venkataraman <svenkata@linkedin.com>

Closes #500 from ahmedahamid/dev/fix-1696

2 months agoRemove the iterable interface from KeyValueSnapshot
xinyuiscool [Thu, 10 May 2018 21:19:08 +0000 (14:19 -0700)] 
Remove the iterable interface from KeyValueSnapshot

The iterable interface makes it hard for the users to close it after using.

Author: xinyuiscool <xiliu@linkedin.com>

Reviewers: Prateek M <prateekm@apache.org>

Closes #516 from xinyuiscool/kv-snapshot

2 months agoFixed test failure for TestRocksDbKeyValueStoreJava#testPerf
Prateek Maheshwari [Thu, 10 May 2018 19:16:33 +0000 (12:16 -0700)] 
Fixed test failure for TestRocksDbKeyValueStoreJava#testPerf

Iterators (incl. those obtained from snapshots) must be closed before store close.

Author: Prateek Maheshwari <pmaheshw@linkedin.com>

Reviewers: Xinyu Liu <xinyuliu.us@gmail.com>

Closes #514 from prateekm/rocksdb-test-fi

2 months agoSAMZA-1700: Clean up SystemAdmins instance creation flows
Cameron Lee [Thu, 10 May 2018 17:45:41 +0000 (10:45 -0700)] 
SAMZA-1700: Clean up SystemAdmins instance creation flows

The SystemAdmins class has a special "test" constructor for building SystemAdmins instances for unit tests. However, that test constructor has leaked into non-test code.
This makes it harder to manage all flows which use SystemAdmins. An upcoming change is to fix lifecycle management for SystemAdmins in the ApplicationRunner classes, so doing some clean up will help that future change.

Author: Cameron Lee <calee@linkedin.com>

Reviewers: Prateek Maheshwari <pmaheshwari@apache.org>

Closes #509 from cameronlee314/systemadmins_cleanup

2 months agoFixed test failure for TestRocksDbKeyValueStoreJava#testIterate
Prateek Maheshwari [Thu, 10 May 2018 16:21:12 +0000 (09:21 -0700)] 
Fixed test failure for TestRocksDbKeyValueStoreJava#testIterate

RocksDB Snapshots and any iterators obtained from them need to be closed before the store is closed.

Otherwise the process aborts with the following message (at least on OSX):
`Assertion failed: (is_last_reference), function ~ColumnFamilyData, file db/column_family.cc, line 457.`

Author: Prateek Maheshwari <pmaheshw@linkedin.com>

Reviewers: Jagadish V <vjagadish1989@gmail.com>

Closes #513 from prateekm/rocksdb-test-fi

2 months agoSAMZA-1706: lazy initialization for eventhub system producer
Hai Lu [Wed, 9 May 2018 22:32:44 +0000 (15:32 -0700)] 
SAMZA-1706: lazy initialization for eventhub system producer

We are seeing slow shutdown issue for eventhub system producers for users who only use eventhub consumer (but then Samza system creates both consumer and producer together no matter what). As a workaround, add lazy initialization to the producer to avoid the slow shutdown

Author: Hai Lu <halu@linkedin.com>

Reviewers: Jagadish <jagadish@apache.org>

Closes #511 from lhaiesp/master

2 months agoSAMZA-1705: Switch to use snapshot in iterable impl of RocksDb
xinyuiscool [Wed, 9 May 2018 00:48:55 +0000 (17:48 -0700)] 
SAMZA-1705: Switch to use snapshot in iterable impl of RocksDb

We should use rocksDb.snapshot() method to keep the snapshot and creates a new iterator with it all the time. The perf shows a little bit more expensive but mostly on par with range iterator query.

Author: xinyuiscool <xiliu@linkedin.com>

Reviewers: Jagadish V <vjagadish@apache.org>

Closes #510 from xinyuiscool/SAMZA-1705

2 months agoSAMZA-1695: Clear events in ScheduleAfterDebounceTime on session expiration
Shanthoosh Venkataraman [Tue, 8 May 2018 02:40:03 +0000 (19:40 -0700)] 
SAMZA-1695: Clear events in ScheduleAfterDebounceTime on session expiration

Scenario:
Let's assume there're three processors in the group [P1, P2, P3] and P1 is the leader.

1. Leader processor(P1) loses connectivity with a zookeeper server in the ensemble and it's ephemeral processor node is deleted(due to session expiration).
2. Immediate successor(P2) to the leader(P1) finds out that the leader is dead and declares itself as leader. Processor P2 Schedules onProcessorChange to publish JobModel.
3. ZkClient connection retry logic helps the Leader(P1) to reconnect to another zkServer in the ensemble and it joins as follower.
4. Processor P1 acts on the stale buffered event in the debounce queue(which it received when it's a leader) and acts as leader. At this point, there're two processors acting as leader(P1 & P2). If P1 proceeds to execute leader actions before P2, P2 will fail(and in worst case can cause state corruption).

Sample exception logs:
https://gist.github.com/shanthoosh/55410fe4ebf3cfb65281b35f16397cad

Author: Shanthoosh Venkataraman <santhoshvenkat1988@gmail.com>
Author: Shanthoosh Venkataraman <shanthoosh@users.noreply.github.com>

Reviewers: Jagadish <jagadish@apache.org>

Closes #499 from shanthoosh/remove_events_from_debounce_queue_on_session_expiry

2 months agoSAMZA-1704: Fix compatibility issues with scala 2.12
xiliu [Tue, 8 May 2018 01:14:29 +0000 (18:14 -0700)] 
SAMZA-1704: Fix compatibility issues with scala 2.12

Need to add override keyword for overriding a method in scala 2.12.

Author: xiliu <xiliu@linkedin.com>

Reviewers: Prateek M <prateekm@apache.org>

Closes #508 from xinyuiscool/SAMZA-1704

2 months agoSAMZA-1703: Disable flaky test TestEmbeddedTaggedRateLimiter.testAcquireWithTimeout
xiliu [Tue, 8 May 2018 00:06:35 +0000 (17:06 -0700)] 
SAMZA-1703: Disable flaky test TestEmbeddedTaggedRateLimiter.testAcquireWithTimeout

Author: xiliu <xiliu@linkedin.com>

Reviewers: Boris S <sborya@apache.org>

Closes #507 from xinyuiscool/SAMZA-1703

2 months agoSAMZA-1702: Prepare 0.14.1 release on the master branch
xiliu [Mon, 7 May 2018 23:13:27 +0000 (16:13 -0700)] 
SAMZA-1702: Prepare 0.14.1 release on the master branch

Author: xiliu <xiliu@linkedin.com>

Reviewers: Prateek M <prateekm@apache.org>

Closes #506 from xinyuiscool/SAMZA-1702-master

2 months agoSAMZA-1261: Fix TestProcessJob flaky test
Ahmed Abdul Hamid [Mon, 7 May 2018 20:45:53 +0000 (13:45 -0700)] 
SAMZA-1261: Fix TestProcessJob flaky test

- Fix flaky test, `TestProcessJob` `testProcessJobKillShouldWork`, which was failing intermittently due to a race condition. In particular, the thread running the test could assert `jobModelManager.stopped` before another thread, enclosed within `ProcessJob.submit`, could actually invoke `jobModelManager.stop`.

+ Refactor `ProcessJob` to improve its overall robustness
  + Handle corner cases, e.g.
     + Fail gracefully if starting process within `ProcessJob.submit` throws
     + Ignore attempts to kill a job before it is submitted
     + Ensure job status is always set appropriately
  + Remove unnecessary stdout/stderr piping code
  + Employ `wait`/`notify` instead of `Thread.sleep`
  + Eliminate all artificial wait method invocations intended to influence inter-thread execution order in unit tests
  + Add more unit tests

Author: Ahmed Abdul Hamid <ahabdulh@ahabdulh-ld1.linkedin.biz>

Reviewers: Boris S<sborya@apache.org>, Shanthoosh V<svenkata@linkedin.com>

Closes #485 from ahmedahamid/master

2 months agoSAMZA-1692: Standalone stability fixes.
Shanthoosh Venkataraman [Mon, 7 May 2018 20:43:05 +0000 (13:43 -0700)] 
SAMZA-1692: Standalone stability fixes.

- Currently, on session expiration processorListener with incorrect generationId is registered with zookeeper(ZkUtils generationId is incremented on reconnect but the generationId in processorListener is zero all the time). When a session reconnect happens to a processor successive to leader, leader expiration event will be skipped. This will prevent leader re-election on a current leader death and will stall the processors group. Fix is to reinstantiate and then register processorChangeListener on session expiration.
- Add processorId to debounce thread name (this can aid debugging when multiple processors are running within a jvm).
- After ScheduleAfterDebounceTime queue is shutdown, don't accept new schedule requests. Current ZkJobCoordinator shutdown sequence comprise of the following steps
     - Shutdown the ScheduleAfterDebounceTime queue.
     - Stop the zkClient  and relinquish it's resources.

After we shutdown ScheduleAfterDebounceTime and before zkclient is stopped, any new operations can be scheduled in ScheduleAfterDebounceTime queue. This will result in RejectedExecutionException, since executorService is stopped.

```
Caused by: java.util.concurrent.RejectedExecutionException: Task java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask23f962a8 rejected from java.util.concurrent.ScheduledThreadPoolExecutor43408be8
```

Author: Shanthoosh Venkataraman <svenkataraman@linkedin.com>
Author: Shanthoosh Venkataraman <santhoshvenkat1988@gmail.com>

Reviewers: Jagadish <jagadish@apache.org>

Closes #496 from shanthoosh/master

2 months agoSAMZA-1563: Make RocksDB store base directory configurable
Bharath Kumarasubramanian [Mon, 7 May 2018 19:34:26 +0000 (12:34 -0700)] 
SAMZA-1563: Make RocksDB store base directory configurable

xinyuiscool ^^

Author: Bharath Kumarasubramanian <bkumaras@linkedin.com>

Reviewers: Prateek M <prateekm@apache.org>

Closes #491 from bharathkk/samza-1563

2 months agoSAMZA-1653: Support waitForFinish in remote application runner and add waitForFinish
Bharath Kumarasubramanian [Mon, 7 May 2018 18:11:01 +0000 (11:11 -0700)] 
SAMZA-1653: Support waitForFinish in remote application runner and add waitForFinish

Added the following APIs to ApplicationRunner
 `void waitForFinish()`
 `boolean waitForFinish(Duration timeout)`

Implemented the wait for finish methods in remote application runner. Note currently, there is disparity in the APIs in terms of associating runners with stream application. Ideally, we want to decide on the cardinal relation between them and change the APIs accordingly.

The goal of the PR is limited to introduce API (waitForFinish) parity between runners in the current setup.
xinyuiscool

Author: Bharath Kumarasubramanian <bkumaras@linkedin.com>

Reviewers: Xinyu Liu <xinyuliu.us@gmail.com>

Closes #503 from bharathkk/samza-1653

2 months agoSAMZA-1691: Support get iterable from KeyValueStore
xinyuiscool [Mon, 7 May 2018 16:51:28 +0000 (09:51 -0700)] 
SAMZA-1691: Support get iterable from KeyValueStore

Right now for KeyValueStore we have a range query to return an iterator. For usage in BEAM, we need a iterable which will 1) create the snapshot when called, and 2) create an iterator when needed. Add the iterate() function in KeyValueStore to support it. It's implemented as follows:

1) for rocksDb, it will create the iterator when it's called, which will has a snapshot of the elements. Then every time when the iterator is needed, we will seek the iterator from beginning;

2) for inMemoryDb, it will create the snapshot submap when iterate() is called. The submap is an iterable and it can return a new iterator when needed.

Author: xinyuiscool <xiliu@linkedin.com>

Reviewers: Boris S <sborya@apache.org>

Closes #492 from xinyuiscool/SAMZA-1691

2 months agoSAMZA-1699: Fix NPE in ClusterResourceManager
Jagadish [Fri, 4 May 2018 21:02:35 +0000 (14:02 -0700)] 
SAMZA-1699: Fix NPE in ClusterResourceManager

When the ClusterResourcedManager receives a notification that a container is started, it moves the container from the "pending queue" to its "running queue".
In the meanwhile, it's possible for another thread to remove the mapping for the key. Here's an example:

NMCallbackThread-1:```
pendingYarnContainers.remove(key);```

NMCallbackThread-2:
```
for (String key : pendingYarnContainers.keySet()) {
  yarnContainer = pendingYarnContainers.get(key); <-- could be null depending on whether the removal happened before it.
}```

Author: Jagadish <jvenkatraman@linkedin.com>

Reviewers: Prateek M<pmaheshw@linkedin.com>

Closes #504 from vjagadish/npe-fix-async

2 months agoSAMZA-1698: Update appStatus on failures in localApplication.run(streamApp).
Shanthoosh Venkataraman [Fri, 4 May 2018 00:37:49 +0000 (17:37 -0700)] 
SAMZA-1698: Update appStatus on failures in localApplication.run(streamApp).

Author: Shanthoosh Venkataraman <santhoshvenkat1988@gmail.com>

Reviewers: Jagadish <jagadish@apache.org>

Closes #502 from shanthoosh/local_application_runner_set_exception_in_finish

2 months agoSAMZA-1243: Fix flaky tests in TestLocalStoreMonitor
Ahmed Abdul Hamid [Tue, 1 May 2018 20:49:21 +0000 (13:49 -0700)] 
SAMZA-1243: Fix flaky tests in TestLocalStoreMonitor

Having attempted to reproduce this issue without success, careful analysis of the code indicates the underlying issue is most likely incomplete cleanup of file system directories after test execution possibly due to incomplete or manual test runs. To address this issue, this fix generates a different local store directory for every test to isolate file system side-effects across different tests and test runs.

Author: Ahmed Abdul Hamid <ahabdulh@ahabdulh-ld1.linkedin.biz>

Reviewers: Shanthoosh Venkataraman <svenkata@linkedin.com>

Closes #497 from ahmedahamid/dev/fix-1243

2 months agoSAMZA-1476: Fix TestStatefulTask flaky test
Ahmed Abdul Hamid [Tue, 1 May 2018 20:00:38 +0000 (13:00 -0700)] 
SAMZA-1476: Fix TestStatefulTask flaky test

Unable to reproduce the issue even after constraining CPU/memory resources available to the JVM, this fix addresses a potential root cause — a `CountDownLatch` shared between 2 threads, main test thread and Kafka producer thread, but not marked volatile even though it is reinitialized by the main test thread. This could cause the reported issue since each of the 2 threads could end up invoking `countDown`/`await` on a different `CountDownLatch` object.

It is worthwhile to mention that without this fix, a different test, `TestShutdownStatefulTask`, should have also exhibited some flakiness since it shares the same `TestTask` base with `TestStatefulTask` and exercises exactly the same portion of code that includes the failing assertion. Since no such flakiness was reported for `TestShutdownStatefulTask` however, the assumption made by this fix is that it might have not been encountered or reported.

Author: Ahmed Abdul Hamid <ahabdulh@ahabdulh-ld1.linkedin.biz>

Reviewers: Prateek Maheshwari <pmaheshwari@apache.org>

Closes #498 from ahmedahamid/dev/fix-1476

2 months agoSAMZA-1693: Samza-sql - Adding Serde for rel record and few other minor fixes for...
Aditya Toomula [Mon, 30 Apr 2018 23:51:37 +0000 (16:51 -0700)] 
SAMZA-1693: Samza-sql - Adding Serde for rel record and few other minor fixes for Avro and Rel conversion.

Adding Serde for rel record, as calcite expects the keys to be in string format. Rel converters are always expected to provide keys as strings. If key is an avro record, it is expected that the rel converter changes the avro record to rel record and serializes it and deserializes it when conerting rel message to samza message.

Author: Aditya Toomula <atoomula@linkedin.com>

Reviewers: Srini P<spunuru@linkedin.com>

Closes #495 from atoomula/rel1

2 months agoSAMZA-1689: Add validations before state transitions in ZkBarrierForVersionUpgrade.
Shanthoosh Venkataraman [Sat, 28 Apr 2018 01:55:15 +0000 (18:55 -0700)] 
SAMZA-1689: Add validations before state transitions in ZkBarrierForVersionUpgrade.

Prevent invalid state updations on barrier.
* Introduced a additional barrier state NEW.
* Add state validations before updating the barrier.
* Fix existing TestZkBarrier tests that are disabled and add new tests
  to verify the intended behavior.

Author: Shanthoosh Venkataraman <svenkataraman@linkedin.com>
Author: Shanthoosh Venkataraman <santhoshvenkat1988@gmail.com>

Reviewers: Jagadish <jagadish@apache.org>

Closes #490 from shanthoosh/fix_barrier_state_transitions

2 months agoSAMZA-1582: removed the yajl-ruby dependency for security reasons
Cameron Lee [Fri, 27 Apr 2018 18:59:33 +0000 (11:59 -0700)] 
SAMZA-1582: removed the yajl-ruby dependency for security reasons

The original ticket suggested updating the yajl-ruby dependency for security reasons.
Many of the ruby dependencies for building the docs site are years old. After upgrading the dependencies, the yajl-ruby dependency just went away.
The dependencies were upgraded to the highest version that was compatible with ruby 2.0.0 (jekyll was the only dependency which has newer versions that are only compatible with ruby versions greater than 2.0.0).

Testing done:
bundle exec jekyll serve --watch --baseurl ""
_docs/local-site-test.sh
sanity checked a couple of links

Author: Cameron Lee <calee@linkedin.com>

Reviewers: Jagadish <jagadish@apache.org>

Closes #486 from cameronlee314/yajlruby

2 months agoSAMZA-1688: use per partition eventhubs client
Hai Lu [Fri, 27 Apr 2018 17:22:57 +0000 (10:22 -0700)] 
SAMZA-1688: use per partition eventhubs client

Use per partition eventhubs client to improve throughput. As each eventhub client maintains only one single TCP connection.
Also reduce the time spent on getPartitionRuntimeInfo on starting up, by making the calls run in parallels in system admin and also completely removing it from system consumer.

Benchmark:
See 8x improvement on consumption throughput from ~3Mb/s (or 1.5k QPS) to ~21Mb/s (or 10.5k QPS). Memory footprint doesn't seem to get affected. CPU usage also increases by 8x. Essentially, with per partition client, we are able to saturate the CPU usages. The benchmark I did was on a machine with 8 cores. If the machine has say 16 CPU cores, I expect the the throughput to improve by around 16x.

Author: Hai Lu <halu@linkedin.com>

Reviewers: Xinyu Liu <xinyuliu.us@gmail.com>

Closes #489 from lhaiesp/master

2 months agoMinor-fix: Fix a white-space in ResourceRequestState
Jagadish [Thu, 26 Apr 2018 00:20:01 +0000 (17:20 -0700)] 
Minor-fix: Fix a white-space in ResourceRequestState

2 months agoSAMZA-1687: Prioritize preferred host requests over ANY-HOST requests
Jagadish [Thu, 26 Apr 2018 00:17:34 +0000 (17:17 -0700)] 
SAMZA-1687: Prioritize preferred host requests over ANY-HOST requests

Working on a documentation that describes this better, but a TL;DR summary is that we should prioritize preferred-host requests over ANY_HOST requests.

Yarn enforces these two checks:
1. ANY_HOST requests should always be made with relax-locality = true
2. A request with relax-locality = false should not be in the same priority as another with relax-locality = true

Since the Samza AM makes preferred-host requests with relax-locality = false, it follows that ANY_HOST requests should specify a different priority-level. We can safely set priority of preferred-host requests to be higher than any-host requests since data-locality is critical.

Author: Jagadish <jvenkatraman@linkedin.com>

Closes #488 from vjagadish/priority-host-affinity

2 months agoSAMZA-1681: Samza-sql - Add support for handling older record schema versions in...
Aditya Toomula [Wed, 25 Apr 2018 16:48:11 +0000 (09:48 -0700)] 
SAMZA-1681: Samza-sql - Add support for handling older record schema versions in AvroRelConverter

In addition to handling older record schema versions in AvroRelConverter, this change also handles Avro enum and fixed types and also handles the proper conversion of samza message key to rel message.

Author: Aditya Toomula <atoomula@linkedin.com>

Reviewers: Srini P <spunuru@linkedin.com>

Closes #481 from atoomula/rel

2 months agoSAMZA-1686: Set finite operation timeout when creating zkClient.
Shanthoosh Venkataraman [Tue, 24 Apr 2018 22:56:05 +0000 (15:56 -0700)] 
SAMZA-1686: Set finite operation timeout when creating zkClient.

Currently zkClient is created with operationRetryTimeOut of -1. This causes zkClient to retry indefinitely in case of irrecoverable exceptions thereby delaying the StreamProcessor shutdown.

Author: Shanthoosh Venkataraman <svenkataraman@linkedin.com>

Reviewers: Jagadish Venkatraman <vjagadish1989@apache.org>

Closes #487 from shanthoosh/master

2 months agoSAMZA-1676: miscellaneous fix and improvement for eventhubs system
Hai Lu [Tue, 24 Apr 2018 19:26:14 +0000 (12:26 -0700)] 
SAMZA-1676: miscellaneous fix and improvement for eventhubs system

Including these changes:
- Log the metadata that we are fetching from the event hubs
- Rename readLatency to consumptionLagMs
- fix the issue that readLatency metric returns negative value

Author: Hai Lu <halu@linkedin.com>

Reviewers: Srini P<spunuru@linkedin.com>, Jagadish <jagadish@apache.org>

Closes #484 from lhaiesp/master

2 months agoSAMZA-1685: Need to resolve redirects for virtualenv download location for running...
Cameron Lee [Tue, 24 Apr 2018 03:28:43 +0000 (20:28 -0700)] 
SAMZA-1685: Need to resolve redirects for virtualenv download location for running integration tests

Testing Done: ran bin/integration-tests.sh locally from mac/linux which did not run the integration tests before

Author: Cameron Lee <calee@linkedin.com>

Reviewers: Jagadish <jagadish@apache.org>

Closes #482 from cameronlee314/inttest

2 months agoSAMZA-1656: EventHubSystemAdmin does not fetch metadata for valid streams.
Shanthoosh Venkataraman [Mon, 23 Apr 2018 20:24:54 +0000 (13:24 -0700)] 
SAMZA-1656: EventHubSystemAdmin does not fetch metadata for valid streams.

Currently successive invocation of EventHubSystemAdmin.getSystemStreamMetadata with the same stream collection returns empty results.

This is an implementation bug, where the streams requested as a part of prior invocations of EventHubSystemAdmin.getSystemStreamMetadata are ignored(due to stale caching).

This bug causes the JobModel generation phase to fail and kills the StreamProcessor.

Author: Shanthoosh Venkataraman <svenkataraman@linkedin.com>

Reviewers: Jagadish <jagadish@apache.org>, Srini P<spunuru@linkedin.com>

Closes #474 from shanthoosh/EventHubSystemAdminBug

2 months agoSAMZA-1667: Skip storing configuration as a part of JobModel in zookeeper data nodes.
Shanthoosh Venkataraman [Mon, 23 Apr 2018 20:20:45 +0000 (13:20 -0700)] 
SAMZA-1667: Skip storing configuration as a part of JobModel in zookeeper data nodes.

In general, jobModel configuration contains service access tokens and certs. It's a common practice to run zookeeper in a non-ACL environment. Hence for security purposes, it's essential not to store configuration as a part of JobModel in zookeeper.

Author: Shanthoosh Venkataraman <svenkataraman@linkedin.com>

Reviewers: Jagadish <jagadish@apache.org>

Closes #479 from shanthoosh/nuke_configuration_stored_in_JobModel and squashes the following commits:

b8d2196 [Shanthoosh Venkataraman] Address review comments.
7876a44 [Shanthoosh Venkataraman] Nuke JobModel configuration in ZkJobCoordinator.

2 months agoUpdate PMC membership on the Samza web-page
Jagadish [Sat, 21 Apr 2018 02:52:35 +0000 (19:52 -0700)] 
Update PMC membership on the Samza web-page

Author: Jagadish <jvenkatraman@linkedin.com>

Reviewers: Jagadish <jagadish@apache.org>

Closes #480 from vjagadish/pmc

3 months agoSAMZA-1671: sql: initial insert support for table destination
Peng Du [Fri, 20 Apr 2018 16:18:31 +0000 (09:18 -0700)] 
SAMZA-1671: sql: initial insert support for table destination

Table is another main type of IO abstraction in Samza which supports
both read and write (optional). For the tables that do support writes, we
should be able to allow Samza SQL users to write a query to do that. One
example is to insert into a database. The current code only supports
inserting into a stream. This change adds the initial support for table
insert operation.

Author: Peng Du <pdu@linkedin.com>

Reviewers: Jagadish <jagadish@apache.org>, Srini P<spunuru@linkedin.com>

Closes #465 from pdu-mn1/sql-insert-table

3 months agoSAMZA-1478: Delete unneeded data from intermediate Kafka topic on offset commit
Dong Lin [Wed, 18 Apr 2018 22:31:06 +0000 (15:31 -0700)] 
SAMZA-1478: Delete unneeded data from intermediate Kafka topic on offset commit

Author: Dong Lin <lindong28@gmail.com>
Author: Dong Lin <dolin@linkedin.com>

Reviewers: Prateek Maheshwari <pmaheshwari@apache.org>, Jacob Maes <jmakes@apache.org>, Yi Pan <nickpan47@gmail.com>

Closes #347 from lindong28/SAMZA-1478

3 months agoMisc. Util cleanup
Prateek Maheshwari [Wed, 18 Apr 2018 20:32:15 +0000 (13:32 -0700)] 
Misc. Util cleanup

Major changes:
1. Broke up 'Util' class into multiple classes: 'FileUtil', 'HttpUtil', 'CoordinatorStreamUtil'.
2. Consolidated some Util classes: MathUtil, ScalaJavaUtil
3. Removed redundant Util classes: ClassloaderUtil, ScalaToJavaUtil
4. Renamed some Util classes for consistency: TimerUtils -> TimerUtil.
5. Inlined some util classes and methods to where they're used: LexicographicComparator to RocksDBKeyValueStore, defaultSerdeFactoryFromSerdeName to SerdeManager, etc.

Rest of the changes are updates to use the new classes and methods.

Testing: Local build and test works. Tested with a locally deployed Samza job.

Author: Prateek Maheshwari <pmaheshw@linkedin.com>

Reviewers: Jacob Maes <jmakes@apache.org>, Jagadish Venkatraman <vjagadish1989@gmail.com>

Closes #455 from prateekm/util-cleanup

3 months agoSAMZA-1640: JobModel Json deserialization error in ZkJobCoordinator.
Shanthoosh Venkataraman [Wed, 18 Apr 2018 17:00:56 +0000 (10:00 -0700)] 
SAMZA-1640: JobModel Json deserialization error in ZkJobCoordinator.

Author: Shanthoosh Venkataraman <svenkataraman@linkedin.com>
Author: Shanthoosh Venkataraman <svenkata@LM-LSNSCDW5132.linkedin.biz>

Reviewers: Xinyu Liu <xinyuliu.us@gmail.com>

Closes #466 from shanthoosh/SAMZA-1640

3 months agoSAMZA-1651: Samza-sql - Implement GROUP BY SQL operator
Aditya Toomula [Tue, 17 Apr 2018 23:02:49 +0000 (16:02 -0700)] 
SAMZA-1651: Samza-sql - Implement GROUP BY SQL operator

Author: Aditya Toomula <atoomula@linkedin.com>

Reviewers: Srini P<spunuru@linkedin.com>

Closes #478 from atoomula/groupby1

3 months agoSAMZA-1664: ZkJobCoordinator stability fixes.
Shanthoosh Venkataraman [Tue, 17 Apr 2018 23:02:16 +0000 (16:02 -0700)] 
SAMZA-1664: ZkJobCoordinator stability fixes.

Issues fixed:
* Handle job coordinator shutdown gracefully in case of unclean container shutdowns.
* Fix the zookeeper session handling logic.
* Fix the forever retry timeout in ZkClient re-connect.

Author: Shanthoosh Venkataraman <svenkataraman@linkedin.com>

Reviewers: Xinyu Liu <xinyuliu.us@gmail.com>

Closes #476 from shanthoosh/pullInK2Changes

3 months agoSAMZA-1666: Benchmark SystemProducer and SystemConsumer
Srinivasulu Punuru [Tue, 17 Apr 2018 22:40:59 +0000 (15:40 -0700)] 
SAMZA-1666: Benchmark SystemProducer and SystemConsumer

* Tests to benchmark the performance of the system consumers and producers.
* Config to test the benchmark for the event hub system producer and consumer.

SystemConsumerBench and SystemProducerBench provides base generic implementation to test the benchmark for the system producers and consumers. Any new system that needs benchmark test needs a properties file.

The benchmark test itself is single threaded in the way it consumes and produces events. Scaling the benchmark tests right now involves running multiple processes of these tests in parallel.

Right now we just calculate the event rate, But in future we could create a logging metrics registry to hookup other metrics and log them in console along with event rate while the benchmark tests are being run.

Author: Srinivasulu Punuru <spunuru@linkedin.com>

Reviewers: Jagadish <jagadish@apache.org>, Wei Song<wsong@linkedin.com>

Closes #473 from srinipunuru/benchmark.1

3 months agoRevert "SAMZA-1645: A few issues found by BEAM stress test"
xiliu [Tue, 17 Apr 2018 18:50:44 +0000 (11:50 -0700)] 
Revert "SAMZA-1645: A few issues found by BEAM stress test"

This reverts commit 26294151642283c1cfb51590b51a86d2eaedd11f.

3 months agoInitial version for in memory system.
Bharath Kumarasubramanian [Tue, 17 Apr 2018 16:09:01 +0000 (09:09 -0700)] 
Initial version for in memory system.

Getting the PR out to unblock sanil.
Pending tasks
 1. Add documentation
 2. Add more tests
 3. Currently initialization of the stream happens using createStream on admin. We need to find a way to expose the same functionality to low level users.
 4. To populate the initial stream, clients need to get a handle on producer and produce the messages. Alternatively, we can support serialization of source data and pass it as config to the system to initialize. We have a hook in place for this but not implemented it completely.
  5. Clean up consumed data on the buffer based on the lowest offsets of the consumers.

I will create JIRAs for all the pending tasks and fix them iteratively.

xinyuiscool ^^

Author: Bharath Kumarasubramanian <bkumaras@linkedin.com>

Reviewers: Xinyu Liu <xinyuliu.us@gmail.com>

Closes #459 from bharathkk/single-node-testing

3 months agoSAMZA-1619: Samza-sql: Support serialization of nested samza sql relational message
Aditya Toomula [Mon, 16 Apr 2018 17:26:51 +0000 (10:26 -0700)] 
SAMZA-1619: Samza-sql: Support serialization of nested samza sql relational message

Added support for serialization of nested samza sql rel message and accordingly fixed the conversion of nested avro records to rel message. Please note that we still do not have support for the sql queries that point to fields in nested records (beyond the top level record).

Author: Aditya Toomula <atoomula@linkedin.com>
Author: Aditya Toomula <atoomula@atoomula-ld1.linkedin.biz>

Reviewers: Srinivasulu P<spunuru@linkedin.com>

Closes #464 from atoomula/serde

3 months agoSAMZA-1643: StreamPartitionCountMonitor should only restart/shut down the job if...
Prateek Maheshwari [Fri, 13 Apr 2018 15:37:20 +0000 (08:37 -0700)] 
SAMZA-1643: StreamPartitionCountMonitor should only restart/shut down the job if partition count increases

As an aside, also update the gauge to report current number of partitions instead of the change, since that's what its name indicates.

Author: Prateek Maheshwari <pmaheshw@linkedin.com>

Reviewers: Jagadish  V <jagadish@apache.org>

Closes #468 from prateekm/partition-monitor-fixes

3 months agoSAMZA-1649: Improve host-aware allocation to account for strict locality
Jagadish [Fri, 13 Apr 2018 02:23:44 +0000 (19:23 -0700)] 
SAMZA-1649: Improve host-aware allocation to account for strict locality

Currently working on a doc for the behavior of the CapacityScheduler and further testing is needed on an actual cluster - but here's a summary of why we should set relax-locality = false:
 - Node-local requests are honored only when relax-locality = false
 - With relax-locality = true, the scheduler biases interactivity over data-locality for requests that ask for few resources relative to the size of the cluster.

In addition to the above change, this PR also modifies the allocator algorithm to fallback to "ANY_HOST" requests so that we make progress when the node is unavailable.

Author: Jagadish <jvenkatraman@linkedin.com>

Reviewers: Prateek Maheshwari <pmaheshw@linkedin.com>

Closes #471 from vjagadish/relax-locality-fix

3 months agoSAMZA-1650: Fix for firing trigger at the end of trigger interval for tumbling window
Aditya Toomula [Thu, 12 Apr 2018 23:50:30 +0000 (16:50 -0700)] 
SAMZA-1650: Fix for firing trigger at the end of trigger interval for tumbling window

Author: Aditya Toomula <atoomula@linkedin.com>

Reviewers: Jagadish <jagadish@apache.org>

Closes #472 from atoomula/window

3 months agoSAMZA-1584: Improve logging in StreamProcessor.
Shanthoosh Venkataraman [Thu, 12 Apr 2018 21:34:45 +0000 (14:34 -0700)] 
SAMZA-1584: Improve logging in StreamProcessor.

Add the processorID in the log lines wherever necessary(since we support running multiple stream applications in a JVM) and improving logging in general in StreamProcessor.

Author: Shanthoosh Venkataraman <svenkataraman@linkedin.com>
Author: Shanthoosh Venkataraman <svenkata@lm-lsnscdw5132.linkedin.biz>

Reviewers: Prateek Maheshwari <pmaheshwari@apache.org>

Closes #441 from shanthoosh/SAMZA-1584

3 months agominor fix on eventhubs size limit for event body and partition key
Hai Lu [Thu, 12 Apr 2018 20:31:39 +0000 (13:31 -0700)] 
minor fix on eventhubs size limit for event body and partition key

Author: Hai Lu <halu@linkedin.com>

Reviewers: Jagadish <jagadish@apache.org>, Srinivasulu Punuru<spunuru@linkedin.com>

Closes #470 from lhaiesp/master

3 months agoSAMZA-1183: Fix TestAsyncRunLoop flaky tests
Shanthoosh Venkataraman [Wed, 11 Apr 2018 22:34:43 +0000 (15:34 -0700)] 
SAMZA-1183: Fix TestAsyncRunLoop flaky tests

A. Fix TestProcessInOrder and TestProcessOutOfOrder failures.

Both the tests has two tasks(T1, T2) and the following task to message assignments:
- Task T1: [M1, M2] (Task T1 is expected to process the messages M1 and M2).
- Task T2: [M3] (Task T2 is expected to process the message M3 and stop the runloop).

In some cases, before the task T1 completes processing all it’s messages, T2 gets scheduled and stops the runloop(Stopping all tasks).

We wait in both tests through a CountdownLatch, expecting the tasks to process all the messages, which will never reach zero in the above scenario(there by causing indefinite wait in tests).

Fix: Remove the manual shutdown invocation from Task T2 and use EOS messages to stop the runloop.

B.  Fix TestCommitSingleTask and TestCommitMultipleTask failures.

Both the tests had two tasks(T1, T2).

Task T2 is expected to stop the runloop and task T1 marks the commit flag through taskCoordinator.commit(requestScope).

Failure occurs in some scenarios when the task T2 stops the runloop before the runLoop commits both the tasks.

Fix: Trigger the runloop shutdown sequence after we’ve committed the tasks.

C. Combine the duplicate tests and unify their assertions.

D. Verification: Ran the following script to verify the fixes.

`root88cf6be1c11a:~/samza# i=0`
`root88cf6be1c11a:~/samza# while [ $i -lt 100 ]; do`
       ` i=expr $i + 1`
       ` ./gradlew clean :samza-core:test -Dtest.single="TestAsyncRunLoop" --debug --stacktrace >> test-logs`
  ` done;`

There were no failures.

Author: Shanthoosh Venkataraman <svenkataraman@linkedin.com>

Reviewers: Xinyu Liu <xinyuliu.us@gmail.com>

Closes #463 from shanthoosh/re_enable_async_run_loop_tests_2

3 months agoSAMZA-1645: A few issues found by BEAM stress test
xinyuiscool [Wed, 11 Apr 2018 20:21:36 +0000 (13:21 -0700)] 
SAMZA-1645: A few issues found by BEAM stress test

1. Revert the priority set to intermediate streams.
2. Fix a watermark propagation condition

Author: xinyuiscool <xiliu@linkedin.com>

Reviewers: Prateek M <prateekm@apache.org>

Closes #469 from xinyuiscool/SAMZA-1645

3 months agoUpgrade to latest event hub version (1.0.1)
Srinivasulu Punuru [Tue, 10 Apr 2018 22:39:21 +0000 (15:39 -0700)] 
Upgrade to latest event hub version (1.0.1)

* Upgrade to the latest event hub version 1.0.1
* Adding configs for prefetchCount and maxEventPerPoll
* Fix the high cpu usage issue in SamzaHistogram
* Fixing a race condition in event hub system producer where the future was getting removed while it was being checked for completion resulting in NPE.

Author: Srinivasulu Punuru <spunuru@linkedin.com>

Reviewers: Prateek Maheshwari <pmaheshwari@apache.org>

Closes #467 from srinipunuru/upgrade-eh-1.0.1