zalando-nakadi / fahrschein Goto Github PK

View Code? Open in Web Editor NEW

24.0 24.0 38.0 1.55 MB

A java client for the nakadi event bus

License: Apache License 2.0

Java 99.32% Shell 0.68%

fahrschein's People

Contributors

Stargazers

Watchers

fahrschein's Issues

Unable to connect to expired/empty topics

It looks like 0.4.1 introduced a regression concerning expired and empty topics (low-level API).

Scenario

On...

http GET https://nakadi.com/event-types/my-empty-topic/partitions
[
    {
        "newest_available_offset": "BEGIN",
        "oldest_available_offset": "0",
        "partition": "0"
    }
]

... with...

cursorManager.updatePartitions(eventName, client.getPartitions(eventName));
client.listen(...);

... leads to...

org.zalando.fahrschein.IOProblem: Problem [http://httpstatus.es/412] with status [412]: [Precondition Failed] [partition 0 is empty]

But Why

Might be in CursorManager#updatePartitions - it is always using the oldest_available_offset if no Cursor is available from the manager. Therefore maybe something like this:

onSuccess(eventName, new Cursor(partition.getPartition(), fallbackToBegin ? "BEGIN" :  partition.getNewestOffset()));

With fallbackToBegin being added to the API.

Potential Connection Leak on Problem

If a problem is detected and thrown there is a potential connection leak happening here. The response is not closed.

Support posting events to nakadi

Or should this library concentrate only on the consuming part?

Should fail early on ClassCastException

java.lang.ClassCastException: UnknownEvent cannot be cast to SomeEvent
    at java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:193)
    at java.util.ArrayList$ArrayListSpliterator.forEachRemaining(ArrayList.java:1374)
    at java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:481)
    ... 4 more
    at java.util.stream.ReferencePipeline.forEachOrdered(ReferencePipeline.java:423)
    at Listener.accept(Listener.java:30)
    at org.zalando.fahrschein.NakadiReader.processBatch(NakadiReader.java:146)
    ... 3 more
    at org.zalando.fahrschein.NakadiClient.listen(NakadiClient.java:83)

With a type hierarchy like this:

@JsonTypeInfo(..., defaultImpl = UnknownEvent.class)
@JsonSubTypes({
    @JsonSubTypes.Type(SomeEvent.class),
})
public interface Event {
    URI getType();
}

Slow Reconnect with HttpComponents

If a client uses a HttpComponentsClientHttpRequestFactory re-connecting on processing failures (i.e. IOException) is very slow.

fahrschein tries to close the ClientHttpResponse in case of a failure. A HttpComponentsClientHttpResponse will always try to release the connection back to the connection manager by consuming the full response body until an EOF. This can take an unpredictable amount of time, effectively pausing fahrschein.

See here for an example failing test.

ManagedCursorManager might not be threadsafe

Subscription and stream ids are stored in a plain HashMap.

Write first test

Commit endpoint in subscription api requires cursors wrapped in object

IllegalArgumentException: Illegal character(s) in message header value:

Illegal characters in header “X-Nakadi-Cursors” 's value “[ {\n “partition”:”0”,\n “offset”:”BEGIN”\n}]”

Caused by: java.lang.IllegalArgumentException: Illegal character(s) in message header value: [ {
  "partition" : "0",
  "offset" : "BEGIN"
} ]
    at sun.net.www.protocol.http.HttpURLConnection.checkMessageHeader(HttpURLConnection.java:508)
    at sun.net.www.protocol.http.HttpURLConnection.isExternalMessageHeaderAllowed(HttpURLConnection.java:459)
    at sun.net.www.protocol.http.HttpURLConnection.addRequestProperty(HttpURLConnection.java:3048)
    at sun.net.www.protocol.https.HttpsURLConnectionImpl.addRequestProperty(HttpsURLConnectionImpl.java:331)
    at org.springframework.http.client.SimpleBufferingClientHttpRequest.addHeaders(SimpleBufferingClientHttpRequest.java:104)
    at org.springframework.http.client.SimpleBufferingClientHttpRequest.executeInternal(SimpleBufferingClientHttpRequest.java:70)
    at org.springframework.http.client.AbstractBufferingClientHttpRequest.executeInternal(AbstractBufferingClientHttpRequest.java:48)
    at org.springframework.http.client.AbstractClientHttpRequest.execute(AbstractClientHttpRequest.java:53)
    at org.zalando.fahrschein.ProblemHandlingClientHttpRequest.execute(ProblemHandlingClientHttpRequest.java:38)
    at org.zalando.fahrschein.NakadiReader.openJsonInput(NakadiReader.java:129)
    at org.zalando.fahrschein.NakadiReader.runInternal(NakadiReader.java:237)
    at org.zalando.fahrschein.NakadiReader.run(NakadiReader.java:224)
    at org.zalando.fahrschein.NakadiClient.listen(NakadiClient.java:110)
    at org.zalando.fahrschein.NakadiClient.listen(NakadiClient.java:101)
    at de.zalando.communication.blacklistservice.nakadiconsumer.NakadiListener.listenEventAndCreateBlacklist(NakadiListener.java:129)
    at de.zalando.communication.blacklistservice.BlacklistServiceApplication.main(BlacklistServiceApplication.java:18)

InMemoryCursor throws NullpointerException when no events in present

2016-06-30 17:02:35,325 {INFO} [NakadiEventListener-1] [] [de.zalando.fulfillment.domain.logic.EventReceiverGateway] Listening for [fulfillment-order-service-test.customer-shipment-shipped]...
2016-06-30 17:02:35,325 {INFO} [NakadiEventListener-2] [] [de.zalando.fulfillment.domain.logic.EventReceiverGateway] Listening for [fulfillment-order-service-test.customer-shipment-item-cancelled]...
2016-06-30 17:02:35,537 {INFO} [main] [] [de.zalando.fulfillment.event.EventReceiverGatewayTest] Started EventReceiverGatewayTest in 3.259 seconds (JVM running for 5.265)2016-06-30 17:02:36,138 {DEBUG} [NakadiEventListener-2] [] [de.zalando.fulfillment.domain.logic.EventReceiverGateway] Partition [0] has oldest offset [10] and newest offset [9]
2016-06-30 17:02:36,138 {DEBUG} [NakadiEventListener-1] [] [de.zalando.fulfillment.domain.logic.EventReceiverGateway] Partition [0] has oldest offset [88] and newest offset [88]
2016-06-30 17:02:36,174 {ERROR} [NakadiEventListener-1] [] [de.zalando.fulfillment.domain.logic.EventReceiverGateway] Woops
java.lang.NullPointerException
    at org.zalando.fahrschein.InMemoryCursorManager.getCursors(InMemoryCursorManager.java:30) ~[fahrschein-0.1.0.jar:?]
    at org.zalando.fahrschein.NakadiReader.openStream(NakadiReader.java:71) ~[fahrschein-0.1.0.jar:?]
    at org.zalando.fahrschein.NakadiReader.run(NakadiReader.java:145) ~[fahrschein-0.1.0.jar:?]
    at org.zalando.fahrschein.NakadiClient.listen(NakadiClient.java:96) ~[fahrschein-0.1.0.jar:?]
    at de.zalando.fulfillment.domain.logic.EventReceiverGateway.listenForEvent(EventReceiverGateway.java:82) [main/:?]
    at de.zalando.fulfillment.domain.logic.EventReceiverGateway$$FastClassBySpringCGLIB$$c4d32a7.invoke(<generated>) [main/:?]
    at org.springframework.cglib.proxy.MethodProxy.invoke(MethodProxy.java:204) [spring-core-4.2.6.RELEASE.jar:4.2.6.RELEASE]
    at org.springframework.aop.framework.CglibAopProxy$CglibMethodInvocation.invokeJoinpoint(CglibAopProxy.java:720) [spring-aop-4.2.6.RELEASE.jar:4.2.6.RELEASE]
    at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:157) [spring-aop-4.2.6.RELEASE.jar:4.2.6.RELEASE]
    at org.springframework.aop.interceptor.AsyncExecutionInterceptor$1.call(AsyncExecutionInterceptor.java:115) [spring-aop-4.2.6.RELEASE.jar:4.2.6.RELEASE]
    at java.util.concurrent.FutureTask.run(FutureTask.java:266) [?:1.8.0_25]
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) [?:1.8.0_25]
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) [?:1.8.0_25]
    at java.lang.Thread.run(Thread.java:745) [?:1.8.0_25]

Handle http status codes and problem responses

Remove requirement for `setSerializationInclusion(JsonInclude.Include.NON_NULL)`

expectToken should throw IOException instead of IllegalStateException

The behaviour should be similar to a JsonParseException, triggering retry and backoff

Add comparison with nakadi-klients to readme

See https://github.com/zalando/nakadi-klients

Better name for `CursorManager.updateCursors`

Fahrschein does not commit empty batches to high-level API

When using the ManagedCursorManager, commiting can fail with

[org.zalando.fahrschein.ManagedCursorManager] Unexpected status code [422] for subscription [ce41ac5c-1c1d-42be-a5db-32f1f976a91b] to event [event] in partition [0] with offset [90458]

Nakadi rquires that

The client must commit at least once every 60 seconds, otherwise Nakadi will consider the client to be gone and will close the connection.

The error above appears if Nakadi closed the stream without the fahrschein-client noticing.

This can be mitigated by commiting even empty batches to the Nakadi high-level API.

Getting exception while trying to enable compression to Nakadi

I'm trying to enable compression while connecting to nakadi,
I'm adding compression as the following

requestFactory = new EnableCompressingRequestFactory(requestFactory); 

class EnableCompressingRequestFactory implements ClientHttpRequestFactory {
    private final ClientHttpRequestFactory delegate;

    public EnableCompressingRequestFactory(final ClientHttpRequestFactory delegate) {
        this.delegate = delegate;
    }

    @Override
    public ClientHttpRequest createRequest(final URI uri, final HttpMethod httpMethod) throws IOException {
        final ClientHttpRequest request = delegate.createRequest(uri, httpMethod);
        request.getHeaders().add("Accept-Encoding", "gzip");
        return request;
    }
}

Unfortunately getting the following errors


com.fasterxml.jackson.core.JsonParseException: Illegal character ((CTRL-CHAR, code 31)): only regular white space (\r, \n, \t) is allowed between tokens
 at [Source: sun.net.www.protocol.http.HttpURLConnection$HttpInputStream@29daf503; line: 1, column: 2]
    at com.fasterxml.jackson.core.JsonParser._constructError(JsonParser.java:1581)
    at com.fasterxml.jackson.core.base.ParserMinimalBase._reportError(ParserMinimalBase.java:533)
    at com.fasterxml.jackson.core.base.ParserMinimalBase._throwInvalidSpace(ParserMinimalBase.java:484)
    at com.fasterxml.jackson.core.json.UTF8StreamJsonParser._skipWSOrEnd(UTF8StreamJsonParser.java:2875)
    at com.fasterxml.jackson.core.json.UTF8StreamJsonParser.nextToken(UTF8StreamJsonParser.java:692)
    at org.zalando.fahrschein.NakadiReader.expectToken(NakadiReader.java:318)
    at org.zalando.fahrschein.NakadiReader.runInternal(NakadiReader.java:247)
    at org.zalando.fahrschein.NakadiReader.run(NakadiReader.java:224)
    at org.zalando.fahrschein.NakadiClient.listen(NakadiClient.java:110)

Do you have any idea how can i solve this issue?

IOExceptions during reading of events are wrapped in RuntimeExceptions

Happens for example with an SSLHandshakeException. The MappingIterator wraps it into RuntimeException but we only unwrap a very specific RuntimeJsonMappingException.

java.lang.RuntimeException: SSL peer shut down incorrectly  
at com.fasterxml.jackson.databind.MappingIterator.next(MappingIterator.java:196)  
at org.zalando.fahrschein.NakadiReader.readEvent(NakadiReader.java:205)

Document and validate stream parameters

Copy description from nakadi api docs as javadoc
Validate relation between different timeout parameters

Support schema for PersistentCursorManager

org.zalando.fahrschein.PersistentCursorManager executes all sql statements against the global schema. It would be nice to tell the PersistentCursorManager against which schema the sql statements should be executed.

Remove guava dependency

It looks like the only thing we use from guava are Preconditions, Joiner and Splitter.

Handle oauth error responses and wrap them in problems

Some error responses from nakadi are not send as application/problem+json but as application/json in the following format:

{"error":"invalid_request","error_description":"Access Token not valid"}

This is done by spring-security in OAuth2ExceptionJackson2Serializer.

Disable Jackson encoding autodetection

Jackson tries to autodetect the used encoding when the JsonParser is created. To do this it needs to read the first 2-4 bytes of the input stream. This can block and even fail with read timeouts depending on the stream parameters.

The main issue is that this causes confusing log output, normally users would assume that blocking only occurs on trying to read the first token.

Closing SimpleHttpClientResponse tries to drain response in spring 4.3.x

Fahrschein tries to close the ClientHttpResponse in case of any exception in order to retry consuming the stream from a clean state. With current spring versions, this blocks trying to consume remaining data, until the nakadi server decides to close the connection.

Related to issue #17 for HttpComponentsClientHttpResponse, the same behaviour was added to SimpleHttpClientResponse in spring-projects/spring-framework@b947bfe

Why wrapping exception?

Callable.call throws Exception, why the wrapping here?

    default Callable<T> unchecked() {
        return () -> {
            try {
                return call();
            } catch (IOException e) {
                throw new UncheckedIOException(e);
            }
        };
    }

Measure and report code coverage using jacoco

Probably without coverage enforcement in the beginning

Split into modules

core
jdbc
redis
metrics-dropwizard

ProblemHandlingClientHttpRequest does not detect content type correctly

PROBLEM_CONTENT_TYPES.contains(contentType.getType())

But getType is only returning the primary type which would be application.

Rename `markMessageSuccessfullyProcessed`?

@tmuehl in https://github.com/zalando-incubator/fahrschein/blob/0.4.4/src/main/java/org/zalando/fahrschein/NakadiReader.java#L270 markMessageSuccessfullyProcessed is only called if there were events contained in the batch. Would it make sense to rename if to markEventsProcessed, maybe event with the event count as parameter?

Add generic `data` property to `DataChangeEvent` interface

As a reminder for users that data change events are nested differently.

Support api changes in high level api

New format is currently being discussed in zalando/nakadi#403

The only place where we currently rely on the format is in CursorManager.updatePartitions (https://github.com/zalando-incubator/fahrschein/blob/master/src/main/java/org/zalando/fahrschein/CursorManager.java#L52)

It should be possible to handle both formats at the same time, new format should be compared lexicographically, the old format would have to be padded so both offsets have the same length before comparing.

Support planned metadata on first and last event batch

See zalando/nakadi#406

Unknown properties in the batch should be ignored, or at least the proposed metadata property needs to be recognized.

Support passing a custom error handler to NakadiClient / NakadiReader

As a user a NakadiClient
I want to pass a own implementation of a error handler
So that I can decide how to handle different kind of exceptions.

In the current implementation NakadiClient catches all IOExceptions and does retries.
This behaviour is somehow unfortunate, when reading an event that can't be deserialized by the provided ObjectMapper. Example:

2016-06-29 13:56:58,604 {WARN} [NakadiEventListener-1] [] [org.zalando.fahrschein.NakadiReader] Got [JsonMappingException] while reading events
com.fasterxml.jackson.databind.JsonMappingException: Failed to deserialize java.time.OffsetDateTime: (java.time.format.DateTimeParseException) Text '2016-06-28T14:41:59' could not be parsed at index 19

NakadiClient will then end up in an (in)finit reconnect-retry-loop.

CursorMananager should have Transactional annotation

Another issue is that the default methods like updateCursors also need to go through a transactional proxy.

Release First Version

As a early adopter
I want to have an early release candidate released
So that I can use this library in production secretly.

Misleading log message after reconnect

On fahrschein 0.4.0 the following logs can be observed:

{WARN} [nakadi-1] [] [org.zalando.fahrschein.ExponentialBackoffStrategy] Got [NoConnectivityOnTheTrainException] on retry [5]
{INFO} [nakadi-1] [] [org.zalando.fahrschein.ExponentialBackoffStrategy] Retry [6], sleeping for [5695] milliseconds
{INFO} [nakadi-2] [] [org.zalando.fahrschein.NakadiReader] Reconnected after [0] errors

I'd expect Reconnected after [5] errors as the last log line.

Local stream timeout does not trigger reconnection

Currently StreamParameters#getStreamTimout is used for a local timeout in NakadiReader (while-loop). Does this make sense?

Also consider adding a log statement if while loop exists.

Implement persistent CursorManager

Based on etcd, redis or a relational db. Should also allow coordination between multiple nodes reading from different partitions.

Event batches should be processed in a threadpool

Simplify building of ClientHttpRequestFactory and NakadiClient itself

Clients have to provide a requestFactory to the NakadiClient, which can handle authorization and problem-responses.

final SimpleClientHttpRequestFactory requestFactoryDelegate = new SimpleClientHttpRequestFactory();
final ProblemHandlingClientHttpRequestFactory problemHandlingRequestFactory = new ProblemHandlingClientHttpRequestFactory(requestFactoryDelegate, objectMapper);
final ClientHttpRequestFactory requestFactory = new AuthorizedClientHttpRequestFactory(problemHandlingRequestFactory, () -> "MY_ACCESS_TOKEN");

As both is mandatory when communicating with nakadi it would be great if fahrschein could offer an easy way to wrap a client's requestFactory.

Add comparison with nakadi-java and kanadi to readme

See https://github.com/zalando-incubator/nakadi-java

Creation of JsonParser blocks at determining the encoding

The creation of the JsonParser in NakadiReader:145 blocks if the input stream has no char in Jackson2 while trying to determine the encoding.

A solution to circumvent this problem is to determine the charset in the NakadiReader itself and then hand over an InputStreamReader to the createParser method

Charset charset = StandardCharsets.UTF_8;
MediaType contentType = response.getHeaders().getContentType();
if (contentType != null) {
    Charset contentCharset = contentType.getCharSet();
    if (contentCharset != null) {
        charset = contentCharset;
    }
}
final JsonParser jsonParser = jsonFactory.createParser(new InputStreamReader(response.getBody(), charset)).disable(JsonParser.Feature.AUTO_CLOSE_SOURCE);

Exception in JdbcCursorManager is not retried

If there is a connection error to the database, the JdbcCursorManager throws an exception which is not caught nor handled by the retry mechanism of fahrschein.

2016-11-30 17:02:18,952 {WARN} [nakadi-2] [] [] [org.zalando.fahrschein.jdbc.JdbcCursorManager] Exception while processing events for [best-event] on part$
tion [0] at offset [39797]
org.springframework.transaction.TransactionSystemException: Could not commit JDBC transaction; nested exception is org.postgresql.util.PSQLException: An I/O error occurred while sending to the backend.
        at org.springframework.jdbc.datasource.DataSourceTransactionManager.doCommit(DataSourceTransactionManager.java:275) ~[spring-jdbc-4.3.3.RELEASE.jar:4.3.3.RELEASE]
        at org.springframework.transaction.support.AbstractPlatformTransactionManager.processCommit(AbstractPlatformTransactionManager.java:761) ~[spring-tx-4.3.3.RELEASE.jar:4.3.3.RELEASE]
        at org.springframework.transaction.support.AbstractPlatformTransactionManager.commit(AbstractPlatformTransactionManager.java:730) ~[spring-tx-4.3.3.RELEASE.jar:4.3.3.RELEASE]
        at org.springframework.transaction.interceptor.TransactionAspectSupport.commitTransactionAfterReturning(TransactionAspectSupport.java:487) ~[spring-tx-4.3.3.RELEASE.jar:4.3.3.RELEASE]
        at org.springframework.transaction.interceptor.TransactionAspectSupport.invokeWithinTransaction(TransactionAspectSupport.java:291) ~[spring-tx-4.3.3.RELEASE.jar:4.3.3.RELEASE]
        at org.springframework.transaction.interceptor.TransactionInterceptor.invoke(TransactionInterceptor.java:96) ~[spring-tx-4.3.3.RELEASE.jar:4.3.3.RELEASE]
        at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:179) ~[spring-aop-4.3.3.RELEASE.jar:4.3.3.RELEASE]
        at org.springframework.aop.framework.JdkDynamicAopProxy.invoke(JdkDynamicAopProxy.java:213) [spring-aop-4.3.3.RELEASE.jar:4.3.3.RELEASE]
        at com.sun.proxy.$Proxy113.onSuccess(Unknown Source) [?:?]
        at org.zalando.fahrschein.NakadiReader.processBatch(NakadiReader.java:138) [fahrschein-0.4.3.jar:?]
        at org.zalando.fahrschein.NakadiReader.runInternal(NakadiReader.java:253) [fahrschein-0.4.3.jar:?]
        at org.zalando.fahrschein.NakadiReader.run(NakadiReader.java:212) [fahrschein-0.4.3.jar:?]
        at org.zalando.fahrschein.NakadiClient.listen(NakadiClient.java:97) [fahrschein-0.4.3.jar:?]
        at de.zalando.Gateway.listenForEvent(EventConsumerGateway.java:45) [main/:?]
        at de.zalando.Gateway$$FastClassBySpringCGLIB$$da487360.invoke(<generated>) [main/:?]
        at org.springframework.cglib.proxy.MethodProxy.invoke(MethodProxy.java:204) [spring-core-4.3.3.RELEASE.jar:4.3.3.RELEASE]
        at org.springframework.aop.framework.CglibAopProxy$CglibMethodInvocation.invokeJoinpoint(CglibAopProxy.java:720) [spring-aop-4.3.3.RELEASE.jar:4.3.3.RELEASE]
        at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:157) [spring-aop-4.3.3.RELEASE.jar:4.3.3.RELEASE]
        at org.springframework.aop.interceptor.AsyncExecutionInterceptor$1.call(AsyncExecutionInterceptor.java:115) [spring-aop-4.3.3.RELEASE.jar:4.3.3.RELEASE]
        at java.util.concurrent.FutureTask.run(FutureTask.java:266) [?:1.8.0_40]
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) [?:1.8.0_40]
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) [?:1.8.0_40]
        at java.lang.Thread.run(Thread.java:745) [?:1.8.0_40]
Caused by: org.postgresql.util.PSQLException: An I/O error occurred while sending to the backend.
        at org.postgresql.core.v3.QueryExecutorImpl.execute(QueryExecutorImpl.java:201) ~[postgresql-9.4-1205-jdbc42.jar:9.4]
        at org.postgresql.jdbc2.AbstractJdbc2Connection.executeTransactionCommand(AbstractJdbc2Connection.java:824) ~[postgresql-9.4-1205-jdbc42.jar:9.4]
        at org.postgresql.jdbc2.AbstractJdbc2Connection.commit(AbstractJdbc2Connection.java:848) ~[postgresql-9.4-1205-jdbc42.jar:9.4]
        at com.zaxxer.hikari.pool.ProxyConnection.commit(ProxyConnection.java:352) ~[HikariCP-2.4.3.jar:?]
        at com.zaxxer.hikari.pool.HikariProxyConnection.commit(HikariProxyConnection.java) ~[HikariCP-2.4.3.jar:?]
        at org.springframework.jdbc.datasource.DataSourceTransactionManager.doCommit(DataSourceTransactionManager.java:272) ~[spring-jdbc-4.3.3.RELEASE.jar:4.3.3.RELEASE]
        ... 22 more
Caused by: java.io.EOFException
        at org.postgresql.core.PGStream.ReceiveChar(PGStream.java:284) ~[postgresql-9.4-1205-jdbc42.jar:9.4]
        at org.postgresql.core.v3.QueryExecutorImpl.processResults(QueryExecutorImpl.java:1718) ~[postgresql-9.4-1205-jdbc42.jar:9.4]
        at org.postgresql.core.v3.QueryExecutorImpl.execute(QueryExecutorImpl.java:173) ~[postgresql-9.4-1205-jdbc42.jar:9.4]
        at org.postgresql.jdbc2.AbstractJdbc2Connection.executeTransactionCommand(AbstractJdbc2Connection.java:824) ~[postgresql-9.4-1205-jdbc42.jar:9.4]
        at org.postgresql.jdbc2.AbstractJdbc2Connection.commit(AbstractJdbc2Connection.java:848) ~[postgresql-9.4-1205-jdbc42.jar:9.4]
        at com.zaxxer.hikari.pool.ProxyConnection.commit(ProxyConnection.java:352) ~[HikariCP-2.4.3.jar:?]
        at com.zaxxer.hikari.pool.HikariProxyConnection.commit(HikariProxyConnection.java) ~[HikariCP-2.4.3.jar:?]
        at org.springframework.jdbc.datasource.DataSourceTransactionManager.doCommit(DataSourceTransactionManager.java:272) ~[spring-jdbc-4.3.3.RELEASE.jar:4.3.3.RELEASE]
        ... 22 more

ReadTimeout during Reconnection is not handled

In NakadiReader the catch guards only opening the stream. createParser will try to read some bytes from the stream, effectively running into a ReadTimeout if nothing happens.

                try {
                    LOG.info("Reconnecting after [{}] errors", errorCount);
                    try {
                        response = backoffStrategy.call(errorCount, e, this::openStream);
                    } catch (BackoffException e1) {
                        LOG.warn("Could not reconnect after [{}] errors", errorCount, e1);
                        return;
                    }
                    jsonParser = jsonFactory.createParser(response.getBody());
                } catch (InterruptedException e1) {
                    LOG.warn("Interrupted during reconnection");

                    Thread.currentThread().interrupt();
                    return;
                }

Provide X-Flow-Id header for all requests

Give clients the possibility to stop NakadiReader

As a user of the Fahrschein library, I want to be able to stop the Nakadi consumption manually. E.g.:

final Closeable listener = nakadi.client.listen(...);
listener.close();

One possibility might be to expose the JsonInput.close() or ClientHttpResponse.close().

Default constructor for org.zalando.fahrschein.domain.Partition

I'm facing the following exception while trying to get all partition for event as following:
nakadiClient.getPartitions(eventName)

Suggestion: could you please add default constructor or @ JsonCreator for the current constructor for org.zalando.fahrschein.domain.Partition class

The error as following:

com.fasterxml.jackson.databind.JsonMappingException: No suitable constructor found for type [simple type, class org.zalando.fahrschein.domain.Partition]: can not instantiate from JSON object (missing default constructor or creator, or perhaps need to add/enable type information?)
 at [Source: sun.net.www.protocol.http.HttpURLConnection$HttpInputStream@41914209; line: 1, column: 3] (through reference chain: java.util.ArrayList[0])
    at com.fasterxml.jackson.databind.JsonMappingException.from(JsonMappingException.java:148)
    at com.fasterxml.jackson.databind.deser.BeanDeserializerBase.deserializeFromObjectUsingNonDefault(BeanDeserializerBase.java:1106)
    at com.fasterxml.jackson.databind.deser.BeanDeserializer.deserializeFromObject(BeanDeserializer.java:296)
    at com.fasterxml.jackson.databind.deser.BeanDeserializer.deserialize(BeanDeserializer.java:133)
    at com.fasterxml.jackson.databind.deser.std.CollectionDeserializer.deserialize(CollectionDeserializer.java:245)
    at com.fasterxml.jackson.databind.deser.std.CollectionDeserializer.deserialize(CollectionDeserializer.java:217)
    at com.fasterxml.jackson.databind.deser.std.CollectionDeserializer.deserialize(CollectionDeserializer.java:25)
    at com.fasterxml.jackson.databind.ObjectMapper._readMapAndClose(ObjectMapper.java:3736)
    at com.fasterxml.jackson.databind.ObjectMapper.readValue(ObjectMapper.java:2803)
    at org.zalando.fahrschein.NakadiClient.getPartitions(NakadiClient.java:55)

Locking of partitions so that multiple consumers can coordinate

Reduce Log Spam on Remote Closing the Stream

Currently there is a warning log whenever remote closes the stream.

Currently...

NakadiReader waits for the next cursor to start
If a null value is returned from the stream a generic IOException is thrown
The exception is caught and logged as a warning
As expected the reconnection logic is triggered

What I'd like to see..

A specific StreamClosedException extends IOException is thrown`
IOExceptions of this kind are logged at INFO level without the stack trace

ParameterNamesModule and JavaTimeModule should be optional

The following error occurs when using the high level api without the ParameterNamesModule registered.

com.fasterxml.jackson.databind.JsonMappingException: Argument #0 of constructor [constructor for org.zalando.fahrschein.domain.Subscription, annotations: {interface com.fasterxml.jackson.annotation.JsonCreator=@com.fasterxml.jackson.annotation.JsonCreator(mode=DEFAULT)}] has no property name annotation; must have name when multiple-parameter constructor annotated as Creator

zalando-nakadi / fahrschein Goto Github PK

fahrschein's People

Contributors

Stargazers

Watchers

Forkers

fahrschein's Issues

Scenario

But Why

Recommend Projects

Recommend Topics

Recommend Org