Changes to work with kafka 0.8 #7
Conversation
…etch to translate and throw exceptions on error in the FetchResponse. Add support for new river configuration named 'startFromNewestOffset'. This flag allow the river to be setup either to start from the newest or oldest partition offset if/when the river encounters an OffsetOutOfRangeException. Fixed a couple tests that were failing due to the new clientName param in the KafkaClient connect method.
…t timeout exceptions
|
+1 |
* Kafka 0.8.1 * ElasticSearch 1.0.1 * Scala 2.10.3
i got this error when i test.Test set: org.elasticsearch.river.kafka.KafkaClientTestTests run: 9, Failures: 0, Errors: 3, Skipped: 0, Time elapsed: 0.456 sec <<< FAILURE! Can you give me information to solved this error. |
|
Hi, I think the 0.8 version submitted by dtabwhite is OK, but there is still a problem whith the unit tests in maven. |
…ncorporating the configured river name in the location in zk where the offsets are stored.
- Have the river discover the set of brokers that exist from zk - Discover the leader broker for a given topic/partition - Don't store the brokerUrl in the zk offset path - On startup keep trying to discover/connect to kafka until there is a registered broker - On startup when no offsets are stored in zk, set the initial offset based on the 'startFromNewestOffset' setting
|
FYI - My latest commit in my fork changes the river to be more resilient within a clustered kafka setup. Here are the set of changes it introduces:
So essentially, the broker_host and broker_port go away with this change and are dynamically (re)discovered for the specified topic/partition. This is essential when running kafka in a clustered setup as the leadership for a given topic/partition can change and the river should really be able to deal with these changes on the fly. |
|
+1 |
These changes are based on an initial set of changes from clippPR to get the river to work with Kafka 0.8. I made a few changes based on the changes to the consumer offset being logical vs. physical as well as some subtle changes to the differences in exception handling between 0.7 and 0.8. Also made a fix for the case where connection issues occur in the dumpStats utility method so that the river will reconnect. One final change was to add a new river configuration param named 'startFromNewestOffset' (defaults to false) which allows you to configure the river to start either from the oldest or newest offset.