Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Add TTL support for compressedBucketWrite fn #2930

Closed
wants to merge 11 commits into from
+180 −16
Closed

Add TTL support for compressedBucketWrite fn #2930

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
11 commits
Select commit Hold shift + click to select a range
da1267c
Add TTL support for compressedBucketWrite fn
Sep 30, 2024
7dcbefb
Added ttl field in ZkBucketDataAccessor
Sep 30, 2024
7da6545
Add java doc.
jacoblukose Oct 1, 2024
e04205a
Add tests
jacoblukose Oct 2, 2024
187f593
Added zkBucketDataAccessor test
jacoblukose Oct 9, 2024
d0258c5
Misc
jacoblukose Oct 9, 2024
6cfe10a
Remove unused import
jacoblukose Oct 11, 2024
019f66f
Correct assertfailure message
jacoblukose Oct 11, 2024
6033d9f
Add java doc for test
jacoblukose Oct 15, 2024
aa0bc5f
Changes based on PR comments
jacoblukose Nov 5, 2024
3d60083
Based on PR comments
jacoblukose Nov 8, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
21 changes: 17 additions & 4 deletions helix-core/src/main/java/org/apache/helix/manager/zk/ZkBaseDataAccessor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -419,10 +419,14 @@ public boolean update(String path, DataUpdater<T> updater, int options) {
return result._retCode == RetCode.OK;
}

public AccessResult doUpdate(String path, DataUpdater<T> updater, int options) {
return doUpdate(path, updater, options, ZkClient.TTL_NOT_SET);
}

/**
* sync update
*/
public AccessResult doUpdate(String path, DataUpdater<T> updater, int options) {
public AccessResult doUpdate(String path, DataUpdater<T> updater, int options, long ttl) {
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Normally helper functions with this naming convention should be private.

Copy link
Author

@jacoblukose jacoblukose Nov 6, 2024 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I checked other fns (ex: doCreate, doSet and also the other invocation of doUpdate) and they are having public as the scope.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Normally if the fuc is not used outside class, we would set as private.
Also more importantly, it is a bit odd to allow user to set ttl but not actually update ttl in update. If user provide a new ttl value and the node already exit, this function will return RetCode.OK without actually change anything.

Copy link
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

as I said earlier there are several instances of doCreate, doSet, doUpdate which are declared public. I followed the similar convention. I am not familiar with the helix code base, this is my first PR. If you can guide me what exact change to make here, happy to correct it.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

+1 on @xyuanlu.

I can understand TTL for create. But why we need TTL for update? The right operation is to let the user delete the node and recreate with a TTL.

I am not confortable with this feature. Before we have strong justification. I dont think we should let this in.

@jacoblukose It would be good to figure out why you would like to support this.

Copy link
Contributor

@xyuanlu xyuanlu Nov 18, 2024 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I feel like this might not be the best approach, but I am OK as long as @junkaixue feels OK. :)

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I am not buying this. As I said, when a node is created, we should not update its property and change it from a persistent node to TTL node.

If you need to do this, then it should be the application logic to make that smoothly migrate from Non TTL to TTL. Also, since this is the API support Helix. I dont see any use cases in HELIX to leverage this.

Accessor is Helix internal thing. If you would like to leverage for company specific feature and not necessary help Helix, please fork the repo and implement it internally.

Copy link
Author

@jacoblukose jacoblukose Nov 19, 2024 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@xyuanlu thanks for reponding and on the ack. It will be helpful if you can also callout alternative approach here to improve the craft if you thinks thats seen comprised. Happy to correct it.

@junkaixue thanks for responding, would you be kind enough to highlight the code part where the said concern: I am not buying this. As I said, when a node is created, we should not update its property and change it from a persistent node to TTL node. is. doUpdate fn is not going to update ttl for an existing node, only when underlying create is called, it leverage ttl passed if any. And also, if you see, public facing doUpdate doesnt have ttl option exposed. It would be great if you can suggest an alternative on how this need to handled as well.

Also I think there is some misunderstanding, the usecase is not to change znodes created with persistent mode to migrate to use ttl mode. Idea is to just expand the scope of ZkBucketDataAccessor to leverage zk ttl feature when the same is being initialized.

Copy link
Contributor

@xyuanlu xyuanlu Nov 19, 2024 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

To be honest I think more time is needed when design the ZK update API to keep consistency. I dont have an solid idea in mind how. But I think the fundamental rule is keep the behavior simple, consistent and not ambiguous.

  • The Update API has a TTL as parameter, but the updated ZNode may or may not have the passed in TTL value, is not a consistent behavior. However, I don't have an idea of how to achieve this, I think this is the reason why we don't have update with TTL in the first place. More effort is needed when design this - if ever needed.

Copy link
Author

@jacoblukose jacoblukose Nov 19, 2024 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

thanks @xyuanlu the only reason why doUpdate need a ttl value to be passed in is because the current code is using a doCreate fn inside the doUpdate. If there is no doCreate call made within doUpdate then there is no need to pass in the ttl as well, reason being, ttl can only be enforced during creation of znode.

the related q is, is it a correct behaviour on the existing doUpdate fn code to call a doCreate internally? If not, this should be removed, and application need to handle it by issuing an explicit doCreate call. if its correct, then comes to the challenge on how to pass in ttl value to the internal doCreate fn inside doUpdate

AccessResult result = new AccessResult();
CreateMode mode = AccessOption.getMode(options);
if (mode == null) {
Expand Down Expand Up @@ -452,7 +456,7 @@ public AccessResult doUpdate(String path, DataUpdater<T> updater, int options) {
T newData = updater.update(null);
RetCode rc;
if (newData != null) {
AccessResult res = doCreate(path, newData, options);
AccessResult res = doCreate(path, newData, options, ttl);
result._pathCreated.addAll(res._pathCreated);
rc = res._retCode;
} else {
Expand Down Expand Up @@ -979,11 +983,20 @@ public boolean[] setChildren(List<String> paths, List<T> records, int options) {
return set(paths, records, null, null, options);
}

public boolean[] setChildren(List<String> paths, List<T> records, int options, long ttl) {
return set(paths, records, null, null, options, ttl);
}

boolean[] set(List<String> paths, List<T> records, List<List<String>> pathsCreated,
List<Stat> stats, int options) {
return set(paths, records, pathsCreated, stats, options, ZkClient.TTL_NOT_SET);
}

/**
* async set, give up on error other than NoNode
*/
boolean[] set(List<String> paths, List<T> records, List<List<String>> pathsCreated,
List<Stat> stats, int options) {
List<Stat> stats, int options, long ttl) {
if (paths == null || paths.size() == 0) {
return new boolean[0];
}
Expand Down Expand Up @@ -1051,7 +1064,7 @@ boolean[] set(List<String> paths, List<T> records, List<List<String>> pathsCreat
// if failOnNoNode, try create
if (failOnNoNode) {
boolean[] needCreate = Arrays.copyOf(needSet, needSet.length);
createCbList = create(paths, records, needCreate, pathsCreated, options);
createCbList = create(paths, records, needCreate, pathsCreated, options, ttl);
for (int i = 0; i < createCbList.length; i++) {
ZkAsyncCallbacks.CreateCallbackHandler createCb = createCbList[i];
if (createCb == null) {
Expand Down
19 changes: 13 additions & 6 deletions helix-core/src/main/java/org/apache/helix/manager/zk/ZkBucketDataAccessor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -48,6 +48,7 @@
import org.apache.helix.zookeeper.impl.factory.DedicatedZkClientFactory;
import org.apache.helix.zookeeper.util.GZipCompressionUtil;
import org.apache.helix.zookeeper.zkclient.DataUpdater;
import org.apache.helix.zookeeper.zkclient.ZkClient;
import org.apache.helix.zookeeper.zkclient.exception.ZkNoNodeException;
import org.apache.helix.zookeeper.zkclient.serialize.ZkSerializer;
import org.slf4j.Logger;
Expand Down Expand Up @@ -76,6 +77,7 @@ public class ZkBucketDataAccessor implements BucketDataAccessor, AutoCloseable {

private final int _bucketSize;
private final long _versionTTLms;
private final long _znodeTTLms;
private final ZkSerializer _zkSerializer;
private final RealmAwareZkClient _zkClient;
private final ZkBaseDataAccessor<byte[]> _zkBaseDataAccessor;
Expand All @@ -89,25 +91,30 @@ public class ZkBucketDataAccessor implements BucketDataAccessor, AutoCloseable {
* @param versionTTLms in ms
*/
public ZkBucketDataAccessor(String zkAddr, int bucketSize, long versionTTLms) {
this(createRealmAwareZkClient(zkAddr), bucketSize, versionTTLms, false);
this(createRealmAwareZkClient(zkAddr), bucketSize, versionTTLms, false, ZkClient.TTL_NOT_SET);
}

public ZkBucketDataAccessor(RealmAwareZkClient zkClient) {
this(zkClient, DEFAULT_BUCKET_SIZE, DEFAULT_VERSION_TTL, true);
this(zkClient, DEFAULT_BUCKET_SIZE, DEFAULT_VERSION_TTL, true, ZkClient.TTL_NOT_SET);
}

public ZkBucketDataAccessor(RealmAwareZkClient zkClient, int bucketSize, long versionTTLms) {
this(zkClient, bucketSize, versionTTLms, true);
this(zkClient, bucketSize, versionTTLms, true, ZkClient.TTL_NOT_SET);
}

public ZkBucketDataAccessor(RealmAwareZkClient zkClient, int bucketSize, long versionTTLms, long znodeTTLms) {
this(zkClient, bucketSize, versionTTLms, true, znodeTTLms);
}

private ZkBucketDataAccessor(RealmAwareZkClient zkClient, int bucketSize, long versionTTLms,
boolean usesExternalZkClient) {
boolean usesExternalZkClient, long znodeTTLms) {
_zkClient = zkClient;
_zkBaseDataAccessor = new ZkBaseDataAccessor<>(_zkClient);
_zkSerializer = new ZNRecordJacksonSerializer();
_bucketSize = bucketSize;
_versionTTLms = versionTTLms;
_usesExternalZkClient = usesExternalZkClient;
_znodeTTLms = znodeTTLms;
}

/**
Expand Down Expand Up @@ -159,7 +166,7 @@ public <T extends HelixProperty> boolean compressedBucketWrite(String rootPath,

// 1. Increment lastWriteVersion using DataUpdater
ZkBaseDataAccessor.AccessResult result = _zkBaseDataAccessor.doUpdate(
rootPath + "/" + LAST_WRITE_KEY, lastWriteVersionUpdater, AccessOption.PERSISTENT);
Copy link
Author

@jacoblukose jacoblukose Nov 19, 2024 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

note(myself): based on how pr discussion concludes, we might need to use explicit create for first time and then update calls from second time onwards.

rootPath + "/" + LAST_WRITE_KEY, lastWriteVersionUpdater, AccessOption.PERSISTENT, _znodeTTLms);
if (result._retCode != ZkBaseDataAccessor.RetCode.OK) {
throw new HelixException(
String.format("Failed to write the write version at path: %s!", rootPath));
Expand Down Expand Up @@ -206,7 +213,7 @@ public <T extends HelixProperty> boolean compressedBucketWrite(String rootPath,
buckets.add(binaryMetadata);

// Do an async set to ZK
boolean[] success = _zkBaseDataAccessor.setChildren(paths, buckets, AccessOption.PERSISTENT);
boolean[] success = _zkBaseDataAccessor.setChildren(paths, buckets, AccessOption.PERSISTENT, _znodeTTLms);
Copy link
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think if we using ttl, accessOption also need to be updated to use PERSISTENT_WITH_TTL

// Exception and fail the write if any failed
for (boolean s : success) {
if (!s) {
Expand Down