Apache
/
druid
mirror of https://github.com/apache/druid.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

less verbose logging for LQP

This commit is contained in:
fjy 2014-07-29 09:51:53 -07:00
parent 7aef463457
commit 382cde9cf2
1 changed files with 37 additions and 35 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

72
server/src/main/java/io/druid/server/coordinator/LoadQueuePeon.java
View File

@ -87,7 +87,7 @@ public class LoadQueuePeon
private final Object lock = new Object(); private final Object lock = new Object();
private volatile SegmentHolder currentlyLoading = null; private volatile SegmentHolder currentlyProcessing = null;
LoadQueuePeon( LoadQueuePeon(
CuratorFramework curator, CuratorFramework curator,
@ -156,10 +156,10 @@ public class LoadQueuePeon
) )
{ {
synchronized (lock) { synchronized (lock) {
if ((currentlyLoading != null) && if ((currentlyProcessing != null) &&
currentlyLoading.getSegmentIdentifier().equals(segment.getIdentifier())) { currentlyProcessing.getSegmentIdentifier().equals(segment.getIdentifier())) {
if (callback != null) { if (callback != null) {
currentlyLoading.addCallback(callback); currentlyProcessing.addCallback(callback);
} }
return; return;
} }
@ -170,13 +170,13 @@ public class LoadQueuePeon
synchronized (lock) { synchronized (lock) {
if (segmentsToLoad.contains(holder)) { if (segmentsToLoad.contains(holder)) {
if ((callback != null)) { if ((callback != null)) {
currentlyLoading.addCallback(callback); currentlyProcessing.addCallback(callback);
} }
return; return;
} }
} }
log.info("Asking server peon[%s] to load segment[%s]", basePath, segment); log.info("Asking server peon[%s] to load segment[%s]", basePath, segment.getIdentifier());
queuedSize.addAndGet(segment.getSize()); queuedSize.addAndGet(segment.getSize());
segmentsToLoad.add(holder); segmentsToLoad.add(holder);
doNext(); doNext();
@ -188,10 +188,10 @@ public class LoadQueuePeon
) )
{ {
synchronized (lock) { synchronized (lock) {
if ((currentlyLoading != null) && if ((currentlyProcessing != null) &&
currentlyLoading.getSegmentIdentifier().equals(segment.getIdentifier())) { currentlyProcessing.getSegmentIdentifier().equals(segment.getIdentifier())) {
if (callback != null) { if (callback != null) {
currentlyLoading.addCallback(callback); currentlyProcessing.addCallback(callback);
} }
return; return;
} }
@ -202,7 +202,7 @@ public class LoadQueuePeon
synchronized (lock) { synchronized (lock) {
if (segmentsToDrop.contains(holder)) { if (segmentsToDrop.contains(holder)) {
if (callback != null) { if (callback != null) {
currentlyLoading.addCallback(callback); currentlyProcessing.addCallback(callback);
} }
return; return;
} }
@ -216,13 +216,13 @@ public class LoadQueuePeon
private void doNext() private void doNext()
{ {
synchronized (lock) { synchronized (lock) {
if (currentlyLoading == null) { if (currentlyProcessing == null) {
if (!segmentsToDrop.isEmpty()) { if (!segmentsToDrop.isEmpty()) {
currentlyLoading = segmentsToDrop.first(); currentlyProcessing = segmentsToDrop.first();
log.info("Server[%s] dropping [%s]", basePath, currentlyLoading); log.info("Server[%s] dropping [%s]", basePath, currentlyProcessing.getSegmentIdentifier());
} else if (!segmentsToLoad.isEmpty()) { } else if (!segmentsToLoad.isEmpty()) {
currentlyLoading = segmentsToLoad.first(); currentlyProcessing = segmentsToLoad.first();
log.info("Server[%s] loading [%s]", basePath, currentlyLoading); log.info("Server[%s] loading [%s]", basePath, currentlyProcessing.getSegmentIdentifier());
} else { } else {
return; return;
} }
@ -235,16 +235,16 @@ public class LoadQueuePeon
{ {
synchronized (lock) { synchronized (lock) {
try { try {
if (currentlyLoading == null) { if (currentlyProcessing == null) {
log.makeAlert("Crazy race condition! server[%s]", basePath) log.makeAlert("Crazy race condition! server[%s]", basePath)
.emit(); .emit();
actionCompleted(); actionCompleted();
doNext(); doNext();
return; return;
} }
log.info("Server[%s] adding segment[%s]", basePath, currentlyLoading.getSegmentIdentifier()); log.info("Server[%s] processing segment[%s]", basePath, currentlyProcessing.getSegmentIdentifier());
final String path = ZKPaths.makePath(basePath, currentlyLoading.getSegmentIdentifier()); final String path = ZKPaths.makePath(basePath, currentlyProcessing.getSegmentIdentifier());
final byte[] payload = jsonMapper.writeValueAsBytes(currentlyLoading.getChangeRequest()); final byte[] payload = jsonMapper.writeValueAsBytes(currentlyProcessing.getChangeRequest());
curator.create().withMode(CreateMode.EPHEMERAL).forPath(path, payload); curator.create().withMode(CreateMode.EPHEMERAL).forPath(path, payload);
zkWritingExecutor.schedule( zkWritingExecutor.schedule(
@ -255,7 +255,7 @@ public class LoadQueuePeon
{ {
try { try {
if (curator.checkExists().forPath(path) != null) { if (curator.checkExists().forPath(path) != null) {
failAssign(new ISE("%s was never removed! Failing this assign!", path)); failAssign(new ISE("%s was never removed! Failing this operation!", path));
} }
} }
catch (Exception e) { catch (Exception e) {
@ -311,7 +311,9 @@ public class LoadQueuePeon
); );
} else { } else {
log.info( log.info(
"Server[%s] skipping doNext() because something is currently loading[%s].", basePath, currentlyLoading "Server[%s] skipping doNext() because something is currently loading[%s].",
basePath,
currentlyProcessing.getSegmentIdentifier()
); );
} }
} }
@ -319,29 +321,29 @@ public class LoadQueuePeon
private void actionCompleted() private void actionCompleted()
{ {
if (currentlyLoading != null) { if (currentlyProcessing != null) {
switch (currentlyLoading.getType()) { switch (currentlyProcessing.getType()) {
case LOAD: case LOAD:
segmentsToLoad.remove(currentlyLoading); segmentsToLoad.remove(currentlyProcessing);
queuedSize.addAndGet(-currentlyLoading.getSegmentSize()); queuedSize.addAndGet(-currentlyProcessing.getSegmentSize());
break; break;
case DROP: case DROP:
segmentsToDrop.remove(currentlyLoading); segmentsToDrop.remove(currentlyProcessing);
break; break;
default: default:
throw new UnsupportedOperationException(); throw new UnsupportedOperationException();
} }
currentlyLoading.executeCallbacks(); currentlyProcessing.executeCallbacks();
currentlyLoading = null; currentlyProcessing = null;
} }
} }
public void stop() public void stop()
{ {
synchronized (lock) { synchronized (lock) {
if (currentlyLoading != null) { if (currentlyProcessing != null) {
currentlyLoading.executeCallbacks(); currentlyProcessing.executeCallbacks();
currentlyLoading = null; currentlyProcessing = null;
} }
if (!segmentsToDrop.isEmpty()) { if (!segmentsToDrop.isEmpty()) {
@ -366,14 +368,14 @@ public class LoadQueuePeon
private void entryRemoved(String path) private void entryRemoved(String path)
{ {
synchronized (lock) { synchronized (lock) {
if (currentlyLoading == null) { if (currentlyProcessing == null) {
log.warn("Server[%s] an entry[%s] was removed even though it wasn't loading!?", basePath, path); log.warn("Server[%s] an entry[%s] was removed even though it wasn't loading!?", basePath, path);
return; return;
} }
if (!ZKPaths.getNodeFromPath(path).equals(currentlyLoading.getSegmentIdentifier())) { if (!ZKPaths.getNodeFromPath(path).equals(currentlyProcessing.getSegmentIdentifier())) {
log.warn( log.warn(
"Server[%s] entry [%s] was removed even though it's not what is currently loading[%s]", "Server[%s] entry [%s] was removed even though it's not what is currently loading[%s]",
basePath, path, currentlyLoading basePath, path, currentlyProcessing
); );
return; return;
} }
@ -387,7 +389,7 @@ public class LoadQueuePeon
private void failAssign(Exception e) private void failAssign(Exception e)
{ {
synchronized (lock) { synchronized (lock) {
log.error(e, "Server[%s], throwable caught when submitting [%s].", basePath, currentlyLoading); log.error(e, "Server[%s], throwable caught when submitting [%s].", basePath, currentlyProcessing);
failedAssignCount.getAndIncrement(); failedAssignCount.getAndIncrement();
// Act like it was completed so that the coordinator gives it to someone else // Act like it was completed so that the coordinator gives it to someone else
actionCompleted(); actionCompleted();