TEZ-4007: Introduce AmExtensions and Zookeeper-based FrameworkServices #427
+3,487
−328
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This comment was marked as outdated.
This comment was marked as outdated.
okumin
reviewed
Sep 11, 2025
| * Curator/Zookeeper impl of AMRegistryClient | ||
| */ | ||
| @InterfaceAudience.Public | ||
| public class ZkAMRegistryClient extends AMRegistryClient { |
Contributor
There was a problem hiding this comment.
Just a question. What is the duty of ZooKeeper here? When I read this patch a while ago, I presumed ZooKeeper would be used for two reasons. My memory might be wrong.
- Generate a unique application ID
- Service discovery for application masters
Contributor
Author
There was a problem hiding this comment.
@okumin: both:
- ZkAMRegistry.generateNewId takes care of generating applicationId (backed by zookeeper)
- it also for service discovery (AM registers itself, client receives updates about that)
I'm happy to see that you're interested in the area!
please be aware that it's under refactor, I'll let you know, when it's ready
Contributor
There was a problem hiding this comment.
Thanks. Background: I wonder if it can be independent of ZooKeeper technically. The reliability and fault-tolerance of ZK are beneficial in many environments, but in other cases, high availability is not necessary. Trino does not provide fault-tolerance with its coordinator, and some users appreciate the ease of use.
Contributor
Author
There was a problem hiding this comment.
absolutely! be aware that these patches have been working for years in downstream distributions. This PR is about contributing them back, so apparently, it's ZK the long-term plan is to improve abstractions and make it more pluggable, and replace ZK with Kubernetes etcd, for example, but the default (yarn-based) behavior doesn't utilize an AM registry at all
can you please elaborate on how Trino's lack of fault tolerance comes to this picture? I mean, this ZK-based registry is more about AM service discovery, not fault tolerance
Contributor
There was a problem hiding this comment.
Hi folks, sorry, I'd like to chime in with a question here. :)
From the description of TEZ-3991, this feature seems to allow the AM to manage its own lifecycle, while clients such as HiveServer2 only need to discover available AMs via ZooKeeper (ZK) and submit tasks.
Based on my current limited understanding, the benefits of this feature should include reducing the burden on HS2 and potentially making task submission more efficient.
However, I have a few questions:
-
Is this feature primarily beneficial for LLAP mode? Since LLAP requires long-running Tez AMs, and IMO this feature seems more useful for long-running tasks.
-
Could this functionality serve as preliminary groundwork for decoupling the Tez framework from YARN? For example, could it pave the way for Tez to run directly on Kubernetes without relying on YARN for resource management?
Thanks.
Contributor
There was a problem hiding this comment.
be aware that these patches have been working for years in downstream distributions
Sounds excellent.
can you please elaborate on how Trino's lack of fault tolerance comes to this picture?
Trino must have SPOF, but still attracts users. Hive on Tez (with LLAP) without fault tolerance is easier to set up than Hive on Tez with fault tolerance. I expect the easy quick start should attract users.
abstractdog
changed the title
abstractdog
force-pushed
the
TEZ-4007
branch
from
4702ed6 to
c24caeb
Compare
This comment was marked as outdated.
This comment was marked as outdated.
This comment was marked as outdated.
This comment was marked as outdated.
This comment was marked as outdated.
This comment was marked as outdated.
abstractdog
force-pushed
the
TEZ-4007
branch
2 times, most recently
from
d3c969a to
b3ae071
Compare
This comment was marked as outdated.
This comment was marked as outdated.
This comment was marked as outdated.
This comment was marked as outdated.
abstractdog
changed the title
This comment was marked as outdated.
This comment was marked as outdated.
abstractdog
force-pushed
the
TEZ-4007
branch
5 times, most recently
from
ec30327 to
db11fb6
Compare
This comment was marked as outdated.
This comment was marked as outdated.
This comment was marked as outdated.
This comment was marked as outdated.
This comment was marked as outdated.
This comment was marked as outdated.
This comment was marked as outdated.
This comment was marked as outdated.
abstractdog
force-pushed
the
TEZ-4007
branch
2 times, most recently
from
2e44662 to
eec88ab
Compare
This comment was marked as outdated.
This comment was marked as outdated.
abstractdog
force-pushed
the
TEZ-4007
branch
from
50a3c90 to
32070fc
Compare
This comment was marked as outdated.
This comment was marked as outdated.
abstractdog
force-pushed
the
TEZ-4007
branch
from
32070fc to
67cc981
Compare
…s - checkstyle, spotbugs, javadoc improvements, refactor, test fixes
abstractdog
force-pushed
the
TEZ-4007
branch
from
67cc981 to
021a52b
Compare
This comment was marked as outdated.
This comment was marked as outdated.
This comment was marked as outdated.
This comment was marked as outdated.
ayushtkn
reviewed
Nov 21, 2025
Member
ayushtkn
left a comment
ayushtkn
left a comment
There was a problem hiding this comment.
Thanx @abstractdog for the PR, Had a first pass, dropped some comments
Comment on lines
76
to
78
| //externalId is optional, if not provided, convert to empty string | ||
| this.externalId = (externalId == null) ? "" : externalId; | ||
| this.computeName = (computeName == null) ? ZkConfig.DEFAULT_COMPUTE_GROUP_NAME : computeName; |
Member
There was a problem hiding this comment.
nit, can we do something like:
this.externalId = Optional.ofNullable(externalId).orElse("");
this.computeName = Optional.ofNullable(computeName).orElse(ZkConfig.DEFAULT_COMPUTE_GROUP_NAME);
Comment on lines
126
to
154
| if (other instanceof AMRecord otherRecord) { | ||
| if (other instanceof AMRecord) { | ||
| AMRecord otherRecord = (AMRecord) other; |
Member
There was a problem hiding this comment.
it can be done this way
if (other instanceof AMRecord otherRecord) {
return appId.equals(otherRecord.appId)
&& hostName.equals(otherRecord.hostName)
&& hostIp.equals(otherRecord.hostIp)
&& port == otherRecord.port
&& externalId.equals(otherRecord.externalId)
&& computeName.equals(otherRecord.computeName);
}
Comment on lines
+41
to
+42
| amRegistry.init(conf); | ||
| amRegistry.start(); |
Contributor
Author
There was a problem hiding this comment.
added javadoc comment to clarify this:
/**
* Returns a singleton {@link AMRegistry} instance backed by ZooKeeper.
*
* <p>If the registry has not yet been created, this method initializes and starts
* a new {@link ZkAMRegistry} using the external AM identifier obtained from the
* {@code TEZ_AM_EXTERNAL_ID} environment variable.</p>
*
* <p>When the registry is used as a service within the DAGAppMaster, the
* DAGAppMaster is responsible for managing its lifecycle, including closure.</p>
*
* @param conf the configuration used to initialize the registry; must not be null
* @return the initialized and started {@link AMRegistry} instance
* @throws IllegalStateException if the {@code TEZ_AM_EXTERNAL_ID} environment variable is not set
* @throws RuntimeException if an error occurs while creating, initializing, or starting the registry
*/
|
|
||
| @Override | ||
| public AMExtensions getAMExtensions() { | ||
| return new ZkStandaloneAMExtensions(this); |
Member
There was a problem hiding this comment.
we are creating new instance on every invocation? should we create once & then send back the same if not null?
|
|
||
| package org.apache.tez.client.registry.zookeeper; | ||
|
|
||
| import static org.junit.Assert.*; |
|
|
||
| private TezConfiguration getTezConfForZkDiscovery() { | ||
| TezConfiguration tezConf = new TezConfiguration(); | ||
| tezConf.set(TezConfiguration.TEZ_FRAMEWORK_MODE, "STANDALONE_ZOOKEEPER"); |
| } | ||
| } | ||
| assertTrue("AM record should be in getAllRecords", found); | ||
| LOG.info("Test completed successfully. AM was discovered: {}", amRecord); |
Member
There was a problem hiding this comment.
I am not sure what value does this log line add :-)
abstractdog
force-pushed
the
TEZ-4007
branch
from
1b01383 to
7f72e4a
Compare
This comment was marked as outdated.
This comment was marked as outdated.
abstractdog
force-pushed
the
TEZ-4007
branch
from
7f72e4a to
4c9ee7f
Compare
This comment was marked as outdated.
This comment was marked as outdated.
This comment was marked as outdated.
This comment was marked as outdated.
abstractdog
force-pushed
the
TEZ-4007
branch
from
4c9ee7f to
731c4df
Compare
|
🎊 +1 overall
This message was automatically generated. |
abstractdog
force-pushed
the
TEZ-4007
branch
from
731c4df to
045ff68
Compare
|
🎊 +1 overall
This message was automatically generated. |
Contributor
Author
|
@ayushtkn : all your comments have been addressed, fixed many things according to them, which made this area much better, thanks! |
|
🎊 +1 overall
This message was automatically generated. |
|
🎊 +1 overall
This message was automatically generated. |
|
🎊 +1 overall
This message was automatically generated. |
|
🎊 +1 overall
This message was automatically generated. |
ayushtkn
reviewed
Dec 11, 2025
| if (other instanceof AMRecord otherRecord) { | ||
| return appId.equals(otherRecord.appId) | ||
| && host.equals(otherRecord.host) | ||
| && hostName.equals(otherRecord.hostName) |
Member
There was a problem hiding this comment.
can this lead to NPE, we don't have any default here
this.hostName = serviceRecord.get(HOST_NAME_RECORD_KEY);
|
|
||
| @Override | ||
| public String toString() { | ||
| return toServiceRecord().attributes().toString(); |
Member
There was a problem hiding this comment.
toString calls toServiceRecord, So, it can lead to initializing the serviceRecord = new ServiceRecord();, are we ok with it. This could lead to issues later, like someone logged it or so, before setting the values and then we have the service record cached as well
| AMRecord createAmRecord(ApplicationId appId, String hostName, String hostIp, int port, | ||
| String computeName); | ||
|
|
||
| void close(); |
Member
There was a problem hiding this comment.
AutoCloseable already defines close, do we need to define again here?
Comment on lines
+42
to
+44
| void add(AMRecord server) throws Exception; | ||
|
|
||
| void remove(AMRecord server) throws Exception; |
| * <p>Listeners may be registered to receive notifications when AM records | ||
| * appear or are removed.</p> | ||
| */ | ||
| public abstract class AMRegistryClient implements Closeable { |
Member
There was a problem hiding this comment.
can we do implements AutoCloseable?
| @Override | ||
| public void start() { | ||
| try { | ||
| amRegistryClient.start(); |
Member
There was a problem hiding this comment.
we should return if isRunning is true here
|
|
||
| public class ZkFrameworkClient extends FrameworkClient { | ||
|
|
||
| private AMRecord amRecord; |
| } else if (modeInEnv != null) { | ||
| return getByMode(interfaze, modeInEnv); | ||
| } else if (defaultClazz != null) { | ||
| return (T) defaultClazz.newInstance(); |
Member
There was a problem hiding this comment.
I think this is deprecated, does this work
defaultClazz.getDeclaredConstructor().newInstance();
| String modeInConf = conf != null ? conf.get(TezConfiguration.TEZ_FRAMEWORK_MODE) : null; | ||
| String modeInEnv = System.getenv(TezConstants.TEZ_FRAMEWORK_MODE); | ||
| try { | ||
| if (modeInConf != null) { |
Member
There was a problem hiding this comment.
should it be StringUtils.isNotEmpty()?
| .thenReturn(YarnApplicationState.RUNNING); | ||
|
|
||
| //Client 1 start | ||
| client.start(); |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
5 participants
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
This PR contains a Zookeeper-based Framework service and a major refactor in this area.
Pieces of this PR: