liuxunorg commented on a change in pull request #3397: [ZEPPELIN-4208] Cluster
synchronize InterpreterSetting
URL: https://github.com/apache/zeppelin/pull/3397#discussion_r300609107
##########
File path:
zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/InterpreterSettingManager.java
##########
@@ -978,4 +1041,85 @@ public void onParagraphUpdate(Paragraph p) throws
IOException {
public void onParagraphStatusChange(Paragraph p, Job.Status status) throws
IOException {
}
+
+ @Override
+ public void onClusterEvent(String msg) {
+ if (LOGGER.isDebugEnabled()) {
+ LOGGER.debug("onClusterEvent : {}", msg);
+ }
+
+ try {
+ Gson gson = new Gson();
+ ClusterMessage message = ClusterMessage.deserializeMessage(msg);
+ String id = message.get("id");
+ String name = message.get("name");
+ String group = message.get("group");
+ InterpreterOption option = null;
+ Map<String, InterpreterProperty> properties = null;
+ List<Dependency> dependencies = null;
+ String jsonOption = message.get("option");
+ if (!StringUtils.isBlank(jsonOption)) {
+ option = InterpreterOption.fromJson(jsonOption);
+ }
+ String jsonProperties = message.get("properties");
+ if (!StringUtils.isBlank(jsonProperties)) {
+ properties = gson.fromJson(jsonProperties,
+ new TypeToken<Map<String, InterpreterProperty>>() {}.getType());
+ }
+ String jsonDependencies = message.get("dependencies");
+ if (!StringUtils.isBlank(jsonOption)) {
+ dependencies = gson.fromJson(jsonDependencies, new
TypeToken<List<Dependency>>() {}.getType());
+ }
+
+ switch (message.clusterEvent) {
+ case CREATE_INTP_SETTING:
+ inlineCreateNewSetting(name, group, dependencies, option,
properties);
+ break;
+ case UPDATE_INTP_SETTING:
+ inlineSetPropertyAndRestart(id, option, properties, dependencies,
false);
+ break;
+ case DELETE_INTP_SETTING:
+ inlineRemove(id, false);
+ break;
+ default:
+ LOGGER.error("Unknown clusterEvent:{}, msg:{} ",
message.clusterEvent, msg);
+ break;
+ }
+ } catch (IOException e) {
+ LOGGER.error(e.getMessage(), e);
+ } catch (InterpreterException e) {
+ LOGGER.error(e.getMessage(), e);
+ }
+ }
+
+ // broadcast cluster event
+ private void broadcastClusterEvent(ClusterEvent event, InterpreterSetting
intpSetting) {
+ if (!conf.isClusterMode()) {
+ return;
+ }
+
+ List<Dependency> dependencies = intpSetting.getDependencies();
+ Map<String, InterpreterProperty> properties
+ = (Map<String, InterpreterProperty>) intpSetting.getProperties();
+ InterpreterOption intpOption = intpSetting.getOption();
+
+ HashMap<String, String> params = new HashMap<>();
+ String jsonDep = gson.toJson(dependencies, new
TypeToken<List<Dependency>>() {
+ }.getType());
+ String jsonProps = gson.toJson(properties, new TypeToken<Map<String,
InterpreterProperty>>() {
+ }.getType());
+ params.put("id", intpSetting.getId());
Review comment:
If I replace the entire `InterpreterSetting` with json, I am worried that
the data is too large, and I only need data for `dependencies` and
`InterpreterProperty`, and I don't need the full `InterpreterSetting` data.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
