SAMZA-1372: Change Latch Interface to Lock Interface for Samza Standalone with ZK #264

PawasChhokra · 2017-08-09T18:57:04Z

No description provided.

sborya · 2017-08-10T23:18:51Z

samza-api/src/main/java/org/apache/samza/system/SystemAdmin.java

+  /**
+   * Check if the stream described by the spec already exists.
+   * @param streamSpec The spec, or blueprint for the physical stream on the system.
+   * @return true if stream exists already, false otherwise


nit. the stream

It seems like we won't be needing this function at all. Hence removed.

sborya · 2017-08-10T23:20:19Z

samza-core/src/main/java/org/apache/samza/coordinator/CoordinationUtils.java


  Latch getLatch(int size, String latchId);
+
+  Lock getLock();


I'd rather named it CoordinationLock (or something), to reduce confusion between this and java concurrency lock.

Renamed it to DistributedLock :)

sborya · 2017-08-10T23:21:45Z

samza-core/src/main/java/org/apache/samza/coordinator/Lock.java

+
+  boolean hasLock();
+
+  void setLockListener(LockListener listener);


I personally think that lock should be synchronous utility. It makes programming much easier.
lock = getLock();

lock.lock()
.....
lock.release/unlock()

Removed the listener interface.

sborya · 2017-08-10T23:23:36Z

samza-core/src/main/java/org/apache/samza/execution/StreamManager.java

+
+  /**
+   * Check if the streams described by the specs already exist.
+   * @param streams A list of stream specs, whose existence we need to check for


nit. s/A/a/

It seems like we won't be needing this function at all. Hence removed.

sborya · 2017-08-10T23:27:38Z

samza-core/src/main/java/org/apache/samza/runtime/LocalApplicationRunner.java

+   * @throws Exception exception for latch timeout
+   */
+  private void createStreamsWithLock(List<StreamSpec> intStreams) throws Exception {
+    if (!intStreams.isEmpty()) {


nit. I prefer if(intStreams.isEmpty()) return; to avoid unnecessary indentation. But that's personal preference.

sborya · 2017-08-10T23:29:30Z

samza-core/src/main/java/org/apache/samza/runtime/LocalApplicationRunner.java

+      if (coordinationUtils != null) {
+        Lock initLock = coordinationUtils.getLock();
+        LockListener lockListener = null;
+        if (coordinationUtils.getClass().getName().equals("org.apache.samza.zk.ZkCoordinationUtils")) {


I know this is a shortcut, but I think it'd be better to put this logic into the coordinationUtils, not here. LocalApplicationRunner should not know about Zk stuff.

sborya · 2017-08-10T23:33:04Z

samza-core/src/test/java/org/apache/samza/zk/TestZkLock.java

+
+package org.apache.samza.zk;
+
+public class TestZkLock {


not very efficient test :)

shanthoosh · 2017-08-11T00:40:13Z

samza-core/src/main/java/org/apache/samza/zk/ZkLock.java

+
+    //Start timer for timeout
+    ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor(new ThreadFactoryBuilder().setNameFormat("ZkLock-%d").build());
+    scheduler.schedule(() -> {


As we'd discussed offline, this exception will not get propagated to the other main
thread which is running the while loop(Since the exception thrown by the
threadPool's thread will be swallowed by threadPool).

So essentially all the followers will go into infinite loop (as long as there's a leader).

Please share a separate state (something like CountdownLatch, AtomicBoolean,
volatile boolean) between the two threads to indicate that the periodic
polling should stop.

Please correct me if i'm wrong.

Following this discussion, I have changed the lock timeout implementation. Kindly take a look at it and let me know if you have any comments. Thank you!

bharathkk · 2017-08-11T13:08:57Z

samza-core/src/main/java/org/apache/samza/coordinator/Lock.java

+  boolean hasLock();
+
+  void setLockListener(LockListener listener);
+


nit: extra line

bharathkk · 2017-08-11T13:10:28Z

samza-core/src/main/java/org/apache/samza/coordinator/Lock.java

+ */
+
+package org.apache.samza.coordinator;
+


Can you please add java docs? Also, add some explanation around the expectations from the implementers of the interface.

bharathkk · 2017-08-11T13:12:42Z

samza-core/src/main/java/org/apache/samza/coordinator/LockListener.java

+ */
+package org.apache.samza.coordinator;
+
+public interface LockListener {


Please add class level and method level java docs.

Removed the class.

bharathkk · 2017-08-11T13:19:13Z

samza-api/src/main/java/org/apache/samza/system/SystemAdmin.java

+   * @param streamSpec The spec, or blueprint for the physical stream on the system.
+   * @return true if stream exists already, false otherwise
+   */
+  default boolean existStream(StreamSpec streamSpec) {


Just throwing out a comment to see everyone's take. I usually prefer boolean method to follow some question like naming scheme to make it more readable. Since this is user facing, it is all the more important to make it as readable and explicit as possible.
For instance, I like checkIfStreamExists better than existStream.

Will keep in mind for the future, thank you!

bharathkk · 2017-08-11T13:29:48Z

samza-core/src/main/java/org/apache/samza/runtime/LocalApplicationRunner.java

+        initLock.unlock();
+      }
+
+      @Override


It is unclear as to what onError is for. If its part of LockListener class, I would assume this method gets invoked during failure cases related to lock acquisition.
However, the code here suggests a different usage of onError.

bharathkk · 2017-08-11T13:31:05Z

samza-core/src/main/java/org/apache/samza/runtime/LocalApplicationRunner.java

+          createIntermediateStreams(intStreams);
+          LOG.info("Created intermediate streams successfully!");
+        } catch (Exception e) {
+          onError();


Why invoke onError here? Refer to the comment below about clarity of the onError method.

bharathkk · 2017-08-11T13:31:40Z

samza-core/src/main/java/org/apache/samza/zk/ZkLock.java

+  public void setLockListener(LockListener listener) {
+    this.zkLockListener = listener;
+  }
+


nit: extra line

bharathkk · 2017-08-11T13:34:18Z

samza-core/src/main/java/org/apache/samza/zk/ZkLock.java

+public class ZkLock implements Lock {
+
+  public static final Logger LOG = LoggerFactory.getLogger(ZkLock.class);
+  private final static String LOCK_PATH = "lock";


consider renaming this or the lockPath instance variable.

Renamed to DistributedLock.

bharathkk · 2017-08-11T13:37:10Z

samza-core/src/main/java/org/apache/samza/zk/ZkLock.java

+        if (!hasLock.get()) {
+          throw new SamzaException("Timed out while acquiring lock for creating intermediate streams.");
+        }
+      }, zkConfig.getZkSessionTimeoutMs(), TimeUnit.MILLISECONDS);


Do we really need zkConfig? How about take the timeout in lock method or maybe use a hardcoded default or both :)

Defined a lock timeout value in LocalApplicationRunner instead of using ZkConfig.

bharathkk · 2017-08-11T15:38:29Z

samza-core/src/main/java/org/apache/samza/runtime/LocalApplicationRunner.java


+
+  private void createIntermediateStreams(List<StreamSpec> intStreams) throws Exception {
+    boolean streamsExist = getStreamManager().checkIfStreamsExist(intStreams);


Why don't we get the subset of intStreams that are not created and only try creating with them? In case of stream already exists, we do the work twice; once for checking if it exists and other time during the actual creation which would again result in no op.

Discussed offline.

prateekm · 2017-08-11T18:00:13Z

samza-api/src/main/java/org/apache/samza/system/SystemAdmin.java

+   * @param streamSpec The spec, or blueprint for the physical stream on the system.
+   * @return true if stream exists already, false otherwise
+   */
+  default boolean existStream(StreamSpec streamSpec) {


Don't have context on this PR, just wondering why we need a new method in the SystemAdmin API for this? Is this just for convenience or do we not currently have a way to check if a stream exists?

Would n't the getStreamMetadata API on SystemAdmin be sufficient?

It seems like we won't be needing this function at all. Hence removed.

vjagadish1989 · 2017-08-11T20:10:50Z

@navina :
Is this for merge into master? Or is the plan to merge this to a separate branch, clean-up and later merge back?

navina · 2017-08-11T23:21:56Z

@vjagadish1989 This is for 0.14.0 branch.

navina

Suggested couple of changes. Otherwise, looks good.

Since we are running short of time, I am ok with skipping testing for this PR. I am sure we will add it prior to the 0.14.0 release for which this is targeted.

+1 thanks!

navina · 2017-08-12T00:28:07Z

samza-api/src/main/java/org/apache/samza/system/SystemAdmin.java

    throw new UnsupportedOperationException();
  }
-}
+}


nit: can we try to get rid of these minor changes that are seemingly unrelated to the patch?

navina · 2017-08-12T00:31:35Z

samza-core/src/main/java/org/apache/samza/runtime/LocalApplicationRunner.java

+          getStreamManager().createStreams(intStreams);
+        } else {
+          LOG.error("Timed out while trying to acquire lock");
+          coordinationUtils.reset();


Can you wrap lines from Line 253 in a try-finally block? You should invoke reset even in the case where lock has been successfully acquired and critical section is complete. Adding reset in finally block will guarantee this behavior.

navina · 2017-08-12T00:33:56Z

samza-core/src/main/java/org/apache/samza/zk/ZkDistributedLock.java

+public class ZkDistributedLock implements DistributedLock {
+
+  public static final Logger LOG = LoggerFactory.getLogger(ZkDistributedLock.class);
+  private final static String LOCK_PATH = "distributed_lock";


Can you make this configurable by making it a constructor argument?
Ideally, this should be unique for different deployments of the streamapplication. Can we make this based on the hashcode of intermediate streams list for now? Check out how @bharathkk is doing it in PR #265

Made it configurable. I have reused Bharath's code for this. Kindly let me know if it's okay.

navina · 2017-08-12T00:36:10Z

samza-core/src/main/java/org/apache/samza/zk/ZkDistributedLock.java

+        return true;
+      } else {
+        // Keep trying to acquire the lock
+        long currentTime = System.currentTimeMillis();


nit: you can get rid of the local variable currentTime

navina · 2017-08-12T00:37:57Z

samza-core/src/main/java/org/apache/samza/zk/ZkDistributedLock.java

+      } else {
+        // Keep trying to acquire the lock
+        long currentTime = System.currentTimeMillis();
+        if ((currentTime - startTime) >= lockTimeout) {


Perhaps you can make this the condition for your while loop?
See if you can accommodate this change. If it doesn't work out, it's fine.

navina · 2017-08-12T00:39:48Z

samza-core/src/main/java/org/apache/samza/zk/ZkDistributedLock.java

+      nodePath = null;
+      LOG.info("Ephemeral lock node deleted. Unlocked!");
+    } else {
+      LOG.error("Ephemeral lock node you want to delete doesn't exist");


Perhaps change this warn since it doesn't actually indicate an error condition.

sborya · 2017-08-15T23:35:14Z

samza-core/src/main/java/org/apache/samza/coordinator/DistributedLock.java

+   * Releases the lock
+   */
+  void unlock();
+}


I think it would be helpful to have a close() method in the lock to be able to close the connection (separately from the unlock)

@sborya I think that we should wait a bit to see some more distributed lock implementation. If all distributed lock implementation would require init() to open a distributed resource and close() to close a distributed resource, we should add both APIs to this interface. If not, then init() and close() would be per implementation.

@nickpan47

When there’s a failure in unlock, a client of DistributedLock will not be able to distinguish if it’s because of releasing resources or due to unlock implementation. A lock owner wanting to release a lock to other processor in group, will not be able to conclude effectively if unlock has happened or not in case of exceptions(Due to two possibilities and will not be able to handle exception effectively).

Also, logically with just lock, unlock operations a client can call lock() after unlock(), essentially trying to join again which wouldn’t be possible if resource establishment happens in constructor/factory.create method(instead of separate init() lifecycle method).

Since any DistributedLock implementation will require resource establishment with a RemoteService(Zookeeper, AzureBlobStore, any), extending Autocloseable and having init() lifecycle methods, we would be clearly defining boundaries and communicating that intent with our interface(Similar to SystemProducer, SystemConsumer).

I’m not suggesting that we should do it, just interested to know your opinion.

Can you please point out why this reasoning is incorrect(not required now).

Please share your thoughts.

@shanthoosh please see my comments on the lock interface definition. You point here is exactly what I tried to say in the interface definition. You can have the lock itself maintain a heart-beat/life-cycle/lease to the remote resource, which may not require init()/close() in that case. However, it seems necessary to define a callback to the lock interface s.t. user can react to loss of possession of remote resource.

nickpan47

@PawasChhokra thanks for the patch! I mainly have two comments: 1) try not to tie the implementation of LocalApplicationRunner to any ZK-specific implementations; 2) need to consider the interface of DistributedLock to handle failure cases between the lock() and unlock().

Thanks!

nickpan47 · 2017-08-16T22:51:43Z

samza-core/src/main/java/org/apache/samza/coordinator/CoordinationUtils.java


  Latch getLatch(int size, String latchId);
+
+  DistributedLock getLock(String initLockPath);


A general comment: in a broader context, lockName sounds better than lockPath. lockPath is more ZK-specific.

nickpan47 · 2017-08-16T22:52:09Z

samza-core/src/main/java/org/apache/samza/coordinator/DistributedLock.java

+import java.util.concurrent.TimeUnit;
+
+
+public interface DistributedLock {


Would be nice to add javadoc for public interface classes.

nickpan47 · 2017-08-16T22:54:26Z

samza-core/src/main/java/org/apache/samza/coordinator/DistributedLock.java

+
+  /**
+   * Tries to acquire the lock
+   * @param timeout Duration of lock acquiring timeout.


nit: timeout to acquire a lock? Or duration of lock after acquired? I think that you were referring to the first. It would be better to clearly state it.

Yes, this is the timeout to acquire the lock. Will make it more clear in the documentation.

nickpan47 · 2017-08-16T22:57:30Z

samza-core/src/main/java/org/apache/samza/coordinator/DistributedLock.java

+   * Releases the lock
+   */
+  void unlock();
+}


@sborya I think that we should wait a bit to see some more distributed lock implementation. If all distributed lock implementation would require init() to open a distributed resource and close() to close a distributed resource, we should add both APIs to this interface. If not, then init() and close() would be per implementation.

nickpan47 · 2017-08-16T23:00:04Z

samza-core/src/main/java/org/apache/samza/runtime/LocalApplicationRunner.java

-  private static final long LEADER_ELECTION_WAIT_TIME_MS = 1000;
+  // Lock timeout is set to 10 seconds here, as we don't want to introduce a new config value currently.
+  private static final long LOCK_TIMEOUT_SECONDS = 10;
+  private static final String ZK_COORDINATION_CLASS = "org.apache.samza.zk.ZkJobCoordinatorFactory";


We should avoid more and more ZK-specific names/details to bleed into the general implementation. LocalApplicationRunner is not just for ZK-based implementation. Hence, I would recommend not to hard-code any actual ZK implementation classes here in the LocalApplicationRunner.

@nickpan47 I agree. but we have to do this string matching because we decided that the user will not configure a CoordinationServiceFactory and that we will derive it from job.coordinator.factory configuration. Are you recommending that we introduce another configuration ?

@navina I am OK if the config has a default, which is ZK-specific. I am against the hard-code ZK-specific class names in the LocalApplicationRunner implementation. I would suggest that in the LocalApplicationRunner, we load the CoordinatorServiceFactory via Utils.getInstance(JobConfig.getJobCoordinatorFactory()), w/o hard-code the ZK job coordinator class names in the LocalApplicationRunner.

nickpan47 · 2017-08-16T23:02:00Z

samza-core/src/main/java/org/apache/samza/runtime/LocalApplicationRunner.java

+    if (ZK_COORDINATION_CLASS.equals(jobCoordinatorFactoryClassName)) {
      ApplicationConfig appConfig = new ApplicationConfig(config);
-      return new ZkCoordinationServiceFactory().getCoordinationService(
+      return Util.<CoordinationServiceFactory>getObj(ZK_COORDINATION_SERVICE_CLASS).getCoordinationService(


Please remove the ZK-specific code here.

nickpan47 · 2017-08-16T23:02:36Z

samza-core/src/main/java/org/apache/samza/runtime/LocalApplicationRunner.java

+   * @throws TimeoutException exception for lock timeout
   */
-  /* package private */ void createStreams(String planId, List<StreamSpec> intStreams) throws TimeoutException {
+   /* package private */ void createStreams(String planId, List<StreamSpec> intStreams) throws TimeoutException {


nit: unnecessary extra leading white space.

nickpan47 · 2017-08-16T23:44:47Z

samza-core/src/main/java/org/apache/samza/zk/ZkLock.java

+      int index = children.indexOf(ZkKeyBuilder.parseIdFromPath(nodePath));
+
+      if (children.size() == 0 || index == -1) {
+        throw new SamzaException("Looks like we are no longer connected to Zk. Need to reconnect!");


I don't see any error handling logic here (e.g. try to re-connect within the session expiration, etc.). Is it on the contract to the caller of DistributedLock.lock() to handle those exceptions? If yes, it would be necessary to define the exception types the DistributedLock.lock() method would throw and clearly state in the document what the caller supposed to do when catch those exceptions. Otherwise, the implementer of lock() should just handle the specific exceptions internally (i.e. re-try) until timeout.

nickpan47 · 2017-08-16T23:45:39Z

samza-core/src/main/java/org/apache/samza/zk/ZkLock.java

+        return true;
+      } else {
+        try {
+          Thread.sleep(random.nextInt(1000));


I thought that in ZK implementation, we can use watch() on children?

nickpan47 · 2017-08-16T23:52:04Z

samza-core/src/main/java/org/apache/samza/zk/ZkLock.java

+  @Override
+  public void unlock() {
+    if (nodePath != null) {
+      zkUtils.getZkClient().delete(nodePath);


So, I would recommend to think it through about the zkClient lifecycle needed in this lock implementation. Are we assuming that the zkClient is initialized by some other class and just passed in to this lock implementation? Or are we assuming that the lock would internally own the lifecycle of the zkClient? If we are assuming the first case, I think that this DistributedLock implementation lack of handling on the failure cases: what if the zkClient connection to the ZK server was dropped after the user has called lock() and proceeded into the critical section? If the disconnect time is longer than the session timeout, there would be race conditions. How do the users of the DistributedLock being notified of this issue after calling the lock()?

prateekm · 2018-03-26T23:09:33Z

@PawasChhokra Does this still need to be merged? If not, can we close this PR?

PawasChhokra · 2018-03-28T19:03:07Z

@prateekm This has been implemented already. Closing the pull request now. Thanks for the reminder!

navina mentioned this pull request Aug 10, 2017

SAMZA-1385: Fix zookeeper path conflict between LocalApplicationRunner and ZkJobCoordinator #265

Closed

sborya requested changes Aug 10, 2017

View reviewed changes

shanthoosh reviewed Aug 11, 2017

View reviewed changes

bharathkk reviewed Aug 11, 2017

View reviewed changes

prateekm reviewed Aug 11, 2017

View reviewed changes

navina approved these changes Aug 12, 2017

View reviewed changes

sborya mentioned this pull request Aug 14, 2017

Refactor coordination utils (with the ZkLock) #271

Closed

sborya approved these changes Aug 15, 2017

View reviewed changes

PawasChhokra added 12 commits August 16, 2017 14:31

Add lock

8fb1ef6

Add create streams with lock

8ec4e59

Fix retry

9750f2d

Add lock timeout

83a9644

Change lock API

afbddac

Address review

03f1bc7

Add lock

774903a

Add create streams with lock

80503f8

Fix retry

23849c8

Add lock timeout

5cf9d37

Removing existStreams

6a790b2

Rebasing with master

ad94c91

PawasChhokra force-pushed the ZkLock branch from 0440627 to ad94c91 Compare August 16, 2017 22:37

nickpan47 requested changes Aug 16, 2017

View reviewed changes

PawasChhokra closed this Mar 28, 2018


		boolean hasLock();

		void setLockListener(LockListener listener);



		private void createIntermediateStreams(List<StreamSpec> intStreams) throws Exception {
		boolean streamsExist = getStreamManager().checkIfStreamsExist(intStreams);


		Latch getLatch(int size, String latchId);

		DistributedLock getLock(String initLockPath);

		import java.util.concurrent.TimeUnit;


		public interface DistributedLock {

SAMZA-1372: Change Latch Interface to Lock Interface for Samza Standalone with ZK #264

SAMZA-1372: Change Latch Interface to Lock Interface for Samza Standalone with ZK #264

Uh oh!

Conversation

PawasChhokra commented Aug 9, 2017

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

PawasChhokra Aug 11, 2017 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

shanthoosh Aug 11, 2017 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

PawasChhokra Aug 11, 2017 •

edited

Loading

shanthoosh Aug 11, 2017 •

edited

Loading

vjagadish1989 commented Aug 11, 2017 •

edited

Loading