MAPREDUCE-2947. Fixed race condition in AuxiliaryServices. Contributed by Vinod K V.
git-svn-id: https://svn.apache.org/repos/asf/hadoop/common/trunk@1166849 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
parent
70f9ace412
commit
6b6ef7b8e3
@ -1230,6 +1230,9 @@ Release 0.23.0 - Unreleased
|
||||
MAPREDUCE-2942. TestNMAuditLogger.testNMAuditLoggerWithIP failing (Thomas Graves
|
||||
via mahadev)
|
||||
|
||||
MAPREDUCE-2947. Fixed race condition in AuxiliaryServices. (vinodkv via
|
||||
acmurthy)
|
||||
|
||||
Release 0.22.0 - Unreleased
|
||||
|
||||
INCOMPATIBLE CHANGES
|
||||
|
@ -48,20 +48,20 @@ public StartContainerResponsePBImpl(StartContainerResponseProto proto) {
|
||||
viaProto = true;
|
||||
}
|
||||
|
||||
public StartContainerResponseProto getProto() {
|
||||
public synchronized StartContainerResponseProto getProto() {
|
||||
mergeLocalToProto();
|
||||
proto = viaProto ? proto : builder.build();
|
||||
viaProto = true;
|
||||
return proto;
|
||||
}
|
||||
|
||||
private void mergeLocalToBuilder() {
|
||||
private synchronized void mergeLocalToBuilder() {
|
||||
if (this.serviceResponse != null) {
|
||||
addServiceResponseToProto();
|
||||
}
|
||||
}
|
||||
|
||||
private void mergeLocalToProto() {
|
||||
private synchronized void mergeLocalToProto() {
|
||||
if (viaProto) {
|
||||
maybeInitBuilder();
|
||||
}
|
||||
@ -70,7 +70,7 @@ private void mergeLocalToProto() {
|
||||
viaProto = true;
|
||||
}
|
||||
|
||||
private void maybeInitBuilder() {
|
||||
private synchronized void maybeInitBuilder() {
|
||||
if (viaProto || builder == null) {
|
||||
builder = StartContainerResponseProto.newBuilder(proto);
|
||||
}
|
||||
@ -79,17 +79,17 @@ private void maybeInitBuilder() {
|
||||
|
||||
|
||||
@Override
|
||||
public Map<String, ByteBuffer> getAllServiceResponse() {
|
||||
public synchronized Map<String, ByteBuffer> getAllServiceResponse() {
|
||||
initServiceResponse();
|
||||
return this.serviceResponse;
|
||||
}
|
||||
@Override
|
||||
public ByteBuffer getServiceResponse(String key) {
|
||||
public synchronized ByteBuffer getServiceResponse(String key) {
|
||||
initServiceResponse();
|
||||
return this.serviceResponse.get(key);
|
||||
}
|
||||
|
||||
private void initServiceResponse() {
|
||||
private synchronized void initServiceResponse() {
|
||||
if (this.serviceResponse != null) {
|
||||
return;
|
||||
}
|
||||
@ -103,14 +103,14 @@ private void initServiceResponse() {
|
||||
}
|
||||
|
||||
@Override
|
||||
public void addAllServiceResponse(final Map<String, ByteBuffer> serviceResponse) {
|
||||
public synchronized void addAllServiceResponse(final Map<String, ByteBuffer> serviceResponse) {
|
||||
if (serviceResponse == null)
|
||||
return;
|
||||
initServiceResponse();
|
||||
this.serviceResponse.putAll(serviceResponse);
|
||||
}
|
||||
|
||||
private void addServiceResponseToProto() {
|
||||
private synchronized void addServiceResponseToProto() {
|
||||
maybeInitBuilder();
|
||||
builder.clearServiceResponse();
|
||||
if (serviceResponse == null)
|
||||
@ -118,24 +118,24 @@ private void addServiceResponseToProto() {
|
||||
Iterable<StringBytesMapProto> iterable = new Iterable<StringBytesMapProto>() {
|
||||
|
||||
@Override
|
||||
public Iterator<StringBytesMapProto> iterator() {
|
||||
public synchronized Iterator<StringBytesMapProto> iterator() {
|
||||
return new Iterator<StringBytesMapProto>() {
|
||||
|
||||
Iterator<String> keyIter = serviceResponse.keySet().iterator();
|
||||
|
||||
@Override
|
||||
public void remove() {
|
||||
public synchronized void remove() {
|
||||
throw new UnsupportedOperationException();
|
||||
}
|
||||
|
||||
@Override
|
||||
public StringBytesMapProto next() {
|
||||
public synchronized StringBytesMapProto next() {
|
||||
String key = keyIter.next();
|
||||
return StringBytesMapProto.newBuilder().setKey(key).setValue(convertToProtoFormat(serviceResponse.get(key))).build();
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean hasNext() {
|
||||
public synchronized boolean hasNext() {
|
||||
return keyIter.hasNext();
|
||||
}
|
||||
};
|
||||
@ -144,17 +144,17 @@ public boolean hasNext() {
|
||||
builder.addAllServiceResponse(iterable);
|
||||
}
|
||||
@Override
|
||||
public void setServiceResponse(String key, ByteBuffer val) {
|
||||
public synchronized void setServiceResponse(String key, ByteBuffer val) {
|
||||
initServiceResponse();
|
||||
this.serviceResponse.put(key, val);
|
||||
}
|
||||
@Override
|
||||
public void removeServiceResponse(String key) {
|
||||
public synchronized void removeServiceResponse(String key) {
|
||||
initServiceResponse();
|
||||
this.serviceResponse.remove(key);
|
||||
}
|
||||
@Override
|
||||
public void clearServiceResponse() {
|
||||
public synchronized void clearServiceResponse() {
|
||||
initServiceResponse();
|
||||
this.serviceResponse.clear();
|
||||
}
|
||||
|
@ -23,6 +23,7 @@
|
||||
import java.util.Collections;
|
||||
import java.util.HashMap;
|
||||
import java.util.Map;
|
||||
import java.util.Map.Entry;
|
||||
|
||||
import org.apache.commons.logging.Log;
|
||||
import org.apache.commons.logging.LogFactory;
|
||||
@ -72,7 +73,14 @@ Collection<AuxiliaryService> getServices() {
|
||||
* the the name of the service as defined in the configuration.
|
||||
*/
|
||||
public Map<String, ByteBuffer> getMeta() {
|
||||
return Collections.unmodifiableMap(serviceMeta);
|
||||
Map<String, ByteBuffer> metaClone = new HashMap<String, ByteBuffer>(
|
||||
serviceMeta.size());
|
||||
synchronized (serviceMeta) {
|
||||
for (Entry<String, ByteBuffer> entry : serviceMeta.entrySet()) {
|
||||
metaClone.put(entry.getKey(), entry.getValue().duplicate());
|
||||
}
|
||||
}
|
||||
return metaClone;
|
||||
}
|
||||
|
||||
@Override
|
||||
|
Loading…
Reference in New Issue
Block a user