KAFKA-9447: Add new customized EOS model example by abbccdda · Pull Request #8031 · apache/kafka

abbccdda · 2020-02-01T17:17:11Z

With the improvement of 447, we are now offering developers a better experience on writing their customized EOS apps with group subscription, instead of manual assignments. With the demo, user should be able to get started more quickly on writing their own EOS app, and understand the processing logic much better.

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

abbccdda · 2020-02-01T17:17:36Z

@@ -51,21 +51,18 @@ public Map<String, KafkaFuture<TopicDescription>> values() {
     */


side cleanup

abbccdda · 2020-02-01T17:17:41Z

@@ -48,23 +48,13 @@ public KafkaFuture<Map<String, TopicListing>> namesToListings() {
     * Return a future which yields a collection of TopicListing objects.


side cleanup

guozhangwang

LGTM overall, just some minor comments.

guozhangwang · 2020-02-04T21:39:42Z

+                    if (this.mode.equals("groupMode")) {
+                        producer.sendOffsetsToTransaction(positions, consumer.groupMetadata());
+                    } else {
+                        producer.sendOffsetsToTransaction(positions, consumerGroupId);


Just to refresh my memory: are we going to eventually deprecate this API, or are we going to keep both, and let users apply this one with manual assignment (like you did here)? I thought we are going to deprecate, but maybe I remembered it wrong.

No, we are intended to keep both

guozhangwang · 2020-02-04T21:41:24Z

+                    throw new KafkaException("Encountered fatal error during processing: " + e.getMessage());
+                }
+            }
+            messageRemaining.set(messagesRemaining(consumer));


Why we need an atomic long here? Seems there's no concurrency.

The tricky thing here is that if we define a primitive long outside of the rebalance callback, it won't compile.

guozhangwang · 2020-02-04T21:42:21Z

+
+    private ProducerRecord<Integer, String> transform(final ConsumerRecord<Integer, String> record) {
+        printWithTxnId("Transformed record (" + record.key() + "," + record.value() + ")");
+        return new ProducerRecord<>(outputTopic, record.key() / 2, "Transformed_" + record.value());


Why divide the key by two?

It's just a way of showing the key gets processed by message copier, all the produced message key are even keys.

guozhangwang · 2020-02-04T21:43:06Z

+                    messageProcessed += records.count();
+                } catch (CommitFailedException e) {
+                    // In case of a retriable exception, suggest aborting the ongoing transaction first for correctness.
+                    producer.abortTransaction();


prop: abortTransaction can also throw ProducerFenced.

Yea, which is ok as we will throw on ProducerFenced anyway

Not sure I understand: we try to capture ProducerFenced below and wrap it as a KafkaException, but here if we throw ProducerFenced it would not be captured and wrapped, is that intentional?

Oh I see your pt.

omkreddy · 2020-02-06T06:28:30Z

@abbccdda looks like :examples:spotbugsMain failing for newly added classes.

abbccdda · 2020-02-06T18:05:48Z

@omkreddy Thanks for catching that! Will do a fix in a minute

hachikuji · 2020-02-06T18:13:33Z

+                try {
+                    // Abort previous transaction if instructed.
+                    if (abortPreviousTransaction) {
+                        producer.abortTransaction();


It's a bit unconventional to have abort logic at the start of the loop. I think what users would expect is something like this:

try { producer.beginTransaction() producer.send(...) producer.sendOffsetsToTransaction(...) producer.commitTransaction() } catch (Exception ) { producer.abortTransaction() }

Conflicts: * build.gradle: moved avro plugin definition below newly added test retry plugin. * apache-github/trunk: MINOR: further InternalTopologyBuilder cleanup (apache#8046) MINOR: Add timer for update limit offsets (apache#8047) HOTFIX: Fix spotsbug failure in Kafka examples (apache#8051) KAFKA-9447: Add new customized EOS model example (apache#8031) KAFKA-8164: Add support for retrying failed (apache#8019) HOTFIX: checkstyle for newly added unit test KAFKA-9261; Client should handle unavailable leader metadata (apache#7770) MINOR: Fix typos introduced in KIP-559 (apache#8042) MINOR: Fixing null handilg in ValueAndTimestampSerializer (apache#7679) KAFKA-9113: Clean up task management and state management (apache#7997) MINOR: fix checkstyle issue in ConsumerConfig.java (apache#8038) KAFKA-9491; Increment high watermark after full log truncation (apache#8037) KAFKA-9477 Document RoundRobinAssignor as an option for partition.assignment.strategy (apache#8007) KAFKA-9074: Correct Connect’s `Values.parseString` to properly parse a time and timestamp literal (apache#7568) KAFKA-9492; Ignore record errors in ProduceResponse for older versions (apache#8030)

With the improvement of 447, we are now offering developers a better experience on writing their customized EOS apps with group subscription, instead of manual assignments. With the demo, user should be able to get started more quickly on writing their own EOS app, and understand the processing logic much better. Reviewers: Guozhang Wang <wangguoz@gmail.com>

…t-for-generated-requests * apache-github/trunk: (410 commits) KAFKA-8843: KIP-515: Zookeeper TLS support MINOR: Add missing quote for malformed line content (apache#8070) MINOR: Simplify KafkaProducerTest (apache#8044) KAFKA-9507; AdminClient should check for missing committed offsets (apache#8057) KAFKA-9519: Deprecate the --zookeeper flag in ConfigCommand (apache#8056) KAFKA-9509; Fixing flakiness of MirrorConnectorsIntegrationTest.testReplication (apache#8048) HOTFIX: Fix two test failures in JDK11 (apache#8063) DOCS - clarify transactionalID and idempotent behavior (apache#7821) MINOR: further InternalTopologyBuilder cleanup (apache#8046) MINOR: Add timer for update limit offsets (apache#8047) HOTFIX: Fix spotsbug failure in Kafka examples (apache#8051) KAFKA-9447: Add new customized EOS model example (apache#8031) KAFKA-8164: Add support for retrying failed (apache#8019) HOTFIX: checkstyle for newly added unit test KAFKA-9261; Client should handle unavailable leader metadata (apache#7770) MINOR: Fix typos introduced in KIP-559 (apache#8042) MINOR: Fixing null handilg in ValueAndTimestampSerializer (apache#7679) KAFKA-9113: Clean up task management and state management (apache#7997) MINOR: fix checkstyle issue in ConsumerConfig.java (apache#8038) KAFKA-9491; Increment high watermark after full log truncation (apache#8037) ...

abbccdda added 7 commits January 24, 2020 21:47

add eos example

b4a6a3a

extend producer demo with even number keys

0cf5868

build entire java class with workflow

b038d7f

create topic

bfc1b93

Only topic clean-up remaining

239e3bb

fix topic deletion

db292cc

fix group mode

392b719

abbccdda commented Feb 1, 2020

View reviewed changes

style fix

a675274

abbccdda force-pushed the eos_example branch 3 times, most recently from 6bf8724 to 340747f Compare February 1, 2020 22:20

More comments

217e3db

abbccdda force-pushed the eos_example branch from 340747f to 217e3db Compare February 1, 2020 22:24

abbccdda requested review from guozhangwang, hachikuji and mjsax February 1, 2020 22:25

guozhangwang reviewed Feb 4, 2020

View reviewed changes

abbccdda added 2 commits February 4, 2020 15:37

guozhang's comment

5827af7

move abort txn out and add commit reset

7a74824

abbccdda force-pushed the eos_example branch from 20be849 to 7a74824 Compare February 5, 2020 20:48

guozhangwang merged commit 9d17bf9 into apache:trunk Feb 6, 2020

hachikuji reviewed Feb 6, 2020

View reviewed changes

hachikuji mentioned this pull request Feb 6, 2020

Hotfix: spotsbug failure for EOS example #8051

Merged

3 tasks

		@@ -51,21 +51,18 @@ public Map<String, KafkaFuture<TopicDescription>> values() {
		*/

		@@ -48,23 +48,13 @@ public KafkaFuture<Map<String, TopicListing>> namesToListings() {
		* Return a future which yields a collection of TopicListing objects.

Conversation

abbccdda commented Feb 1, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Committer Checklist (excluded from commit message)

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

guozhangwang left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

omkreddy commented Feb 6, 2020

Uh oh!

abbccdda commented Feb 6, 2020

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

abbccdda commented Feb 1, 2020 •

edited

Loading