KAFKA-14576: Move ConsoleConsumer to tools by mimaison · Pull Request #15274 · apache/kafka

mimaison · 2024-01-26T16:51:44Z

Committer Checklist (excluded from commit message)

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

OmniaGM

Nice work, overall looks good. I just left few comments

OmniaGM · 2024-01-31T11:48:25Z

+        }
+    }
+
+    static void run(ConsoleConsumerOptions conf) {


Just for consistency can we rename conf to optslike L.57?

OmniaGM · 2024-01-31T12:10:08Z

+
+        Iterator<ConsumerRecord<byte[], byte[]>> recordIter = Collections.emptyIterator();
+
+        public ConsumerWrapper(Optional<String> topic,


Do we really need to wrap these parameters in Optional? Also, topic, partitionId, offset, includedTopics values all come from ConsoleConsumerOptions so maybe we can just send ConsoleConsumerOptions here and get ripped of the extra if/else in line 72 and the optional wrappers

I opted to copy the previous logic to make it easier to review and avoid changing any behavior by mistake. I agree this could be refactored but I'd prefer doing that in a follow up PR.

OmniaGM · 2024-01-31T12:15:09Z

+            } else if (!topic.isPresent() && !partitionId.isPresent() && !offset.isPresent() && includedTopics.isPresent()) {
+                consumer.subscribe(Pattern.compile(includedTopics.get()));
+            } else {
+                throw new IllegalArgumentException("An invalid combination of arguments is provided. " +


this line feel more fit for ConsoleConsumerOptions.checkRequiredArgs if the combination is invalid we should exit earlier.

I kept the logic this way to make it easier to review and compare against the previous Scala code. There are tons of options and even with the many tests we have and the lot of manual testing I did, I don't want to change the behavior of this tool.

What you propose sounds like a nice improvement but I'd prefer keeping this PR as a "translation" and do improvements in follow up PRs.

OmniaGM · 2024-01-31T12:37:39Z

+            this.timeoutMs = timeoutMs;
+
+            if (topic.isPresent() && partitionId.isPresent() && offset.isPresent() && !includedTopics.isPresent()) {
+                seek(topic.get(), partitionId.getAsInt(), offset.getAsLong());


If we opt in for my suggestion above we can also refactor seek to decide what to do if offset is provided or not instead of the first 2 ifs we have here.

OmniaGM · 2024-01-31T12:43:57Z

+                seek(topic.get(), partitionId.getAsInt(), offset.getAsLong());
+            } else if (topic.isPresent() && partitionId.isPresent() && !offset.isPresent() && !includedTopics.isPresent()) {
+                // default to latest if no offset is provided
+                seek(topic.get(), partitionId.getAsInt(), ListOffsetsRequest.LATEST_TIMESTAMP);


can't we use ConsoleConsumerOptions.parseOffset to either grab the given offset or the default instead of these 2 ifs?

OmniaGM · 2024-01-31T12:56:55Z

+    }
+
+    private MessageFormatter buildFormatter() {
+        MessageFormatter formatter = null;


is =null necessary?

Yes it's needed otherwise the compiler complains formatter may not have been initialized as it does not understand the catch clause causes the tool to exit.

OmniaGM · 2024-01-31T12:58:54Z

+    private byte[] headersSeparator = utfBytes(",");
+    private byte[] nullLiteral = utfBytes("null");
+
+    private Optional<Deserializer<?>> keyDeserializer;


Shouldn't we specify the default value to be Optional.empty here

OmniaGM · 2024-01-31T12:59:07Z

+    private byte[] nullLiteral = utfBytes("null");
+
+    private Optional<Deserializer<?>> keyDeserializer;
+    private Optional<Deserializer<?>> valueDeserializer;


OmniaGM · 2024-01-31T12:59:18Z

+
+    private Optional<Deserializer<?>> keyDeserializer;
+    private Optional<Deserializer<?>> valueDeserializer;
+    private Optional<Deserializer<?>> headersDeserializer;


same as above

jlprat

Thanks for the PR @mimaison
I have a couple of comments, they are minor, but I believe the ones in LoggingMessageFormatter should be addressed as the output would be different

jlprat

To the best of my "in-head Scala -> Java translator" knowledge, the code generated is equivalent to the original one.
LGTM, thanks @mimaison

mimaison · 2024-02-05T15:16:35Z

 from kafkatest.directory_layout.kafka_path import KafkaPathResolverMixin
 from kafkatest.services.monitor.jmx import JmxMixin, JmxTool
-from kafkatest.version import DEV_BRANCH, LATEST_0_8_2, LATEST_0_9, LATEST_0_10_0, V_0_10_0_0, V_0_11_0_0, V_2_0_0
+from kafkatest.version import DEV_BRANCH, LATEST_0_8_2, LATEST_0_9, LATEST_0_10_0, V_0_10_0_0, V_0_11_0_0, V_2_0_0, LATEST_3_7


LATEST_3_7 does not exist yet, we need to wait for 3.7.0 to be out before adding it.

Do I understand it right that you want to merge this only after 3.7.0 is released?

Actually since DEV_VERSION already points to 3.8, I should be able to introduce LATEST_3_7 even if it's not released yet. I'll try that.

That seems to work. However while running the system tests I found an issue loading configs via --consumer.config. I pushed c6355fe to fix it.

I've kicked another run on our system tests CI, hopefully it will be clean now.

mimaison · 2024-02-05T15:20:02Z

+            if node.version > LATEST_3_7:
+                cmd += " --formatter org.apache.kafka.tools.consumer.LoggingMessageFormatter"
+            else:
+                cmd += " --formatter kafka.tools.LoggingMessageFormatter"


LoggingMessageFormatter, DefaultMessageFormatter, NoOpMessageFormatter are not part of the public API and as far as I can tell the class names are not visible anywhere so this shouldn't be considered an API change.

mimaison · 2024-02-13T11:08:49Z

I rebased on trunk to resolve conflicts. It seems my last fix has resolved the issues with the system tests so it is ready to review again.

jlprat

LGTM. Thanks @mimaison

jlprat · 2024-02-13T11:17:13Z

+        Properties consumerProps = new Properties();
+        consumerProps.putAll(consumerPropsFromFile);


Strange that this change is not doing the exact same thing. But reading the API, constructor with Property says that it creates an empty list with the passed properties as defaults. Tricky API design from Java's part...

Yep! Default values are only retrieved when using getProperty() and not when using get(). This is why the unit test was passing.

mimaison · 2024-02-13T11:57:06Z

@OmniaGM do you want to take another look?

OmniaGM

LGTM, thanks @mimaison. We properly just need a jira for some of the improvements and refactoring here to get ripe of some of the inherited stuff from translating scala into java so we don't forget about them.

mimaison · 2024-02-13T12:23:11Z

Thanks @OmniaGM !
I created https://issues.apache.org/jira/browse/KAFKA-16246 to track the follow up refactoring work.

mimaison · 2024-02-13T17:45:57Z

None of the test failures seem related, merging to trunk

Reviewers: Josep Prat <josep.prat@aiven.io>, Omnia Ibrahim <o.g.h.ibrahim@gmail.com>

mimaison force-pushed the KAFKA-14576 branch from 79d9238 to 2bcbd9a Compare January 29, 2024 09:56

mimaison mentioned this pull request Jan 29, 2024

MINOR: Clean up core metrics, migration, network, raft, security, serializer, tools, utils, and zookeeper modules #15279

Merged

3 tasks

OmniaGM reviewed Jan 31, 2024

View reviewed changes

mimaison force-pushed the KAFKA-14576 branch from 8b17fcf to 74edbee Compare February 1, 2024 21:15

jlprat reviewed Feb 5, 2024

View reviewed changes

jlprat approved these changes Feb 5, 2024

View reviewed changes

mimaison commented Feb 5, 2024

View reviewed changes

mimaison force-pushed the KAFKA-14576 branch from c6355fe to ff18f43 Compare February 9, 2024 18:10

mimaison added 8 commits February 13, 2024 10:25

KAFKA-14576: Move ConsoleConsumer to tools

e1bdd94

Address feedback

dcf3b28

Fix system tests

2ab9fa2

Fix compilation/checktyle issues

0c6edcb

Address Josep's comments

90643a1

Fix for system tests

abe3851

Add LATEST_3_7 to version.py

bc1a4fa

Fix loading props from file

bd32996

mimaison force-pushed the KAFKA-14576 branch from ff18f43 to bd32996 Compare February 13, 2024 10:53

jlprat approved these changes Feb 13, 2024

View reviewed changes

OmniaGM approved these changes Feb 13, 2024

View reviewed changes

mimaison merged commit 0bf830f into apache:trunk Feb 13, 2024

mimaison deleted the KAFKA-14576 branch February 13, 2024 18:24

yyu1993 pushed a commit to yyu1993/kafka that referenced this pull request Feb 15, 2024

KAFKA-14576: Move ConsoleConsumer to tools (apache#15274)

007d7ad

Reviewers: Josep Prat <josep.prat@aiven.io>, Omnia Ibrahim <o.g.h.ibrahim@gmail.com>

clolov pushed a commit to clolov/kafka that referenced this pull request Apr 5, 2024

KAFKA-14576: Move ConsoleConsumer to tools (apache#15274)

8ecfc1f

Reviewers: Josep Prat <josep.prat@aiven.io>, Omnia Ibrahim <o.g.h.ibrahim@gmail.com>

Phuc-Hong-Tran pushed a commit to Phuc-Hong-Tran/kafka that referenced this pull request Jun 6, 2024

KAFKA-14576: Move ConsoleConsumer to tools (apache#15274)

f35d689

Reviewers: Josep Prat <josep.prat@aiven.io>, Omnia Ibrahim <o.g.h.ibrahim@gmail.com>


		Iterator<ConsumerRecord<byte[], byte[]>> recordIter = Collections.emptyIterator();

		public ConsumerWrapper(Optional<String> topic,

		Properties consumerProps = new Properties();
		consumerProps.putAll(consumerPropsFromFile);

Conversation

mimaison commented Jan 26, 2024

Committer Checklist (excluded from commit message)

Uh oh!

OmniaGM left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

OmniaGM Jan 31, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

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!

OmniaGM Jan 31, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

OmniaGM Jan 31, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

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!

jlprat left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

jlprat 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!

mimaison commented Feb 13, 2024

Uh oh!

jlprat 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!

mimaison commented Feb 13, 2024

Uh oh!

OmniaGM left a comment

Choose a reason for hiding this comment

Uh oh!

mimaison commented Feb 13, 2024

Uh oh!

mimaison commented Feb 13, 2024

Uh oh!

Reviewers

Assignees

Labels

Projects

OmniaGM Jan 31, 2024 •

edited

Loading

OmniaGM Jan 31, 2024 •

edited

Loading

OmniaGM Jan 31, 2024 •

edited

Loading