[NO ISSUE] Expose concurrent map, not a wrapper
- user model changes: no
- storage format changes: no
- interface changes: yes
Details:
JobSerializerDeserializerContainer is a ConcurrentHashMap wrapper. This
is unnecessary code, but more than that with concurrency it makes it
hard to implement things correctly.
For example, there were unnecessary synchronized methods. Also, "deploy"
in DeploymentUtils did not update the map atomically, because the
wrapper did not expose computeIfAbsent.
This is the simplest case I found, other over-aggressive/non-atomic
synchronization will be harder to clean up.
Change-Id: Ibf659ef4306d8a14b8b9f4345b2fe10ef7ef489c
Reviewed-on: https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/11965
Integration-Tests: Jenkins <jenkins@fulliautomatix.ics.uci.edu>
Reviewed-by: Michael Blow <mblow@apache.org>
Tested-by: Michael Blow <mblow@apache.org>
diff --git a/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/application/IServiceContext.java b/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/application/IServiceContext.java
index 0a99c45..7145f50 100644
--- a/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/application/IServiceContext.java
+++ b/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/application/IServiceContext.java
@@ -19,11 +19,13 @@
package org.apache.hyracks.api.application;
import java.io.Serializable;
+import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ThreadFactory;
import org.apache.hyracks.api.config.IApplicationConfig;
+import org.apache.hyracks.api.deployment.DeploymentId;
import org.apache.hyracks.api.io.IPersistedResourceRegistry;
-import org.apache.hyracks.api.job.IJobSerializerDeserializerContainer;
+import org.apache.hyracks.api.job.IJobSerializerDeserializer;
import org.apache.hyracks.api.messages.IMessageBroker;
import org.apache.hyracks.api.service.IControllerService;
@@ -41,7 +43,7 @@
IMessageBroker getMessageBroker();
- IJobSerializerDeserializerContainer getJobSerializerDeserializerContainer();
+ ConcurrentMap<DeploymentId, IJobSerializerDeserializer> getJobSerializerDeserializerContainer();
ThreadFactory getThreadFactory();
diff --git a/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/job/IJobSerializerDeserializerContainer.java b/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/job/IJobSerializerDeserializerContainer.java
deleted file mode 100644
index e63c18b..0000000
--- a/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/job/IJobSerializerDeserializerContainer.java
+++ /dev/null
@@ -1,49 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-package org.apache.hyracks.api.job;
-
-import org.apache.hyracks.api.deployment.DeploymentId;
-
-public interface IJobSerializerDeserializerContainer {
-
- /**
- * Get the IJobSerializerDeserializer implementation instance for a specific deployment id
- *
- * @param deploymentId
- * @return
- */
- public IJobSerializerDeserializer getJobSerializerDeserializer(DeploymentId deploymentId);
-
- /**
- * Add a deployment with the job serializer deserializer
- *
- * @param deploymentId
- * @param jobSerDe
- */
- public void addJobSerializerDeserializer(DeploymentId deploymentId, IJobSerializerDeserializer jobSerDe);
-
- /**
- * Remove a deployment
- *
- * @param deploymentId
- */
- public void removeJobSerializerDeserializer(DeploymentId deploymentId);
-
-}
diff --git a/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/job/JobSerializerDeserializerContainer.java b/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/job/JobSerializerDeserializerContainer.java
deleted file mode 100644
index 7f3194e..0000000
--- a/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/job/JobSerializerDeserializerContainer.java
+++ /dev/null
@@ -1,58 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-package org.apache.hyracks.api.job;
-
-import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
-
-import org.apache.hyracks.api.deployment.DeploymentId;
-
-public class JobSerializerDeserializerContainer implements IJobSerializerDeserializerContainer {
-
- private IJobSerializerDeserializer defaultJobSerDe = new JobSerializerDeserializer();
- private Map<DeploymentId, IJobSerializerDeserializer> jobSerializerDeserializerMap =
- new ConcurrentHashMap<DeploymentId, IJobSerializerDeserializer>();
-
- @Override
- public synchronized IJobSerializerDeserializer getJobSerializerDeserializer(DeploymentId deploymentId) {
- if (deploymentId == null) {
- return defaultJobSerDe;
- }
- IJobSerializerDeserializer jobSerDe = jobSerializerDeserializerMap.get(deploymentId);
- return jobSerDe;
- }
-
- @Override
- public synchronized void addJobSerializerDeserializer(DeploymentId deploymentId,
- IJobSerializerDeserializer jobSerDe) {
- jobSerializerDeserializerMap.put(deploymentId, jobSerDe);
- }
-
- @Override
- public synchronized void removeJobSerializerDeserializer(DeploymentId deploymentId) {
- jobSerializerDeserializerMap.remove(deploymentId);
- }
-
- @Override
- public String toString() {
- return jobSerializerDeserializerMap.toString();
- }
-
-}
diff --git a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-common/src/main/java/org/apache/hyracks/control/common/application/ServiceContext.java b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-common/src/main/java/org/apache/hyracks/control/common/application/ServiceContext.java
index b6d32a9..411b134 100644
--- a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-common/src/main/java/org/apache/hyracks/control/common/application/ServiceContext.java
+++ b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-common/src/main/java/org/apache/hyracks/control/common/application/ServiceContext.java
@@ -19,14 +19,16 @@
package org.apache.hyracks.control.common.application;
import java.io.Serializable;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ThreadFactory;
import org.apache.hyracks.api.application.IServerContext;
import org.apache.hyracks.api.application.IServiceContext;
import org.apache.hyracks.api.config.IApplicationConfig;
+import org.apache.hyracks.api.deployment.DeploymentId;
import org.apache.hyracks.api.io.IPersistedResourceRegistry;
-import org.apache.hyracks.api.job.IJobSerializerDeserializerContainer;
-import org.apache.hyracks.api.job.JobSerializerDeserializerContainer;
+import org.apache.hyracks.api.job.IJobSerializerDeserializer;
import org.apache.hyracks.api.messages.IMessageBroker;
public abstract class ServiceContext implements IServiceContext {
@@ -35,7 +37,7 @@
protected ThreadFactory threadFactory;
protected Serializable distributedState;
protected IMessageBroker messageBroker;
- protected IJobSerializerDeserializerContainer jobSerDeContainer = new JobSerializerDeserializerContainer();
+ protected ConcurrentMap<DeploymentId, IJobSerializerDeserializer> jobSerDeContainer = new ConcurrentHashMap<>();
protected IPersistedResourceRegistry persistedResourceRegistry;
public ServiceContext(IServerContext serverCtx, IApplicationConfig appConfig, ThreadFactory threadFactory) {
@@ -60,7 +62,7 @@
}
@Override
- public IJobSerializerDeserializerContainer getJobSerializerDeserializerContainer() {
+ public ConcurrentMap<DeploymentId, IJobSerializerDeserializer> getJobSerializerDeserializerContainer() {
return this.jobSerDeContainer;
}
diff --git a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-common/src/main/java/org/apache/hyracks/control/common/deployment/DeploymentUtils.java b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-common/src/main/java/org/apache/hyracks/control/common/deployment/DeploymentUtils.java
index ceb0da8..0b33d1c 100644
--- a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-common/src/main/java/org/apache/hyracks/control/common/deployment/DeploymentUtils.java
+++ b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-common/src/main/java/org/apache/hyracks/control/common/deployment/DeploymentUtils.java
@@ -26,6 +26,7 @@
import java.net.URL;
import java.util.ArrayList;
import java.util.List;
+import java.util.concurrent.ConcurrentMap;
import org.apache.commons.io.FileUtils;
import org.apache.http.HttpEntity;
@@ -37,7 +38,6 @@
import org.apache.hyracks.api.deployment.DeploymentId;
import org.apache.hyracks.api.exceptions.HyracksException;
import org.apache.hyracks.api.job.IJobSerializerDeserializer;
-import org.apache.hyracks.api.job.IJobSerializerDeserializerContainer;
import org.apache.hyracks.api.util.JavaSerializationUtils;
import org.apache.hyracks.control.common.context.ServerContext;
import org.apache.logging.log4j.LogManager;
@@ -62,9 +62,10 @@
* @param ctx
* @throws HyracksException
*/
- public static void undeploy(DeploymentId deploymentId, IJobSerializerDeserializerContainer container,
- ServerContext ctx) throws HyracksException {
- container.removeJobSerializerDeserializer(deploymentId);
+ public static void undeploy(DeploymentId deploymentId,
+ ConcurrentMap<DeploymentId, IJobSerializerDeserializer> container, ServerContext ctx)
+ throws HyracksException {
+ container.remove(deploymentId);
String rootDir = ctx.getBaseDir().toString();
String deploymentDir = rootDir.endsWith(File.separator) ? rootDir + DEPLOYMENT + File.separator + deploymentId
: rootDir + File.separator + DEPLOYMENT + File.separator + deploymentId;
@@ -93,13 +94,11 @@
* true is NC/false is CC
* @throws HyracksException
*/
- public static void deploy(DeploymentId deploymentId, List<URL> urls, IJobSerializerDeserializerContainer container,
- ServerContext ctx, boolean isNC) throws HyracksException {
- IJobSerializerDeserializer jobSerDe = container.getJobSerializerDeserializer(deploymentId);
- if (jobSerDe == null) {
- jobSerDe = new ClassLoaderJobSerializerDeserializer();
- container.addJobSerializerDeserializer(deploymentId, jobSerDe);
- }
+ public static void deploy(DeploymentId deploymentId, List<URL> urls,
+ ConcurrentMap<DeploymentId, IJobSerializerDeserializer> container, ServerContext ctx, boolean isNC)
+ throws HyracksException {
+ IJobSerializerDeserializer jobSerDe =
+ container.computeIfAbsent(deploymentId, d -> new ClassLoaderJobSerializerDeserializer());
String rootDir = ctx.getBaseDir().toString();
String deploymentDir = rootDir.endsWith(File.separator) ? rootDir + DEPLOYMENT + File.separator + deploymentId
: rootDir + File.separator + DEPLOYMENT + File.separator + deploymentId;
@@ -120,9 +119,8 @@
public static Object deserialize(byte[] bytes, DeploymentId deploymentId, IServiceContext serviceCtx)
throws HyracksException {
try {
- IJobSerializerDeserializerContainer jobSerDeContainer = serviceCtx.getJobSerializerDeserializerContainer();
IJobSerializerDeserializer jobSerDe =
- deploymentId == null ? null : jobSerDeContainer.getJobSerializerDeserializer(deploymentId);
+ deploymentId == null ? null : serviceCtx.getJobSerializerDeserializerContainer().get(deploymentId);
return jobSerDe == null ? JavaSerializationUtils.deserialize(bytes) : jobSerDe.deserialize(bytes);
} catch (Exception e) {
throw HyracksException.create(e);
@@ -141,9 +139,8 @@
public static Class<?> loadClass(String className, DeploymentId deploymentId, IServiceContext serviceCtx)
throws HyracksException {
try {
- IJobSerializerDeserializerContainer jobSerDeContainer = serviceCtx.getJobSerializerDeserializerContainer();
IJobSerializerDeserializer jobSerDe =
- deploymentId == null ? null : jobSerDeContainer.getJobSerializerDeserializer(deploymentId);
+ deploymentId == null ? null : serviceCtx.getJobSerializerDeserializerContainer().get(deploymentId);
return jobSerDe == null ? JavaSerializationUtils.loadClass(className) : jobSerDe.loadClass(className);
} catch (ClassNotFoundException | IOException e) {
throw HyracksException.create(e);
@@ -160,9 +157,8 @@
*/
public static ClassLoader getClassLoader(DeploymentId deploymentId, IServiceContext appCtx)
throws HyracksException {
- IJobSerializerDeserializerContainer jobSerDeContainer = appCtx.getJobSerializerDeserializerContainer();
IJobSerializerDeserializer jobSerDe =
- deploymentId == null ? null : jobSerDeContainer.getJobSerializerDeserializer(deploymentId);
+ deploymentId == null ? null : appCtx.getJobSerializerDeserializerContainer().get(deploymentId);
return jobSerDe == null ? DeploymentUtils.class.getClassLoader() : jobSerDe.getClassLoader();
}
diff --git a/hyracks-fullstack/hyracks/hyracks-test-support/src/main/java/org/apache/hyracks/test/support/TestNCServiceContext.java b/hyracks-fullstack/hyracks/hyracks-test-support/src/main/java/org/apache/hyracks/test/support/TestNCServiceContext.java
index 63126ab..7b0abc2 100644
--- a/hyracks-fullstack/hyracks/hyracks-test-support/src/main/java/org/apache/hyracks/test/support/TestNCServiceContext.java
+++ b/hyracks-fullstack/hyracks/hyracks-test-support/src/main/java/org/apache/hyracks/test/support/TestNCServiceContext.java
@@ -19,6 +19,7 @@
package org.apache.hyracks.test.support;
import java.io.Serializable;
+import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ThreadFactory;
import org.apache.hyracks.api.application.INCServiceContext;
@@ -26,8 +27,9 @@
import org.apache.hyracks.api.application.IStateDumpHandler;
import org.apache.hyracks.api.comm.IChannelInterfaceFactory;
import org.apache.hyracks.api.config.IApplicationConfig;
+import org.apache.hyracks.api.deployment.DeploymentId;
import org.apache.hyracks.api.io.IIOManager;
-import org.apache.hyracks.api.job.IJobSerializerDeserializerContainer;
+import org.apache.hyracks.api.job.IJobSerializerDeserializer;
import org.apache.hyracks.api.lifecycle.ILifeCycleComponentManager;
import org.apache.hyracks.api.lifecycle.LifeCycleComponentManager;
import org.apache.hyracks.api.messages.IMessageBroker;
@@ -102,7 +104,7 @@
}
@Override
- public IJobSerializerDeserializerContainer getJobSerializerDeserializerContainer() {
+ public ConcurrentMap<DeploymentId, IJobSerializerDeserializer> getJobSerializerDeserializerContainer() {
return null;
}