diff --git a/tools/src/test/java/org/apache/kafka/tools/consumer/ConsoleConsumerOptionsTest.java b/tools/src/test/java/org/apache/kafka/tools/consumer/ConsoleConsumerOptionsTest.java
index 4639ff63a85..843d9785cf8 100644
--- a/tools/src/test/java/org/apache/kafka/tools/consumer/ConsoleConsumerOptionsTest.java
+++ b/tools/src/test/java/org/apache/kafka/tools/consumer/ConsoleConsumerOptionsTest.java
@@ -151,7 +151,7 @@ public class ConsoleConsumerOptionsTest {
     }
 
     @Test
-    public void shouldParseValidSimpleConsumerValidConfigWithStringOffset() throws Exception {
+    public void shouldParseValidSimpleConsumerValidConfigWithStringOffsetDeprecated() throws Exception {
         String[] args = new String[]{
             "--bootstrap-server", "localhost:9092",
             "--topic", "test",
@@ -171,6 +171,27 @@ public class ConsoleConsumerOptionsTest {
         assertFalse(((DefaultMessageFormatter) config.formatter()).printValue());
     }
 
+    @Test
+    public void shouldParseValidSimpleConsumerValidConfigWithStringOffset() throws Exception {
+        String[] args = new String[]{
+            "--bootstrap-server", "localhost:9092",
+            "--topic", "test",
+            "--partition", "0",
+            "--offset", "LatEst",
+            "--formatter-property", "print.value=false"
+        };
+
+        ConsoleConsumerOptions config = new ConsoleConsumerOptions(args);
+
+        assertEquals("localhost:9092", config.bootstrapServer());
+        assertEquals("test", config.topicArg().orElse(""));
+        assertTrue(config.partitionArg().isPresent());
+        assertEquals(0, config.partitionArg().getAsInt());
+        assertEquals(-1, config.offsetArg());
+        assertFalse(config.fromBeginning());
+        assertFalse(((DefaultMessageFormatter) config.formatter()).printValue());
+    }
+
     @Test
     public void shouldParseValidConsumerConfigWithAutoOffsetResetLatestDeprecated() throws IOException {
         String[] args = new String[]{
@@ -355,7 +376,7 @@ public class ConsoleConsumerOptionsTest {
     }
 
     @Test
-    public void testCustomPropertyShouldBePassedToConfigureMethod() throws Exception {
+    public void testCustomPropertyShouldBePassedToConfigureMethodDeprecated() throws Exception {
         String[] args = new String[]{
             "--bootstrap-server", "localhost:9092",
             "--topic", "test",
@@ -377,6 +398,56 @@ public class ConsoleConsumerOptionsTest {
         assertTrue(keyDeserializer.isKey);
     }
 
+    @Test
+    public void testCustomPropertyShouldBePassedToConfigureMethod() throws Exception {
+        String[] args = new String[]{
+            "--bootstrap-server", "localhost:9092",
+            "--topic", "test",
+            "--formatter-property", "print.key=true",
+            "--formatter-property", "key.deserializer=org.apache.kafka.test.MockDeserializer",
+            "--formatter-property", "key.deserializer.my-props=abc"
+        };
+
+        ConsoleConsumerOptions config = new ConsoleConsumerOptions(args);
+
+        assertInstanceOf(DefaultMessageFormatter.class, config.formatter());
+        assertTrue(config.formatterArgs().containsKey("key.deserializer.my-props"));
+        DefaultMessageFormatter formatter = (DefaultMessageFormatter) config.formatter();
+        assertTrue(formatter.keyDeserializer().isPresent());
+        assertInstanceOf(MockDeserializer.class, formatter.keyDeserializer().get());
+        MockDeserializer keyDeserializer = (MockDeserializer) formatter.keyDeserializer().get();
+        assertEquals(1, keyDeserializer.configs.size());
+        assertEquals("abc", keyDeserializer.configs.get("my-props"));
+        assertTrue(keyDeserializer.isKey);
+    }
+
+    @Test
+    public void testCustomConfigShouldBePassedToConfigureMethodDeprecated() throws Exception {
+        Map<String, String> configs = new HashMap<>();
+        configs.put("key.deserializer.my-props", "abc");
+        configs.put("print.key", "false");
+        File propsFile = ToolsTestUtils.tempPropertiesFile(configs);
+        String[] args = new String[]{
+            "--bootstrap-server", "localhost:9092",
+            "--topic", "test",
+            "--property", "print.key=true",
+            "--property", "key.deserializer=org.apache.kafka.test.MockDeserializer",
+            "--formatter-config", propsFile.getAbsolutePath()
+        };
+
+        ConsoleConsumerOptions config = new ConsoleConsumerOptions(args);
+
+        assertInstanceOf(DefaultMessageFormatter.class, config.formatter());
+        assertTrue(config.formatterArgs().containsKey("key.deserializer.my-props"));
+        DefaultMessageFormatter formatter = (DefaultMessageFormatter) config.formatter();
+        assertTrue(formatter.keyDeserializer().isPresent());
+        assertInstanceOf(MockDeserializer.class, formatter.keyDeserializer().get());
+        MockDeserializer keyDeserializer = (MockDeserializer) formatter.keyDeserializer().get();
+        assertEquals(1, keyDeserializer.configs.size());
+        assertEquals("abc", keyDeserializer.configs.get("my-props"));
+        assertTrue(keyDeserializer.isKey);
+    }
+
     @Test
     public void testCustomConfigShouldBePassedToConfigureMethod() throws Exception {
         Map<String, String> configs = new HashMap<>();
@@ -386,8 +457,8 @@ public class ConsoleConsumerOptionsTest {
         String[] args = new String[]{
             "--bootstrap-server", "localhost:9092",
             "--topic", "test",
-            "--property", "print.key=true",
-            "--property", "key.deserializer=org.apache.kafka.test.MockDeserializer",
+            "--formatter-property", "print.key=true",
+            "--formatter-property", "key.deserializer=org.apache.kafka.test.MockDeserializer",
             "--formatter-config", propsFile.getAbsolutePath()
         };
 
diff --git a/tools/src/test/java/org/apache/kafka/tools/consumer/ConsoleShareConsumerOptionsTest.java b/tools/src/test/java/org/apache/kafka/tools/consumer/ConsoleShareConsumerOptionsTest.java
index fecf53dbbec..a097a9bf536 100644
--- a/tools/src/test/java/org/apache/kafka/tools/consumer/ConsoleShareConsumerOptionsTest.java
+++ b/tools/src/test/java/org/apache/kafka/tools/consumer/ConsoleShareConsumerOptionsTest.java
@@ -222,7 +222,7 @@ public class ConsoleShareConsumerOptionsTest {
     }
 
     @Test
-    public void testCustomPropertyShouldBePassedToConfigureMethod() throws Exception {
+    public void testCustomPropertyShouldBePassedToConfigureMethodDeprecated() throws Exception {
         String[] args = new String[]{
             "--bootstrap-server", "localhost:9092",
             "--topic", "test",
@@ -244,6 +244,56 @@ public class ConsoleShareConsumerOptionsTest {
         assertTrue(keyDeserializer.isKey);
     }
 
+    @Test
+    public void testCustomPropertyShouldBePassedToConfigureMethod() throws Exception {
+        String[] args = new String[]{
+            "--bootstrap-server", "localhost:9092",
+            "--topic", "test",
+            "--formatter-property", "print.key=true",
+            "--formatter-property", "key.deserializer=org.apache.kafka.test.MockDeserializer",
+            "--formatter-property", "key.deserializer.my-props=abc"
+        };
+
+        ConsoleShareConsumerOptions config = new ConsoleShareConsumerOptions(args);
+
+        assertInstanceOf(DefaultMessageFormatter.class, config.formatter());
+        assertTrue(config.formatterArgs().containsKey("key.deserializer.my-props"));
+        DefaultMessageFormatter formatter = (DefaultMessageFormatter) config.formatter();
+        assertTrue(formatter.keyDeserializer().isPresent());
+        assertInstanceOf(MockDeserializer.class, formatter.keyDeserializer().get());
+        MockDeserializer keyDeserializer = (MockDeserializer) formatter.keyDeserializer().get();
+        assertEquals(1, keyDeserializer.configs.size());
+        assertEquals("abc", keyDeserializer.configs.get("my-props"));
+        assertTrue(keyDeserializer.isKey);
+    }
+
+    @Test
+    public void testCustomConfigShouldBePassedToConfigureMethodDeprecated() throws Exception {
+        Map<String, String> configs = new HashMap<>();
+        configs.put("key.deserializer.my-props", "abc");
+        configs.put("print.key", "false");
+        File propsFile = ToolsTestUtils.tempPropertiesFile(configs);
+        String[] args = new String[]{
+            "--bootstrap-server", "localhost:9092",
+            "--topic", "test",
+            "--property", "print.key=true",
+            "--property", "key.deserializer=org.apache.kafka.test.MockDeserializer",
+            "--formatter-config", propsFile.getAbsolutePath()
+        };
+
+        ConsoleShareConsumerOptions config = new ConsoleShareConsumerOptions(args);
+
+        assertInstanceOf(DefaultMessageFormatter.class, config.formatter());
+        assertTrue(config.formatterArgs().containsKey("key.deserializer.my-props"));
+        DefaultMessageFormatter formatter = (DefaultMessageFormatter) config.formatter();
+        assertTrue(formatter.keyDeserializer().isPresent());
+        assertInstanceOf(MockDeserializer.class, formatter.keyDeserializer().get());
+        MockDeserializer keyDeserializer = (MockDeserializer) formatter.keyDeserializer().get();
+        assertEquals(1, keyDeserializer.configs.size());
+        assertEquals("abc", keyDeserializer.configs.get("my-props"));
+        assertTrue(keyDeserializer.isKey);
+    }
+
     @Test
     public void testCustomConfigShouldBePassedToConfigureMethod() throws Exception {
         Map<String, String> configs = new HashMap<>();
@@ -253,8 +303,8 @@ public class ConsoleShareConsumerOptionsTest {
         String[] args = new String[]{
             "--bootstrap-server", "localhost:9092",
             "--topic", "test",
-            "--property", "print.key=true",
-            "--property", "key.deserializer=org.apache.kafka.test.MockDeserializer",
+            "--formatter-property", "print.key=true",
+            "--formatter-property", "key.deserializer=org.apache.kafka.test.MockDeserializer",
             "--formatter-config", propsFile.getAbsolutePath()
         };