Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Kafka support #1510

Merged
merged 11 commits into from
Apr 23, 2020
Merged

Kafka support #1510

merged 11 commits into from
Apr 23, 2020

Conversation

jbescos
Copy link
Member

@jbescos jbescos commented Mar 13, 2020

Find here the spec of this:
https://download.eclipse.org/microprofile/microprofile-reactive-messaging-1.0/microprofile-reactive-messaging-spec.pdf

It was mostly done by @danielkec months ago. I only took his changes and I did some small fixes and some tests.

@danielkec danielkec self-assigned this Mar 16, 2020
@danielkec danielkec added the reactive Reactive streams and related components label Mar 16, 2020
@danielkec danielkec added this to the 2.0.0 milestone Mar 16, 2020
tomas-langer
tomas-langer requested changes Mar 16, 2020
View reviewed changes
Copy link
Member

@tomas-langer tomas-langer left a comment

Choose a reason for hiding this comment

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

I have commented on a few issues.
Please have a look at https://github.com/oracle/helidon/blob/master/DEV-GUIDELINES.md and fix your PR accordingly.
Please add module-info.java to your module.

messaging/kafka/pom.xml Outdated Show resolved Hide resolved
messaging/kafka/pom.xml Outdated Show resolved Hide resolved
messaging/kafka/pom.xml Outdated Show resolved Hide resolved
messaging/kafka/pom.xml Outdated Show resolved Hide resolved
messaging/kafka/src/main/java/io/helidon/messaging/kafka/connector/KafkaConnectorFactory.java Outdated Show resolved Hide resolved
messaging/kafka/src/main/java/io/helidon/messaging/kafka/SimpleKafkaConsumer.java Outdated Show resolved Hide resolved
messaging/kafka/src/main/java/io/helidon/messaging/kafka/SimpleKafkaConsumer.java Outdated Show resolved Hide resolved
messaging/kafka/src/main/java/io/helidon/messaging/kafka/SimpleKafkaConsumer.java Outdated Show resolved Hide resolved
messaging/kafka/src/main/java/io/helidon/messaging/kafka/SimpleKafkaConsumer.java Outdated Show resolved Hide resolved
messaging/kafka/src/main/java/io/helidon/messaging/kafka/SimpleKafkaConsumer.java Outdated Show resolved Hide resolved
@tomas-langer tomas-langer added the dependencies Pull requests that update a dependency file label Mar 16, 2020
jbescos
jbescos commented Mar 18, 2020
View reviewed changes
...nectors/kafka/src/main/java/io/helidon/microprofile/connectors/kafka/BasicKafkaConsumer.java Outdated
public void close() {
// Stops pooling
consumer.wakeup();
while (running.get()) {
Copy link
Member Author

Choose a reason for hiding this comment

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

I am not sure if this while is acceptable because of CPU usage. However I put it there because the time frame must be very small (few milliseconds) or zero.

Let me know if you prefer other way, with count down latch for example.

Copy link
Member Author

Choose a reason for hiding this comment

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

I will do this with a synchronize.

Copy link
Member Author

Choose a reason for hiding this comment

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

In this case I think it is much better to use a lock.

danielkec
danielkec reviewed Mar 18, 2020
View reviewed changes
...nectors/kafka/src/main/java/io/helidon/microprofile/connectors/kafka/BasicKafkaConsumer.java Outdated
Comment on lines 95 to 98
@Override
public void request(long n) {
// Pushing Kafka consumer doesn't support requests.
}
Copy link
Contributor

Choose a reason for hiding this comment

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

Uf this was my super bad idea, we have to do something about this, it basically ignores backpressure. Something like this would be much better:
https://github.com/oracle/helidon/blob/1e5ae594bc356ecd1283e487a7e7f85e26355ee9/microprofile/reactive-streams/src/main/java/io/helidon/microprofile/reactive/EmittingPublisher.java

But that depends on protected RS with SequentialSubscriber,
I expect David to remove SequentialSubscriber from RS implemetation in #1511 so it gets little more complicated then.

Copy link
Member Author

Choose a reason for hiding this comment

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

I integrated the EmittingSubscriber

tomas-langer
tomas-langer requested changes Mar 19, 2020
View reviewed changes
...nectors/kafka/src/main/java/io/helidon/microprofile/connectors/kafka/BasicKafkaConsumer.java Outdated Show resolved Hide resolved
...nectors/kafka/src/main/java/io/helidon/microprofile/connectors/kafka/BasicKafkaConsumer.java Outdated Show resolved Hide resolved
...nectors/kafka/src/main/java/io/helidon/microprofile/connectors/kafka/BasicKafkaConsumer.java Outdated Show resolved Hide resolved
...nectors/kafka/src/main/java/io/helidon/microprofile/connectors/kafka/BasicKafkaConsumer.java Outdated Show resolved Hide resolved
...nectors/kafka/src/main/java/io/helidon/microprofile/connectors/kafka/BasicKafkaConsumer.java Outdated Show resolved Hide resolved
...tors/kafka/src/main/java/io/helidon/microprofile/connectors/kafka/KafkaConnectorFactory.java Outdated Show resolved Hide resolved
...tors/kafka/src/main/java/io/helidon/microprofile/connectors/kafka/KafkaConnectorFactory.java Outdated Show resolved Hide resolved
...tors/kafka/src/main/java/io/helidon/microprofile/connectors/kafka/KafkaConnectorFactory.java Outdated Show resolved Hide resolved
...tors/kafka/src/main/java/io/helidon/microprofile/connectors/kafka/KafkaConnectorFactory.java Outdated Show resolved Hide resolved
...tors/kafka/src/main/java/io/helidon/microprofile/connectors/kafka/KafkaConnectorFactory.java Outdated Show resolved Hide resolved
@romain-grecourt
Copy link
Contributor

/oca-checked

@jbescos jbescos force-pushed the kafkaConnector branch 7 times, most recently from aad54c4 to ede74fc Compare March 30, 2020 07:05
@tomas-langer tomas-langer reopened this Mar 30, 2020
@jbescos jbescos force-pushed the kafkaConnector branch 11 times, most recently from 89106e8 to 3aac549 Compare April 3, 2020 06:45
@jbescos jbescos force-pushed the kafkaConnector branch 2 times, most recently from 6ce484a to 5a1897e Compare April 16, 2020 12:11
danielkec
danielkec previously approved these changes Apr 16, 2020
View reviewed changes
Copy link
Contributor

@danielkec danielkec left a comment

Choose a reason for hiding this comment

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

👍

@jbescos jbescos force-pushed the kafkaConnector branch 3 times, most recently from 3d22f6c to 9f76d6c Compare April 17, 2020 06:44
tomas-langer
tomas-langer requested changes Apr 17, 2020
View reviewed changes
Copy link
Member

@tomas-langer tomas-langer left a comment

Choose a reason for hiding this comment

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

Requested changes through DM.

@jbescos jbescos force-pushed the kafkaConnector branch 6 times, most recently from c556cc8 to e17c7ab Compare April 21, 2020 10:11
jbescos and others added 9 commits April 22, 2020 14:14
@jbescos
Kafka support
3a896fa 
Signed-off-by: Jorge Bescos Gascon <jorge.bescos.gascon@oracle.com>
@tomas-langer @jbescos
A few small changes.
b5e59d0 
Signed-off-by: Tomas Langer <tomas.langer@oracle.com>
@jbescos
To trigger the build
0bae61f 
Signed-off-by: Jorge Bescos Gascon <jorge.bescos.gascon@oracle.com>
@jbescos
TCK in KafkaConnector
32be94a 
Signed-off-by: Jorge Bescos Gascon <jorge.bescos.gascon@oracle.com>
@jbescos
Move tck tests inside kafka module
150c3e8 
Signed-off-by: Jorge Bescos Gascon <jorge.bescos.gascon@oracle.com>
@jbescos
Refactorings and ACK fixes
c455898 
Signed-off-by: Jorge Bescos Gascon <jorge.bescos.gascon@oracle.com>
@jbescos
Move from microprofile to new module messaging
430995f 
Signed-off-by: Jorge Bescos Gascon <jorge.bescos.gascon@oracle.com>
@jbescos
Improving the tests
7d68ce5 
Signed-off-by: Jorge Bescos Gascon <jorge.bescos.gascon@oracle.com>
@jbescos
Remove public modifiers in TCK tests, and improve someEventsNoAckWith…
8ba7b9b 
…DifferentPartitions test

Signed-off-by: Jorge Bescos Gascon <jorge.bescos.gascon@oracle.com>
@jbescos
Tomas review comments iteration 1
8f82cfa 
Signed-off-by: Jorge Bescos Gascon <jorge.bescos.gascon@oracle.com>
@jbescos
Copyright fixes
620d5a4 
Signed-off-by: Jorge Bescos Gascon <jorge.bescos.gascon@oracle.com>
@jbescos jbescos merged commit 23a4d65 into helidon-io:master Apr 23, 2020
@danielkec danielkec mentioned this pull request Jul 7, 2020
Closed
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Reviewers

@tomas-langer tomas-langer tomas-langer approved these changes

@danielkec danielkec danielkec left review comments

Assignees

@danielkec danielkec

Labels
dependencies Pull requests that update a dependency file messaging Reactive Messaging reactive Reactive streams and related components
Projects
None yet
Milestone
2.0.0
Development

Successfully merging this pull request may close these issues.

None yet
4 participants
@jbescos @romain-grecourt @danielkec @tomas-langer