major: Extend Consumer and Function for more cohesive API #303

astubbs · 2022-05-13T12:39:22Z

Experiment: extend Consumer and Function - centralises documentation and makes usage more explicit

This is a diverged fork of PR #268 - the relevant changes start from the commit titled:
Experiment: extend Consumer and Function - centralises documentation and makes usage more explicit

Checklist

Documentation (if applicable)
Changelog

Blocked by:

feature: #242 Explicit terminal and retry exceptions for cleaner logging and poison pills #291

PriorityQueue only provides a sorted `poll`, whereas TreeSet iterates in sorted order.

…queue # Conflicts: # parallel-consumer-core/src/test/java/io/confluent/parallelconsumer/truth/TruthGeneratorTests.java

rkolesnev · 2022-05-13T13:17:52Z

LGTM - nice addition i think.

…tation and makes usage more explicit

rkolesnev · 2022-05-13T12:41:36Z

parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/PCRetriableException.java

+
+/**
+ * A user's processing function can throw this exception, which signals to PC that processing of the message has failed,
+ * and that it should be retired at a later time.


Suggested change

* and that it should be retired at a later time.

* and that it should be retried at a later time.

rkolesnev · 2022-05-13T13:17:02Z

parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/RecordProcessor.java

+         *
+         * @param records the Kafka records to process
+         */
+        void accept(PollContext records);


Suggested change

void accept(PollContext records);

void accept(PollContext<K,V> records);

rkolesnev · 2022-05-13T13:17:19Z

parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/RecordProcessor.java

+         * @return the function result
+         */
+        @Override
+        R apply(PollContext t);


Suggested change

R apply(PollContext t);

R apply(PollContext<K,V> t);

rkolesnev · 2022-05-13T14:31:20Z

parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/RecordProcessor.java

+         * @see PollConsumer#accept(PollContext)
+         */
+        @Override
+        List<ProducerRecord<K, V>> apply(PollContext records);


Suggested change

List<ProducerRecord<K, V>> apply(PollContext records);

List<ProducerRecord<K, V>> apply(PollContext<K,V> records);

rkolesnev · 2022-05-13T14:33:06Z

parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/RecordProcessor.java

+    }
+
+    @FunctionalInterface
+    interface PollConsumerAndProducer<K, V> extends java.util.function.Function<PollContext<K, V>, List<ProducerRecord<K, V>>> {


Staying with PollConsumerAndProducer ? Would MessageTransformer or Processor or Mapper sound better given the messages in / out nature?

yeah - other thing is, if it affects your ideas - is it can produce many messages, not just one, from a single input

Would it be worth going to KStream DSL naming? have a map and flatmap where map takes 1 in - 1 out, and flatmap allows 1->many ? The typing needs fixed to allow transform of key / value anyway - so maybe can be done in separate PR that deals with it. Or Transform (with or without flatTransform).
Thinking about it - "Transformer" sounds best to me.
Are side effects allowed / expected in this scenario (i.e. external system calls for processing / enrichment)? If so something neutral like Processor would be ok i think.

rkolesnev · 2022-05-13T14:34:24Z

...consumer-example-core/src/main/java/io/confluent/parallelconsumer/examples/core/CoreApp.java

+        pc.pollAndProduceMany();
+        pc.poll(records ->);
+


presumably - to be removed?

yeah I was playing with what the api ended up looking like

…xp/merge-retry # Conflicts: # parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/internal/AbstractParallelEoSStreamProcessor.java # parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/internal/BrokerPollSystem.java # parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/internal/EpochAndRecordsMap.java # parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/offsets/OffsetMapCodecManager.java # parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/state/PartitionState.java # parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/state/PartitionStateManager.java # parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/state/ProcessingShard.java # parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/state/RemovedPartitionState.java # parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/state/ShardManager.java # parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/state/WorkContainer.java # parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/state/WorkManager.java # parallel-consumer-core/src/test/java/io/confluent/parallelconsumer/AbstractParallelEoSStreamProcessorTestBase.java # parallel-consumer-core/src/test/java/io/confluent/parallelconsumer/offsets/OffsetEncodingBackPressureTest.java # parallel-consumer-core/src/test/java/io/confluent/parallelconsumer/offsets/OffsetEncodingBackPressureUnitTest.java # parallel-consumer-core/src/test/java/io/confluent/parallelconsumer/offsets/OffsetEncodingTests.java # parallel-consumer-core/src/test/java/io/confluent/parallelconsumer/state/WorkManagerTest.java # parallel-consumer-core/src/test/resources/logback-test.xml

…unctional # Conflicts: # parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/ParallelStreamProcessor.java # parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/internal/AbstractParallelEoSStreamProcessor.java # parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/internal/UserFunctions.java

eddyv · 2023-06-15T14:34:40Z

Closing - Stale.

astubbs added 30 commits April 6, 2022 09:55

START: the basics of a single queue

21f08f6

step: remove work mailbox manager

3f1f3ff

step: BROKEN: assign epoch to record immediately

92b50cb

step - trying to test perf

1653bc2

update

103c677

logs

eed2190

fix: Debug output for sorted encoding pairs

02ee3cd

PriorityQueue only provides a sorted `poll`, whereas TreeSet iterates in sorted order.

save

83fda73

rebase update

2a37d46

step

b17c838

save

e1141e4

save

5a3bb55

save: unit test version of offset encoding backpressure test

6a1464c

save

0604934

omg - hashsets vs queues, wow

4eeb008

review

bcfc9c1

review

6054ac5

review

c968629

review

0f993dd

Merge remote-tracking branch 'confluent/master' into features/single-…

416fd2f

…queue # Conflicts: # parallel-consumer-core/src/test/java/io/confluent/parallelconsumer/truth/TruthGeneratorTests.java

step

189dc59

step

908d8ed

fix test

7547ec6

step - test fix?

939a15e

step - test fix?

3fa6ae3

step - test fix? - make sure unit test also cleans up

c44f50a

step - test fix? - make sure unit test also cleans up

eff0b13

START: Explicit retry exception for cleaner logging

a6dc167

step: reduce consumer max poll

74e0efb

step: loosen duplicate check a bit for jenkins

ae1ce22

Merge branch 'features/single-queue' into features/retry-exception

1b087e5

astubbs marked this pull request as draft May 13, 2022 12:39

astubbs changed the base branch from master to features/retry-exception May 13, 2022 12:41

astubbs changed the base branch from features/retry-exception to master May 13, 2022 12:41

START: Experiment: extend Consumer and Function - centralises documen…

88e9f0e

…tation and makes usage more explicit

astubbs force-pushed the features/extend-functional branch from 16b13c7 to cc379ec Compare May 13, 2022 13:39

astubbs changed the base branch from master to features/retry-exception May 13, 2022 13:41

astubbs changed the base branch from features/retry-exception to master May 13, 2022 13:41

astubbs force-pushed the features/extend-functional branch from cc379ec to 88e9f0e Compare May 13, 2022 14:28

rkolesnev suggested changes May 13, 2022

View reviewed changes

step

6862a56

astubbs changed the title ~~Experiment: extend Consumer and Function~~ major: Extend Consumer and Function for more cohesive API May 16, 2022

astubbs linked an issue Jul 15, 2022 that may be closed by this pull request

Let user function throw retry exception, instead of raw errors which pollute log #242

Closed

2 tasks

astubbs self-assigned this Jul 15, 2022

astubbs added 11 commits October 19, 2022 18:23

refactor confluentinc#409: Clarify truncation code

14f1d47

refactor confluentinc#409: Clarify truncation code - tests passing

24c61f1

refactor confluentinc#409: Clarify truncation code - tests passing

62be60b

bump jabel for j19

2122e9e

cleanup

0131d35

fix

158ca87

merge fix

40ba4e3

migrate

f97c3cb

step

f831137

astubbs mentioned this pull request Oct 20, 2022

WIP! Playing around with some interface naming #23

Closed

astubbs removed a link to an issue Oct 21, 2022

Let user function throw retry exception, instead of raw errors which pollute log #242

Closed

2 tasks

eddyv closed this Jun 15, 2023

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

major: Extend Consumer and Function for more cohesive API #303

major: Extend Consumer and Function for more cohesive API #303

astubbs commented May 13, 2022 •

edited

Loading

rkolesnev commented May 13, 2022

rkolesnev May 13, 2022

rkolesnev May 13, 2022

rkolesnev May 13, 2022

rkolesnev May 13, 2022

rkolesnev May 13, 2022

astubbs May 13, 2022

rkolesnev May 16, 2022

rkolesnev May 13, 2022

astubbs May 13, 2022

eddyv commented Jun 15, 2023

	* and that it should be retired at a later time.
	* and that it should be retried at a later time.

	void accept(PollContext records);
	void accept(PollContext<K,V> records);

	List<ProducerRecord<K, V>> apply(PollContext records);
	List<ProducerRecord<K, V>> apply(PollContext<K,V> records);

major: Extend Consumer and Function for more cohesive API #303

major: Extend Consumer and Function for more cohesive API #303

Conversation

astubbs commented May 13, 2022 • edited Loading

Checklist

Blocked by:

rkolesnev commented May 13, 2022

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

eddyv commented Jun 15, 2023

astubbs commented May 13, 2022 •

edited

Loading