Skip to content
Merged
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -2,9 +2,11 @@

import static org.hypertrace.core.query.service.RowChunkingOperator.chunkRows;

import com.google.common.collect.ImmutableMap;
import io.grpc.Status;
import io.grpc.stub.ServerCallStreamObserver;
import io.grpc.stub.StreamObserver;
import io.micrometer.core.instrument.Counter;
import io.reactivex.rxjava3.core.Maybe;
import io.reactivex.rxjava3.core.Observable;
import javax.inject.Inject;
Expand All @@ -16,14 +18,32 @@
import org.hypertrace.core.query.service.api.QueryServiceGrpc;
import org.hypertrace.core.query.service.api.ResultSetChunk;
import org.hypertrace.core.query.service.validation.QueryValidator;
import org.hypertrace.core.serviceframework.metrics.PlatformMetricsRegistry;

@Singleton
@Slf4j
class QueryServiceImpl extends QueryServiceGrpc.QueryServiceImplBase {

private final RequestHandlerSelector handlerSelector;
private final QueryTransformationPipeline queryTransformationPipeline;
private final QueryValidator queryValidator;

private Counter serviceResponseErrorCounter;

class QueryServiceObserver<T> extends ServerCallStreamRxObserver<T> {

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can you move this to a separate file?

Would it make sense to move org.hypertrace.core.grpcutils.server.rx of repo https://github.com/hypertrace/java-grpc-utils/tree/main/grpc-server-rx-utils? Something like as below

class ServerCallStreamRxObserverWihMetricRecorder<T> extends ServerCallStreamRxObserver<T> ?

cc: @aaron-steinfeld

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If it were me, I probably wouldn't use the subscription to do this - you only get one (or in some cases, none, if you're not at a terminal op). Extending classes, and introducing classes that themselves may be extended, inevitably leads to more coupling. Instead, taking functional programming to heart, I would prefer composition here. That is, using either doOnError + doOnNext, or doOnEach (which can accept its own observer, or individual callbacks).

In short, this is an exercise in SRP - we can use two individual observers to accomplish the same thing in a clearer and more maintainable way.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

hmmmm... Okay, I got your point. Made the change.
Can you check again? Let me know if it's fine.


public QueryServiceObserver(StreamObserver<T> serverCallStreamObserver) {
super(serverCallStreamObserver);
}

@Override
public void onError(Throwable th) {
serviceResponseErrorCounter.increment();
super.onError(th);
}

}

@Inject
public QueryServiceImpl(
RequestHandlerSelector handlerSelector,
Expand All @@ -32,6 +52,15 @@ public QueryServiceImpl(
this.handlerSelector = handlerSelector;
this.queryTransformationPipeline = queryTransformationPipeline;
this.queryValidator = queryValidator;
initMetrics();
}

private static final String COUNTER_NAME = "hypertrace.query-service.response.errors";

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Move these declarations to the top of the class. And move the inner class defn down. Can you check if the formatter does this automatically?

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Moved them.


private void initMetrics() {
serviceResponseErrorCounter =
PlatformMetricsRegistry.registerCounter(
COUNTER_NAME, ImmutableMap.of());
}

@Override
Expand All @@ -47,7 +76,7 @@ public void execute(
originalRequest, requestContext.getTenantId().orElseThrow())))
.doOnError(error -> log.error("Query failed: {}", originalRequest, error))
.subscribe(
new ServerCallStreamRxObserver<>(
new QueryServiceObserver<>(
(ServerCallStreamObserver<ResultSetChunk>) callStreamObserver));
}

Expand Down