diff --git a/CHANGELOG.md b/CHANGELOG.md
index d6bb53291..ee69938be 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,9 +1,9 @@
 ## v0.10.3 [unreleased]
 
 ### Release Notes
-- **Breaking Change** all users of the exec plugin must specify
-`data_format = "json"` in the config file for the plugin to continue
-parsing JSON data as it does now.
+- Users of the `exec` and `kafka_consumer` can now specify the incoming data
+format that they would like to parse. Currently supports: "json", "influx", and
+"graphite"
 - More info on parsing arbitrary data formats can be found
 [here](https://github.com/influxdata/telegraf/blob/master/DATA_FORMATS_INPUT.md)
 
diff --git a/internal/config/config.go b/internal/config/config.go
index 42eb9df24..766ba1189 100644
--- a/internal/config/config.go
+++ b/internal/config/config.go
@@ -611,7 +611,10 @@ func buildParser(name string, tbl *ast.Table) (parsers.Parser, error) {
 		}
 	}
 
-	if c.DataFormat == "" {
+	// Legacy support, exec plugin originally parsed JSON by default.
+	if name == "exec" && c.DataFormat == "" {
+		c.DataFormat = "json"
+	} else if c.DataFormat == "" {
 		c.DataFormat = "influx"
 	}
 
diff --git a/internal/config/config_test.go b/internal/config/config_test.go
index c7f48b7c3..0e9f2c967 100644
--- a/internal/config/config_test.go
+++ b/internal/config/config_test.go
@@ -92,7 +92,7 @@ func TestConfig_LoadDirectory(t *testing.T) {
 		"Testdata did not produce correct memcached metadata.")
 
 	ex := inputs.Inputs["exec"]().(*exec.Exec)
-	p, err := parsers.NewInfluxParser()
+	p, err := parsers.NewJSONParser("exec", nil, nil)
 	assert.NoError(t, err)
 	ex.SetParser(p)
 	ex.Command = "/usr/bin/myothercollector --foo=bar"