Comments (4)
Yeah, it seems like the benefit of changing it to be CommitRecovery[F[_]]
will cause more difficulty in working with the API than it is worth. I agree that it should stay as-is, at least for now 🙂
from fs2-kafka.
I’ll take a look at this!
from fs2-kafka.
I may be missing something here, but it seems like this change is going to cause both Jitter
and Timer
to be required by all callers of CommitRecovery#Default
.
def Default[F[_]](
implicit F: MonadError[F, Throwable],
jitter: Jitter[F],
timer: Timer[F]
): CommitRecovery[F]
This is assuming that the new CommitRecovery
looks like:
abstract class CommitRecovery[F[_]] {
def recoverCommitWith(
offsets: Map[TopicPartition, OffsetAndMetadata],
commit: F[Unit]
): Throwable => F[Unit]
}
Do we think it is alright to "leak" Jitter
and Timer
around the codebase? Or is there another way you had in mind for doing this?
from fs2-kafka.
Thanks for taking a look at this @lewisjkl! The default CommitRecovery
would only really be used in ConsumerSettings
, where a default Jitter
instance could be created. I think we can't get away from ConsumerSettings
requiring Timer
though, even though it might not be used (as is the case when a custom CommitRecovery
is used). Sounds like maybe we should let it stay as-is. WDYT?
from fs2-kafka.
Related Issues (20)
- Kafka Queues? HOT 2
- Records are missing when second rebalance happens. HOT 6
- Is there a way to check if the producer is ready to produce?
- One message from Consumer? HOT 6
- withCreateConsumer and withCreateProducter methods have been removed HOT 13
- TransactionalKafkaProducer don't commit offset with empty records HOT 2
- Does KafkaConsumer.stopConsuming actually allow to process all in-flight messages? HOT 7
- It would be nice to have a `.option` method on `RecordSerializer` HOT 2
- Using Codec[A] with KafkaStreams
- CE 3.5.0 async cancelation semantics HOT 3
- Question around custom Partition assignments HOT 3
- Multiple consumers read from the same partition after rebalance HOT 2
- Any reason `RebalanceInProgressException` isn't considered in`CommitRecovery.Default`? HOT 1
- Confluent platform incompatibility HOT 3
- Stuck partitions - question about the pause/resume logic HOT 20
- upgrade 3.x to CE 3.5.1 / fs2 3.8.0 HOT 5
- Feature request (possibly) : Dynamic metering on Streams HOT 2
- retry CommitTimeoutException
- KafkaConsumer closed prematurely on finite streams
- Zipping finite KafkaConsumer-based streams doesn't behave as expected (first stream gets recycled after first 1000 elements) HOT 1
Recommend Projects
-
React
A declarative, efficient, and flexible JavaScript library for building user interfaces.
-
Vue.js
🖖 Vue.js is a progressive, incrementally-adoptable JavaScript framework for building UI on the web.
-
Typescript
TypeScript is a superset of JavaScript that compiles to clean JavaScript output.
-
TensorFlow
An Open Source Machine Learning Framework for Everyone
-
Django
The Web framework for perfectionists with deadlines.
-
Laravel
A PHP framework for web artisans
-
D3
Bring data to life with SVG, Canvas and HTML. 📊📈🎉
-
Recommend Topics
-
javascript
JavaScript (JS) is a lightweight interpreted programming language with first-class functions.
-
web
Some thing interesting about web. New door for the world.
-
server
A server is a program made to process requests and deliver data to clients.
-
Machine learning
Machine learning is a way of modeling and interpreting data that allows a piece of software to respond intelligently.
-
Visualization
Some thing interesting about visualization, use data art
-
Game
Some thing interesting about game, make everyone happy.
Recommend Org
-
Facebook
We are working to build community through open source technology. NB: members must have two-factor auth.
-
Microsoft
Open source projects and samples from Microsoft.
-
Google
Google ❤️ Open Source for everyone.
-
Alibaba
Alibaba Open Source for everyone
-
D3
Data-Driven Documents codes.
-
Tencent
China tencent open source team.
from fs2-kafka.