igalshilman commented on a change in pull request #110:
URL: https://github.com/apache/flink-statefun/pull/110#discussion_r426629629
##########
File path:
statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/httpfn/HttpFunctionProvider.java
##########
@@ -41,12 +48,56 @@ public RequestReplyFunction functionOfType(FunctionType
type) {
if (spec == null) {
throw new IllegalArgumentException("Unsupported type " + type);
}
+ return new RequestReplyFunction(
+ spec.states(), spec.maxNumBatchRequests(), buildHttpClient(spec));
+ }
+
+ private RequestReplyClient buildHttpClient(HttpFunctionSpec spec) {
+ // We need to build a UDS HTTP client
+ if (spec.isUnixDomainSocket()) {
+
+ // We need to split the path in order to get the sock file and the path
after the sock file
+ Map.Entry<String, String> splittedFilePathAndEndpoint =
+ splitFilePathAndEndpointForUDS(spec.endpoint());
+
+ OkHttpClient specificClient =
+ sharedClient
+ .newBuilder()
+ .socketFactory(
+ new
AFUNIXSocketFactory.FactoryArg(splittedFilePathAndEndpoint.getKey()))
Review comment:
Unfortunately this does not seems to be working.
I get the following stack trace:
It seems like okhttp is trying to set the socket timeout, before connecting
(that makes senes) but the library
expects first a connection and then setting a timeout (which also makes
senes)
```
orker_1 | Caused by: java.net.SocketException: Not open
worker_1 | at
org.newsclub.net.unix.AFUNIXSocketImpl.validFdOrException(AFUNIXSocketImpl.java:424)
worker_1 | at
org.newsclub.net.unix.AFUNIXSocketImpl.setOption(AFUNIXSocketImpl.java:504)
worker_1 | at
org.newsclub.net.unix.AFUNIXSocketImpl$Lenient.setOption(AFUNIXSocketImpl.java:568)
worker_1 | at
java.net.Socket.setSoTimeout(Socket.java:1159)
worker_1 | at
okhttp3.internal.connection.RealConnection.connectSocket(RealConnection.java:261)
worker_1 | at
okhttp3.internal.connection.RealConnection.connect(RealConnection.java:183)
worker_1 | at
okhttp3.internal.connection.ExchangeFinder.findConnection(ExchangeFinder.java:224)
worker_1 | at
okhttp3.internal.connection.ExchangeFinder.findHealthyConnection(ExchangeFinder.java:108)
worker_1 | at
okhttp3.internal.connection.ExchangeFinder.find(ExchangeFinder.java:88)
worker_1 | at
okhttp3.internal.connection.Transmitter.newExchange(Transmitter.java:169)
worker_1 | at
okhttp3.internal.connection.ConnectInterceptor.intercept(ConnectInterceptor.java:41)
worker_1 | at
okhttp3.internal.http.RealInterceptorChain.proceed(RealInterceptorChain.java:142)
worker_1 | at
okhttp3.internal.http.RealInterceptorChain.proceed(RealInterceptorChain.java:117)
worker_1 | at
okhttp3.internal.cache.CacheInterceptor.intercept(CacheInterceptor.java:94)
worker_1 | at
okhttp3.internal.http.RealInterceptorChain.proceed(RealInterceptorChain.java:142)
worker_1 | at
okhttp3.internal.http.RealInterceptorChain.proceed(RealInterceptorChain.java:117)
worker_1 | at
okhttp3.internal.http.BridgeInterceptor.intercept(BridgeInterceptor.java:93)
worker_1 | at
okhttp3.internal.http.RealInterceptorChain.proceed(RealInterceptorChain.java:142)
worker_1 | at
okhttp3.internal.http.RetryAndFollowUpInterceptor.intercept(RetryAndFollowUpInterceptor.java:88)
worker_1 | at
okhttp3.internal.http.RealInterceptorChain.proceed(RealInterceptorChain.java:142)
worker_1 | at
okhttp3.internal.http.RealInterceptorChain.proceed(RealInterceptorChain.java:117)
worker_1 | at
okhttp3.RealCall.getResponseWithInterceptorChain(RealCall.java:229)
worker_1 | at
okhttp3.RealCall$AsyncCall.execute(RealCall.java:172)
```
##########
File path: pom.xml
##########
@@ -170,6 +171,11 @@ under the License.
</configuration>
</execution>
</executions>
+ <configuration>
Review comment:
Can you explain that change?
I am not able to build (on OS X) the project with that.
##########
File path:
statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/httpfn/HttpFunctionProvider.java
##########
@@ -41,12 +48,56 @@ public RequestReplyFunction functionOfType(FunctionType
type) {
if (spec == null) {
throw new IllegalArgumentException("Unsupported type " + type);
}
+ return new RequestReplyFunction(
+ spec.states(), spec.maxNumBatchRequests(), buildHttpClient(spec));
+ }
+
+ private RequestReplyClient buildHttpClient(HttpFunctionSpec spec) {
+ // We need to build a UDS HTTP client
+ if (spec.isUnixDomainSocket()) {
+
+ // We need to split the path in order to get the sock file and the path
after the sock file
+ Map.Entry<String, String> splittedFilePathAndEndpoint =
+ splitFilePathAndEndpointForUDS(spec.endpoint());
+
+ OkHttpClient specificClient =
+ sharedClient
+ .newBuilder()
+ .socketFactory(
+ new
AFUNIXSocketFactory.FactoryArg(splittedFilePathAndEndpoint.getKey()))
+ .callTimeout(spec.maxRequestDuration())
+ .build();
+
+ return new HttpRequestReplyClient(
+ // Only the path matters!
+ HttpUrl.get(URI.create(splittedFilePathAndEndpoint.getValue())),
specificClient);
Review comment:
There are two unfortunate things going on here:
1) `HttpUrl.get` would silently return `null` if it was unable to parse the
URI
2) the endpoint only (rightfully so) contains the path without the schema
and `HttpUrl` doesn't consider it to be valid.
##########
File path:
statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/httpfn/OkHttpUtils.java
##########
@@ -38,6 +41,9 @@ static OkHttpClient newClient() {
.followRedirects(true)
.followSslRedirects(true)
.retryOnConnectionFailure(true)
+ // Enable HTTP/2 if available (uses H2 upgrade),
+ // otherwise fallback to HTTP/1.1
+ .protocols(Collections.singletonList(Protocol.HTTP_2))
Review comment:
It seems like okhttp has HTTP_2 already in the default protocol list,
so this is not needed.
Moreover the above code throws an exception (when specifying HTTP_2 you must
also specify HTTP_1) but if you would look at the default builder you would see
that okhttp already has HTTP_2 and HTTP_1 in the list of default protocols.
As an attempt to "make it work" I've artificially added the prefix
`http://localhost/` to the endpoint, what do you think about that (it is
technically "localhost")?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
