Add a cli option for producer encoding

go.mod

@@ -5,7 +5,7 @@ go 1.15
 require (
 	github.com/Financial-Times/go-fthealth v0.0.0-20171204124831-1b007e2b37b7
 	github.com/Financial-Times/go-logger v0.0.0-20170914081945-83fc3e64dc55
-	github.com/Financial-Times/message-queue-go-producer v0.1.1-0.20170622111849-0bb065111416
+	github.com/Financial-Times/message-queue-go-producer v0.1.1-0.20210805140503-941e89a60d7f
 	github.com/Financial-Times/message-queue-gonsumer v0.0.0-20180518165041-cd41937c7566
 	github.com/Financial-Times/service-status-go v0.0.0-20160323111542-3f5199736a3d
 	github.com/dchest/uniuri v0.0.0-20160212164326-8902c56451e9
@@ -14,7 +14,7 @@ require (
 	github.com/onsi/ginkgo v1.14.2 // indirect
 	github.com/onsi/gomega v1.10.3 // indirect
 	github.com/sirupsen/logrus v1.0.4-0.20170822132746-89742aefa4b2 // indirect
-	github.com/stretchr/testify v1.3.0
+	github.com/stretchr/testify v1.7.0
 	gopkg.in/airbrake/gobrake.v2 v2.0.9 // indirect
 	gopkg.in/gemnasium/logrus-airbrake-hook.v2 v2.1.2 // indirect
 )

go.sum

@@ -1,14 +1,31 @@
 github.com/Financial-Times/go-fthealth v0.0.0-20171204124831-1b007e2b37b7 h1:dkf1EOTiHXA2lG2EJuePEim6y0HEOPt0hcqsT/qUr/k=
 github.com/Financial-Times/go-fthealth v0.0.0-20171204124831-1b007e2b37b7/go.mod h1:gpAzq6W5rCheYlY32JOIxS/VjVcYHbC2PkMzQngHT9c=
+github.com/Financial-Times/go-fthealth v0.0.0-20180807113633-3d8eb430d5b5 h1:XH5h45aAyG1bAFBYmkgJkT4q13CbkCJ+gj9+rIfzuL8=
+github.com/Financial-Times/go-fthealth v0.0.0-20180807113633-3d8eb430d5b5/go.mod h1:gpAzq6W5rCheYlY32JOIxS/VjVcYHbC2PkMzQngHT9c=
 github.com/Financial-Times/go-logger v0.0.0-20170914081945-83fc3e64dc55 h1:+/OZNdWdTPM7A7JxR/8ul1QZgXYfPxJdo6iIcK3K17I=
 github.com/Financial-Times/go-logger v0.0.0-20170914081945-83fc3e64dc55/go.mod h1:NI4Dg39A21H57YC2nG8C42C6ENz/YVsI0jMQWngJzR0=
-github.com/Financial-Times/message-queue-go-producer v0.1.1-0.20170622111849-0bb065111416 h1:7qa+AbW0xYSUhqzUtWgYN9JFva9CqsKy1A0lNs3WpYI=
-github.com/Financial-Times/message-queue-go-producer v0.1.1-0.20170622111849-0bb065111416/go.mod h1:1mfkRkBPglgCeE8w0gzoC4ujulQAqWfMKVCIdR+2svQ=
+github.com/Financial-Times/go-logger v0.0.0-20180323124113-febee6537e90 h1:U7wPaeMESlG0WVwOobaw4qv6I6s9F8b0SdmJKH3Vh6A=
+github.com/Financial-Times/go-logger v0.0.0-20180323124113-febee6537e90/go.mod h1:NI4Dg39A21H57YC2nG8C42C6ENz/YVsI0jMQWngJzR0=
+github.com/Financial-Times/http-handlers-go v0.0.0-20170809121007-229ac16f1d9e h1:/Y2wrSfkueFmdOIyQSABebfEe5P+yFyxBnmtnx1C0HM=
+github.com/Financial-Times/http-handlers-go v0.0.0-20170809121007-229ac16f1d9e/go.mod h1:sAkXv1oPYgNTYBYsYs83HwpYp7R50mvgBGGcsOlJtOw=
+github.com/Financial-Times/message-queue-go-producer v0.1.1-0.20210727141350-ffe1f42420fa h1:ar/IMEcnU7otfDH/pLM8pbj2ImNWJIAFWOJiaD/ozek=
+github.com/Financial-Times/message-queue-go-producer v0.1.1-0.20210727141350-ffe1f42420fa/go.mod h1:O+R/PyICNDU76FSWeusXl/crkwUBo1X6J9Te8Q6fQ0I=
+github.com/Financial-Times/message-queue-go-producer v0.1.1-0.20210728114901-dd89c7140d11 h1:IaB/Y4snlcLeyHx5pCfNQq9HhH+hx3wMDtFFsmFY6jQ=
+github.com/Financial-Times/message-queue-go-producer v0.1.1-0.20210728114901-dd89c7140d11/go.mod h1:O+R/PyICNDU76FSWeusXl/crkwUBo1X6J9Te8Q6fQ0I=
+github.com/Financial-Times/message-queue-go-producer v0.1.1-0.20210728120155-bd0f255f821a h1:6hmJtu00PKdlpERIGHPJ4BgzcAOqrITw4SUmVV4ROrg=
+github.com/Financial-Times/message-queue-go-producer v0.1.1-0.20210728120155-bd0f255f821a/go.mod h1:O+R/PyICNDU76FSWeusXl/crkwUBo1X6J9Te8Q6fQ0I=
+github.com/Financial-Times/message-queue-go-producer v0.1.1-0.20210805085551-b267a0f9582d h1:H4YIluCZesW3nUwW1bGpcg76dlFHR8c+umFvulnsa8Y=
+github.com/Financial-Times/message-queue-go-producer v0.1.1-0.20210805085551-b267a0f9582d/go.mod h1:O+R/PyICNDU76FSWeusXl/crkwUBo1X6J9Te8Q6fQ0I=
+github.com/Financial-Times/message-queue-go-producer v0.1.1-0.20210805140503-941e89a60d7f h1:XudPZs2dqW7idLZ3wQcsPTzUP8WqoLK2CblQ/w8MHN4=
+github.com/Financial-Times/message-queue-go-producer v0.1.1-0.20210805140503-941e89a60d7f/go.mod h1:f0BOegIQhC4zQyokEaH40dEQ8CG5I3qzalmWz7uWIPA=
 github.com/Financial-Times/message-queue-gonsumer v0.0.0-20180518165041-cd41937c7566 h1:RZ8LXUuj6nDvGoXKF5m+TmHlR0PxG8hU85PLHV8GRBE=
 github.com/Financial-Times/message-queue-gonsumer v0.0.0-20180518165041-cd41937c7566/go.mod h1:A88i3psx3Zm80Ai2OYTrwzKkZGKj+x5KL02z+YrRd10=
+github.com/Financial-Times/post-publication-combiner/v2 v2.0.0-20200429053931-78b45d72d0ff h1:1CcRc4KcKU5ouy9VCoRclnX0+wBSUqtiU6C/uyv9zag=
+github.com/Financial-Times/post-publication-combiner/v2 v2.0.0-20200429053931-78b45d72d0ff/go.mod h1:AH/4UbHiDqRs5OtvVlD0dJSXuxzLBwTa2CVqEQsmMUg=
 github.com/Financial-Times/service-status-go v0.0.0-20160323111542-3f5199736a3d h1:USNBTIof6vWGM49SYrxvC5Y8NqyDL3YuuYmID81ORZQ=
 github.com/Financial-Times/service-status-go v0.0.0-20160323111542-3f5199736a3d/go.mod h1:7zULC9rrq6KxFkpB3Y5zNVaEwrf1g2m3dvXJBPDXyvM=
-github.com/davecgh/go-spew v1.1.0 h1:ZDRjVQ15GmhC3fiQ8ni8+OwkZQO4DARzQgrnXU1Liz8=
+github.com/Financial-Times/transactionid-utils-go v0.2.0 h1:YcET5Hd1fUGWWpQSVszYUlAc15ca8tmjRetUuQKRqEQ=
+github.com/Financial-Times/transactionid-utils-go v0.2.0/go.mod h1:tPAcAFs/dR6Q7hBDGNyUyixHRvg/n9NW/JTq8C58oZ0=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
@@ -29,29 +46,54 @@ github.com/google/go-cmp v0.3.0/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMyw
 github.com/google/go-cmp v0.3.1/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMywk6iLU=
 github.com/google/go-cmp v0.4.0 h1:xsAVV57WRhGj6kEIi8ReJzQlHHqcBYCElAvkovg3B/4=
 github.com/google/go-cmp v0.4.0/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
+github.com/google/uuid v1.3.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
+github.com/gorilla/context v1.1.1 h1:AWwleXJkX/nhcU9bZSnZoi3h/qGYqQAGhq6zZe/aQW8=
+github.com/gorilla/context v1.1.1/go.mod h1:kBGZzfjB9CEq2AlWe17Uuf7NDRt0dE0s8S51q0aT7Yg=
+github.com/gorilla/handlers v1.4.0 h1:XulKRWSQK5uChr4pEgSE4Tc/OcmnU9GJuSwdog/tZsA=
+github.com/gorilla/handlers v1.4.0/go.mod h1:Qkdc/uu4tH4g6mTK6auzZ766c4CA0Ng8+o/OAirnOIQ=
+github.com/gorilla/mux v1.4.1-0.20170830053917-a659b61323b0 h1:WufQb+4501Pn15bGwgA1eE6QREDVyecaTILO3GJv/UQ=
+github.com/gorilla/mux v1.4.1-0.20170830053917-a659b61323b0/go.mod h1:1lud6UwP+6orDFRuTfBEV8e9/aOM/c4fVVCaMa2zaAs=
 github.com/hashicorp/go-version v0.0.0-20170202080759-03c5bf6be031 h1:c3Xdf5fTpk+hqhxqCO+ymqjfUXV9+GZqNgTtlnVzDos=
 github.com/hashicorp/go-version v0.0.0-20170202080759-03c5bf6be031/go.mod h1:fltr4n8CU8Ke44wwGCBoEymUuxUHl09ZGVZPK5anwXA=
+github.com/hashicorp/go-version v1.0.0 h1:21MVWPKDphxa7ineQQTrCU5brh7OuVVAzGOCnnCPtE8=
+github.com/hashicorp/go-version v1.0.0/go.mod h1:fltr4n8CU8Ke44wwGCBoEymUuxUHl09ZGVZPK5anwXA=
+github.com/hpcloud/tail v1.0.0 h1:nfCOvKYfkgYP8hkirhJocXT2+zOD8yUNjXaWfTlyFKI=
 github.com/hpcloud/tail v1.0.0/go.mod h1:ab1qPbhIpdTxEkNHXyeSf5vhxWSCs/tWer42PpOxQnU=
+github.com/jawher/mow.cli v1.0.4/go.mod h1:5hQj2V8g+qYmLUVWqu4Wuja1pI57M83EChYLVZ0sMKk=
 github.com/jawher/mow.cli v1.1.0 h1:NdtHXRc0CwZQ507wMvQ/IS+Q3W3x2fycn973/b8Zuk8=
 github.com/jawher/mow.cli v1.1.0/go.mod h1:aNaQlc7ozF3vw6IJ2dHjp2ZFiA4ozMIYY6PyuRJwlUg=
 github.com/nxadm/tail v1.4.4 h1:DQuhQpB1tVlglWS2hLQ5OV6B5r8aGxSrPc5Qo6uTN78=
 github.com/nxadm/tail v1.4.4/go.mod h1:kenIhsEOeOJmVchQTgglprH7qJGnHDVpk1VPCcaMI8A=
 github.com/onsi/ginkgo v1.6.0/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE=
+github.com/onsi/ginkgo v1.12.0/go.mod h1:oUhWkIvk5aDxtKvDDuw8gItl8pKl42LzjC9KZE0HfGg=
 github.com/onsi/ginkgo v1.12.1/go.mod h1:zj2OWP4+oCPe1qIXoGWkgMRwljMUYCdkwsT2108oapk=
 github.com/onsi/ginkgo v1.14.2 h1:8mVmC9kjFFmA8H4pKMUhcblgifdkOIXPvbhN1T36q1M=
 github.com/onsi/ginkgo v1.14.2/go.mod h1:iSB4RoI2tjJc9BBv4NKIKWKya62Rps+oPG/Lv9klQyY=
 github.com/onsi/gomega v1.7.1/go.mod h1:XdKZgCCFLUoM/7CFJVPcG8C1xQ1AJ0vpAezJrB7JYyY=
+github.com/onsi/gomega v1.9.0/go.mod h1:Ho0h+IUsWyvy1OpqCwxlQ/21gkhVunqlU8fDGcoTdcA=
 github.com/onsi/gomega v1.10.1/go.mod h1:iN09h71vgCQne3DLsj+A5owkum+a2tYe+TOCB1ybHNo=
 github.com/onsi/gomega v1.10.3 h1:gph6h/qe9GSUw1NhH1gp+qb+h8rXD8Cy60Z32Qw3ELA=
 github.com/onsi/gomega v1.10.3/go.mod h1:V9xEwhxec5O8UDM77eCW8vLymOMltsqPVYWrpDsH8xc=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/rcrowley/go-metrics v0.0.0-20161128210544-1f30fe9094a5 h1:gwcdIpH6NU2iF8CmcqD+CP6+1CkRBOhHaPR+iu6raBY=
+github.com/rcrowley/go-metrics v0.0.0-20161128210544-1f30fe9094a5/go.mod h1:bCqnVzQkZxMG4s8nGwiZ5l3QUCyqpo9Y+/ZMZ9VjZe4=
+github.com/satori/go.uuid v1.2.1-0.20181016170032-d91630c85102 h1:WAQaHPfnpevd8SKXCcy5nk3JzEv2h5Q0kSwvoMqXiZs=
+github.com/satori/go.uuid v1.2.1-0.20181016170032-d91630c85102/go.mod h1:dA0hQrYB0VpLJoorglMZABFdXlWrHn1NEOzdhQKdks0=
 github.com/sirupsen/logrus v1.0.4-0.20170822132746-89742aefa4b2 h1:a07zp0wovcAE2jH+wlD22JLqUH6Rdl8Aon+NiyPxE+0=
 github.com/sirupsen/logrus v1.0.4-0.20170822132746-89742aefa4b2/go.mod h1:pMByvHTf9Beacp5x1UXfOR9xyW/9antXMhjMPG0dEzc=
+github.com/stathat/go v1.0.0 h1:HFIS5YkyaI6tXu7JXIRRZBLRvYstdNZm034zcCeaybI=
+github.com/stathat/go v1.0.0/go.mod h1:+9Eg2szqkcOGWv6gfheJmBBsmq9Qf5KDbzy8/aYYR0c=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
+github.com/stretchr/objx v0.2.0 h1:Hbg2NidpLE8veEBkEZTL3CvlkUIVzuU9jDplZO54c48=
 github.com/stretchr/objx v0.2.0/go.mod h1:qt09Ya8vawLte6SNmTgCsAVtYtaKzEcn8ATUoHMkEqE=
+github.com/stretchr/testify v1.1.5-0.20170130113145-4d4bfba8f1d1/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs=
+github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs=
 github.com/stretchr/testify v1.3.0 h1:TivCn/peBQ7UY8ooIcPgZFpTNSz0Q2U6UrFlUfqbe0Q=
 github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
+github.com/stretchr/testify v1.7.0 h1:nwc3DEeHmmLAfoZucVR881uASk0Mfjw8xYJ99tb5CcY=
+github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
+golang.org/x/crypto v0.0.0-20180904163835-0709b304e793/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
 golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9 h1:psW17arqaxU48Z5kZ0CQnkZWQJsqcURM6tKiBApRjXI=
 golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
@@ -60,6 +102,7 @@ golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn
 golang.org/x/net v0.0.0-20200520004742-59133d7f0dd7/go.mod h1:qpuaurCH72eLCgpAm/N6yyVIVM9cpaDIP3A8BGJEC5A=
 golang.org/x/net v0.0.0-20201006153459-a7d1128ccaa0 h1:wBouT66WTYFXdxfVdz9sVWARVd/2vfGcmI45D2gj45M=
 golang.org/x/net v0.0.0-20201006153459-a7d1128ccaa0/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU=
+golang.org/x/sync v0.0.0-20180314180146-1d60e4601c6f h1:wMNYb4v58l5UBM7MYRLPG6ZhfOqbKu7X5eyFl8ZhKvA=
 golang.org/x/sync v0.0.0-20180314180146-1d60e4601c6f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sys v0.0.0-20180909124046-d0be0721c37e/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
@@ -75,7 +118,9 @@ golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.2/go.mod h1:bEr9sfX3Q8Zfm5fL9x+3itogRgK3+ptLWKqgva+5dAk=
 golang.org/x/text v0.3.3 h1:cokOdA+Jmi5PJGXLlLllQSgYigAEfHXJAERHVMaCc2k=
 golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
+golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e h1:FDhOuMEY4JVRztM/gsbk+IKUQ8kj74bxZrgw87eMMVc=
 golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
+golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543 h1:E7g+9GITq07hpfrRu66IVDexMakfv52eLZ2CXBWiKr4=
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 google.golang.org/protobuf v0.0.0-20200109180630-ec00e32a8dfd/go.mod h1:DFci5gLYBciE7Vtevhsrf46CRTquxDuWsQurQQe4oz8=
@@ -89,6 +134,7 @@ gopkg.in/airbrake/gobrake.v2 v2.0.9 h1:7z2uVWwn7oVeeugY1DtlPAy5H+KYgB1KeKTnqjNat
 gopkg.in/airbrake/gobrake.v2 v2.0.9/go.mod h1:/h5ZAUhDkGaJfjzjKLSjv6zCL6O0LLBxU4K+aSYdM/U=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/fsnotify.v1 v1.4.7 h1:xOHLXZwVvI9hhs+cLKq5+I5onOuwQLhQwiu63xxlHs4=
 gopkg.in/fsnotify.v1 v1.4.7/go.mod h1:Tz8NjZHkW78fSQdbUxIjBTcgA1z1m8ZHf0WmKUhAMys=
 gopkg.in/gemnasium/logrus-airbrake-hook.v2 v2.1.2 h1:OAj3g0cR6Dx/R07QgQe8wkA9RNjB2u4i700xBkIT4e0=
 gopkg.in/gemnasium/logrus-airbrake-hook.v2 v2.1.2/go.mod h1:Xk6kEKp8OKb+X14hQBKWaSkCsqBpgog8nAV2xsGOxlo=
@@ -97,3 +143,6 @@ gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7/go.mod h1:dt/ZhP58zS4L8KSrWD
 gopkg.in/yaml.v2 v2.2.4/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v2 v2.3.0 h1:clyUAQHOM3G0M3f5vQj7LuJrETvjVot3Z5el9nffUtU=
 gopkg.in/yaml.v2 v2.3.0/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
+gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
+gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b h1:h8qDotaEPuJATrMmW04NCwg7v22aHH28wwpauUhK9Oo=
+gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

healthchecks.go

@@ -35,7 +35,6 @@ func (hc HealthCheck) Health(serviceName string) func(w http.ResponseWriter, r *
 	if hc.producerType == proxy {
 		description = "Services: source-kafka-proxy, destination-kafka-proxy"
 		checks = []fthealth.Check{hc.consumeHealthcheck(), hc.proxyForwarderHealthcheck()}
-
 	}
 
 	healthCheck := fthealth.TimedHealthCheck{

healthchecks_test.go

@@ -30,7 +30,7 @@ func (p *mockProducerInstance) ConnectivityCheck() (string, error) {
 		return "", nil
 	}
 
-	return "", errors.New("Error connecting to the queue")
+	return "", errors.New("error connecting to the queue")
 }
 
 func (c *mockConsumerInstance) Start() {
@@ -44,7 +44,7 @@ func (c *mockConsumerInstance) ConnectivityCheck() (string, error) {
 		return "", nil
 	}
 
-	return "", errors.New("Error connecting to the queue")
+	return "", errors.New("error connecting to the queue")
 }
 
 func initializeHealthcheck(isProducerConnectionHealthy bool, isConsumerConnectionHealthy bool, producerType string) HealthCheck {
@@ -81,15 +81,15 @@ func TestGTGBrokenConsumer(t *testing.T) {
 
 	status := hc.GTG()
 	assert.False(t, status.GoodToGo)
-	assert.Equal(t, "Error connecting to the queue", status.Message)
+	assert.Equal(t, "error connecting to the queue", status.Message)
 }
 
 func TestGTGCheckBrokenProducer(t *testing.T) {
 	hc := initializeHealthcheck(false, true, proxy)
 
 	status := hc.GTG()
 	assert.False(t, status.GoodToGo)
-	assert.Equal(t, "Error connecting to the queue", status.Message)
+	assert.Equal(t, "error connecting to the queue", status.Message)
 }
 
 func TestHealthHappyFlow(t *testing.T) {

kafka-bridge.go

@@ -1,10 +1,12 @@
 package main
 
 import (
+	"errors"
 	"fmt"
 	"net"
 	"net/http"
 	"os"
+	"reflect"
 	"strings"
 	"time"
 
@@ -31,7 +33,7 @@ const (
 	proxy     = "proxy"
 )
 
-func newBridgeApp(consumerAddrs string, consumerGroupID string, consumerOffset string, consumerAutoCommitEnable bool, consumerAuthorizationKey string, topic string, producerAddress string, producerAuth string, producerType string, serviceName string) *BridgeApp {
+func newBridgeApp(consumerAddrs string, consumerGroupID string, consumerOffset string, consumerAutoCommitEnable bool, consumerAuthorizationKey string, topic string, producerAddress string, producerAuth string, producerType string, serviceName string, producerEncoding int) *BridgeApp {
 	consumerConfig := consumer.QueueConfig{}
 	consumerConfig.Addrs = strings.Split(consumerAddrs, ",")
 	consumerConfig.Group = consumerGroupID
@@ -48,11 +50,11 @@ func newBridgeApp(consumerAddrs string, consumerGroupID string, consumerOffset s
 	var producerInstance producer.MessageProducer
 	switch producerType {
 	case proxy:
-		producerInstance = producer.NewMessageProducer(producerConfig)
+		producerInstance = producer.NewMessageProducerWithEncoder(producerConfig, producer.NewEncoder(producerEncoding))
 	case plainHTTP:
 		producerInstance = newPlainHTTPMessageProducer(producerConfig)
 	default:
-		logger.Fatalf(nil, fmt.Errorf("Unknown producer type %s", producerType), "The provided producer type '%v' is invalid", producerType)
+		logger.Fatalf(nil, fmt.Errorf("unknown producer type %s", producerType), "The provided producer type '%v' is invalid", producerType)
 	}
 
 	httpClient := &http.Client{
@@ -154,13 +156,43 @@ func main() {
 	})
 
 	logger.InitDefaultLogger(*serviceName)
-	logger.Infof(nil, "Starting Kafka Bridge")
-
+	argument2Encoding := map[string]int{
+		"base64": producer.Base64E,
+		"model":  producer.CombinedModelE,
+	}
+	producerEncoding := app.String(cli.StringOpt{
+		Name:   "producer_encoding",
+		Value:  reflect.ValueOf(argument2Encoding).MapKeys()[0].String(),
+		Desc:   "Two possible values are accepted: model - if you want to send data encoded in a json mapping the model; or base64 if you want a base64 encoding.",
+		EnvVar: "PRODUCER_ENCODING",
+	})
 	app.Action = func() {
-		bridgeApp := newBridgeApp(*consumerAddrs, *consumerGroup, *consumerOffset, *consumerAutoCommitEnable, *consumerAuthorizationKey, *topic, *producerAddress, *producerAuth, *producerType, *serviceName)
+		if _, ok := argument2Encoding[*producerEncoding]; !ok {
+			var options []string
+			for k := range argument2Encoding {
+				options = append(options, k)
+			}
+			logger.Errorf(map[string]interface{}{"valid options": options}, errors.New("producer_encoding"), "invalid argument")
+			app.PrintHelp()
+			cli.Exit(-1)
+		}
+		bridgeApp := newBridgeApp(
+			*consumerAddrs,
+			*consumerGroup,
+			*consumerOffset,
+			*consumerAutoCommitEnable,
+			*consumerAuthorizationKey,
+			*topic,
+			*producerAddress,
+			*producerAuth,
+			*producerType,
+			*serviceName,
+			argument2Encoding[*producerEncoding],
+		)
 		go bridgeApp.enableHealthchecksAndGTG(bridgeApp.serviceName)
 		bridgeApp.consumeMessages()
 	}
+	logger.Infof(nil, "Starting Kafka Bridge")
 
 	err := app.Run(os.Args)
 	if err != nil {

message_consumer.go

@@ -12,10 +12,10 @@ import (
 	queueConsumer "github.com/Financial-Times/message-queue-gonsumer/consumer"
 )
 
-func (bridge BridgeApp) consumeMessages() {
-	consumerConfig := bridge.consumerConfig
+func (bridgeApp BridgeApp) consumeMessages() {
+	consumerConfig := bridgeApp.consumerConfig
 
-	consumer := queueConsumer.NewAgeingConsumer(*consumerConfig, bridge.forwardMsg, queueConsumer.AgeingClient{
+	consumer := queueConsumer.NewAgeingConsumer(*consumerConfig, bridgeApp.forwardMsg, queueConsumer.AgeingClient{
 		Client: &http.Client{
 			Timeout: 60 * time.Second,
 			Transport: &http.Transport{
@@ -36,7 +36,7 @@ func (bridge BridgeApp) consumeMessages() {
 		wg.Done()
 	}()
 
-	ch := make(chan os.Signal)
+	ch := make(chan os.Signal, 2)
 	signal.Notify(ch, syscall.SIGINT, syscall.SIGTERM)
 	<-ch
 	consumer.Stop()