Merge branch 'master' of github.com:metamx/druid into new-guava

This commit is contained in:
fjy 2014-06-13 11:30:07 -07:00
commit 4cc0353be2
3 changed files with 37 additions and 21 deletions

View File

@ -132,17 +132,17 @@ public class DruidDefaultSerializersModule extends SimpleModule
public void serialize(Yielder yielder, final JsonGenerator jgen, SerializerProvider provider)
throws IOException, JsonProcessingException
{
jgen.writeStartArray();
try {
jgen.writeStartArray();
while (!yielder.isDone()) {
final Object o = yielder.get();
jgen.writeObject(o);
yielder = yielder.next(null);
}
jgen.writeEndArray();
} finally {
yielder.close();
}
jgen.writeEndArray();
}
}
);

View File

@ -243,7 +243,9 @@ public class ChainedExecutionQueryRunnerTest
Assert.assertTrue(future.isCancelled());
Assert.assertTrue(runner1.hasStarted);
Assert.assertTrue(runner2.hasStarted);
Assert.assertFalse(runner3.hasStarted);
Assert.assertTrue(runner1.interrupted);
Assert.assertTrue(runner2.interrupted);
Assert.assertTrue(!runner3.hasStarted || runner3.interrupted);
Assert.assertFalse(runner1.hasCompleted);
Assert.assertFalse(runner2.hasCompleted);
Assert.assertFalse(runner3.hasCompleted);
@ -256,6 +258,7 @@ public class ChainedExecutionQueryRunnerTest
private final CountDownLatch latch;
private boolean hasStarted = false;
private boolean hasCompleted = false;
private boolean interrupted = false;
public DyingQueryRunner(CountDownLatch latch)
{
@ -268,6 +271,7 @@ public class ChainedExecutionQueryRunnerTest
hasStarted = true;
latch.countDown();
if (Thread.interrupted()) {
interrupted = true;
throw new QueryInterruptedException("I got killed");
}
@ -276,6 +280,7 @@ public class ChainedExecutionQueryRunnerTest
Thread.sleep(500);
}
catch (InterruptedException e) {
interrupted = true;
throw new QueryInterruptedException("I got killed");
}

View File

@ -139,28 +139,29 @@ public class QueryResource
log.debug("Got query [%s]", query);
}
Sequence results = query.run(texasRanger);
if (results == null) {
Sequence res = query.run(texasRanger);
final Sequence results;
if (res == null) {
results = Sequences.empty();
} else {
results = res;
}
try (
final Yielder yielder = results.toYielder(
null,
new YieldingAccumulator()
{
@Override
public Object accumulate(Object accumulated, Object in)
{
yield();
return in;
}
}
)
) {
long requestTime = System.currentTimeMillis() - start;
final Yielder yielder = results.toYielder(
null,
new YieldingAccumulator()
{
@Override
public Object accumulate(Object accumulated, Object in)
{
yield();
return in;
}
}
);
try {
long requestTime = System.currentTimeMillis() - start;
emitter.emit(
new ServiceMetricEvent.Builder()
.setUser2(DataSourceUtil.getMetricName(query.getDataSource()))
@ -194,6 +195,7 @@ public class QueryResource
@Override
public void write(OutputStream outputStream) throws IOException, WebApplicationException
{
// json serializer will always close the yielder
jsonWriter.writeValue(outputStream, yielder);
outputStream.close();
}
@ -203,6 +205,15 @@ public class QueryResource
.header("X-Druid-Query-Id", queryId)
.build();
}
catch (Exception e) {
// make sure to close yieder if anything happened before starting to serialize the response.
yielder.close();
throw Throwables.propagate(e);
}
finally {
// do not close yielder here, since we do not want to close the yielder prior to
// StreamingOutput having iterated over all the results
}
}
catch (QueryInterruptedException e) {
try {