1"use strict";(self.webpackChunkmia_platform_docs=self.webpackChunkmia_platform_docs||[]).push([["14760"],{157650(e,r,s){s.r(r),s.d(r,{metadata:()=>a,default:()=>h,frontMatter:()=>i,contentTitle:()=>l,toc:()=>d,assets:()=>o});var a=JSON.parse('{"id":"runtime-components/libraries/kafka2rest/overview","title":"Kafka2Rest Kotlin Library","description":"This library allow to consume kafka messages and handle them making HTTP REST calls.","source":"@site/versioned_docs/version-15.0.1/runtime-components/libraries/kafka2rest/10_overview.md","sourceDirName":"runtime-components/libraries/kafka2rest","slug":"/runtime-components/libraries/kafka2rest/overview","permalink":"/docs/15.0.1/runtime-components/libraries/kafka2rest/overview","draft":false,"unlisted":false,"tags":[],"version":"15.0.1","sidebarPosition":10,"frontMatter":{"id":"overview","title":"Kafka2Rest Kotlin Library","sidebar_label":"Kafka2Rest"},"sidebar":"marketplace","previous":{"title":"CHANGELOG","permalink":"/docs/15.0.1/runtime-components/libraries/kafka-healthchecker/changelog"},"next":{"title":"CHANGELOG","permalink":"/docs/15.0.1/runtime-components/libraries/lc39/changelog"}}'),n=s(474848),t=s(28453);let i={id:"overview",title:"Kafka2Rest Kotlin Library",sidebar_label:"Kafka2Rest"},l,o={},d=[{value:"Install",id:"install",level:2},{value:"install Kafka2Rest",id:"install-kafka2rest",level:3},{value:"Maven",id:"maven",level:4},{value:"Gradle",id:"gradle",level:4},{value:"install Kafka2Rest dependencies",id:"install-kafka2rest-dependencies",level:3},{value:"Gradle",id:"gradle-1",level:4},{value:"Usage",id:"usage",level:2},{value:"Initialization",id:"initialization",level:3},{value:"Config Parameters",id:"config-parameters",level:4},{value:"Code Sample",id:"code-sample",level:4},{value:"Messages Handling",id:"messages-handling",level:3},{value:"Start",id:"start",level:3},{value:"Health Check",id:"health-check",level:3},{value:"ProducerManager (beta)",id:"producermanager-beta",level:2}];function c(e){let r={a:"a",code:"code",em:"em",h2:"h2",h3:"h3",h4:"h4",li:"li",p:"p",pre:"pre",strong:"strong",table:"table",tbody:"tbody",td:"td",th:"th",thead:"thead",tr:"tr",ul:"ul",...(0,t.R)(),...e.components};return(0,n.jsxs)(n.Fragment,{children:[(0,n.jsx)(r.p,{children:"This library allow to consume
1kafka messages and handle them making HTTP REST calls."}),"\n",(0,n.jsx)(r.h2,{id:"install",children:"Install"}),"\n",(0,n.jsx)(r.p,{children:"To include the library as a dependency of your Maven/Gradle project use the following snippets."}),"\n",(0,n.jsx)(r.h3,{id:"install-kafka2rest",children:"install Kafka2Rest"}),"\n",(0,n.jsx)(r.h4,{id:"maven",children:"Maven"}),"\n",(0,n.jsx)(r.pre,{children:(0,n.jsx)(r.code,{className:"language-xml",children:"<dependency>\n <groupId>eu.mia-platform</groupId>\n <artifactId>kafka2rest</artifactId>\n <version>{VERSION}</version>\n</dependency>\n"})}),"\n",(0,n.jsx)(r.h4,{id:"gradle",children:"Gradle"}),"\n",(0,n.jsx)(r.pre,{children:(0,n.jsx)(r.code,{className:"language-java",children:"implementation 'eu.mia-platform:kafka2rest:{VERSION}'\n"})}),"\n",(0,n.jsx)(r.h3,{id:"install-kafka2rest-dependencies",children:"install Kafka2Rest dependencies"}),"\n",(0,n.jsxs)(r.p,{children:["You need to add ",(0,n.jsx)(r.a,{href:"https://github.com/kittinunf/fuel",children:"Fuel"})," repository to your ",(0,n.jsx)(r.code,{children:"build.gradle"})," or ",(0,n.jsx)(r.code,{children:"pom.xml"}),"."]}),"\n",(0,n.jsx)(r.h4,{id:"gradle-1",children:"Gradle"}),"\n",(0,n.jsx)(r.pre,{children:(0,n.jsx)(r.code,{className:"language-java",children:"repositories {\n maven {\n name 'spring-lib-release'\n url 'https://repo.spring.io/libs-release/'\n }\n ...\n}\n"})}),"\n",(0,n.jsx)(r.h2,{id:"usage",children:"Usage"}),"\n",(0,n.jsxs)(r.p,{children:["Kafka2Rest is based on ",(0,n.jsx)(r.strong,{children:"Filters"})," and ",(0,n.jsx)(r.strong,{children:"Processors"}),". You can develop your custom Filters and Processor implementing the ",(0,n.jsx)(r.a,{href:"https://git.tools.mia-platform.eu/platform/libraries/kafka2rest/blob/master/src/main/kotlin/eu/miaplatform/kafka2rest/interfaces/FilterInterface.kt",children:"FilterInterface"})," and ",(0,n.jsx)(r.a,{href:"https://git.tools.mia-platform.eu/platform/libraries/kafka2rest/blob/master/src/main/kotlin/eu/miaplatform/kafka2rest/interfaces/ProcessorInterface.kt",children:"ProcessorInterface"})," or use existent ",(0,n.jsx)(r.a,{href:"https://git.tools.mia-platform.eu/platform/libraries/kafka2rest/tree/master/src/main/kotlin/eu/miaplatform/kafka2rest/filters",children:"filters"})," and ",(0,n.jsx)(r.a,{href:"https://git.tools.mia-platform.eu/platform/libraries/kafka2rest/tree/master/src/main/kotlin/eu/miaplatform/kafka2rest/processors",children:"processors"}),"."]}),"\n",(0,n.jsx)(r.h3,{id:"initialization",children:"Initialization"}),"\n",(0,n.jsx)(r.p,{children:"First of all you need to initialize Kafka2Rest with:"}),"\n",(0,n.jsxs)(r.ul,{children:["\n",(0,n.jsxs)(r.li,{children:["Generics ",(0,n.jsx)(r.code,{children:"<K, V>"})," used in library ",(0,n.jsx)(r.a,{href:"https://kafka.apache.org/11/javadoc/org/apache/kafka/clients/consumer/ConsumerRecords.html",children:"ConsumerRecords"}),"."]}),"\n",(0,n.jsxs)(r.li,{children:[(0,n.jsx)(r.a,{href:"https://kafka.apache.org/20/javadoc/org/apache/kafka/clients/consumer/Consumer.html",children:"Kafka Consumer"}),'. Remember to set false "enable.auto.commit" property if you need to handle processors failures.']}),"\n",(0,n.jsxs)(r.li,{children:["A ",(0,n.jsx)(r.a,{href:"https://git.tools.mia-platform.eu/platform/libraries/kafka2rest/blob/master/src/main/kotlin/eu/miaplatform/kafka2rest/config/ConfigBuilder.kt",children:"Config"})," object."]}),"\n",(0,n.jsxs)(r.li,{children:[(0,n.jsx)(r.a,{href:"https://kafka.apache.org/20/javadoc/org/apache/kafka/clients/producer/Producer.html",children:"Kafka Producer (Optional)"}),". This parameter must be set in order to use the ",(0,n.jsx)(r.code,{children:"onExceptionTopic"})," configuration."]}),"\n"]}),"\n",(0,n.jsx)(r.h4,{id:"config-parameters",children:"Config Parameters"}),"\n",(0,n.jsxs)(r.table,{children:[(0,n.jsx)(r.thead,{children:(0,n.jsxs)(r.tr,{children:[(0,n.jsx)(r.th,{children:"Parameter"}),(0,n.jsx)(r.th,{children:"Meaning"})]})}),(0,n.jsxs)(r.tbody,{children:[(0,n.jsxs)(r.tr,{children:[(0,n.jsx)(r.td,{children:"maxIntervalBetweenPolls"}),(0,n.jsxs)(r.td,{children:["The maximum time (Duration) interval between two consecutive poll. When this timeout is exceeded the method ",(0,n.jsx)(r.code,{children:"isRunning()"})," returns false."]})]}),(0,n.jsxs)(r.tr,{children:[(0,n.jsx)(r.td,{children:"topicList"}),(0,n.jsx)(r.td,{children:"The list of topic to subscribe to."})]}),(0,n.jsxs)(r.tr,{children:[(0,n.jsx)(r.td,{children:"pollTimeout"}),(0,n.jsx)(r.td,{children:"The time (Duration) spent waiting in poll if data is not available in the buffer. If 0, returns immediately with any records that are available currently in the buffer, else returns empty. Must not be negative."})]}),(0,n.jsxs)(r.tr,{children:[(0,n.jsx)(r.td,{children:"onExceptionTopic"}),(0,n.jsxs)(r.td,{children:["(Optional) Defines a topic where messages causing unhandled processor exceptions will be stored. Kafka2Rest will enrich the message headers with two properties ",(0,n.jsx)(r.code,{children:"retryNumber"})," and ",(0,n.jsx)(r.code,{children:"errors"}),"."]})]}),(0,n.jsxs)(r.tr,{children:[(0,n.jsx)(r.td,{children:"retriesLimit"}),(0,n.jsx)(r.td,{})]}),(0,n.jsxs)(r.tr,{children:[(0,n.jsx)(r.td,{children:"sleepPeriod"}),(0,n.jsx)(r.td,{})]})]})]}),"\n",(0,n.jsx)(r.h4,{id:"code-sample",children:"Code Sample"}),"\n",(0,n.jsx)(r.pre,{children:(0,n.jsx)(r.code,{className:"language-kotlin",children:' val props = Properties()\n props["bootstrap.servers"] = "http://your-kafka-host:9092"\n props["group.id"] = "consumer-group-id"\n props["key.deserializer"] = StringDeserializer::class.java\n props["value.deserializer"] = StringDeserializer::class.java\n props["max.poll.interval.ms"] = Int.MAX_VALUE\n props["connections.max.idle.ms"] = Int.MAX_VALUE\n props["enable.auto.commit"] = false\n\n val consumer = KafkaConsumer<String, String>(props)\n\n val config = ConfigBuilder\n .addTopicList(listOf("topic1", "topic2"))\n .addPollTimeout(Duration.ofSeconds(10L))\n .addMaxIntervalBetweenPolls(Duration.ofMinutes(5L))\n .build()\n\n val kafka2rest = Kafka2Rest<String, String>(consumer, config)\n'})}),"\n",(0,n.jsx)(r.h3,{id:"messages-handling",children:"Messages Handling"}),"\n",(0,n.jsxs)(r.p,{children:["Then you need to set the Filter And Processor couples. Filter parse kafka message and return ",(0,n.jsx)(r.em,{children:"boolean"})," .\nIf true the linked processor run its ",(0,n.jsx)(r.code,{children:"process()"})," method and do something."]}),"\n",(0,n.jsx)(r.pre,{children:(0,n.jsx)(r.code,{className:"language-kotlin",children:" kafka2rest.set(Filter1(), HTTPPostProcessor())\n kafka2rest.set(Filter1(), MyCustomProcessor())\n kafka2rest.set(Filter2(), OtherProcessor())\n // ...\n"})}),"\n",(0,n.jsx)(r.h3,{id:"start",children:"Start"}),"\n",(0,n.jsxs)(r.p,{children:["Just start Kafka2Rest. Invoking the ",(0,n.jsx)(r.code,{children:"start()"}
1)," method Kafka2Rest will run on new thread and listen kafka topics."]}),"\n",(0,n.jsx)(r.pre,{children:(0,n.jsx)(r.code,{className:"language-kotlin",children:" kafka2rest.start()\n"})}),"\n",(0,n.jsx)(r.h3,{id:"health-check",children:"Health Check"}),"\n",(0,n.jsx)(r.p,{children:"Kafka2rest offers two method to verify service and connection healthiness."}),"\n",(0,n.jsxs)(r.ul,{children:["\n",(0,n.jsxs)(r.li,{children:[(0,n.jsx)(r.strong,{children:"isRunning()"})," This method verifies the timeout between two consecutive poll. It returns ",(0,n.jsx)(r.code,{children:"true"})," when the config ",(0,n.jsx)(r.code,{children:"maxIntervalBetweenPolls"})," timeout is respected. This ensures that the consumer is polling and the processors aren't getting stuck."]}),"\n",(0,n.jsxs)(r.li,{children:[(0,n.jsx)(r.strong,{children:"isConnected()"})," This method verifies that consumer can list broker topics and it must be connected to kafka server to do this. It returns ",(0,n.jsx)(r.code,{children:"true"})," when topics are more than 0."]}),"\n"]}),"\n",(0,n.jsx)(r.h2,{id:"producermanager-beta",children:"ProducerManager (beta)"}),"\n",(0,n.jsxs)(r.p,{children:["You can also send messages to kafka host building ProducerManager and using the ",(0,n.jsx)(r.code,{children:"sendMessage()"})," method."]}),"\n",(0,n.jsx)(r.pre,{children:(0,n.jsx)(r.code,{className:"language-kotlin",children:' val producerManager = ProducerManager("localhost:9092")\n producerManager.sendMessage(myKafkaMessage, "topic1")\n'})})]})}function h(e={}){let{wrapper:r}={...(0,t.R)(),...e.components};return r?(0,n.jsx)(r,{...e,children:(0,n.jsx)(c,{...e})}):c(e)}},28453(e,r,s){s.d(r,{R:()=>i,x:()=>l});var a=s(296540);let n={},t=a.createContext(n);function i(e){let r=a.useContext(t);return a.useMemo(function(){return"function"==typeof e?e(r):{...r,...e}},[r,e])}function l(e){let r;return r=e.disableParentContext?"function"==typeof e.components?e.components(n):e.components||n:i(e.components),a.createElement(t.Provider,{value:r},e.children)}}}]);
Line numbers count LF bytes from the start of the resource, as the search results do. Vendor segments are library code the classifier recognised; they are stored but not indexed. Bytes are shown as Latin1 characters, one per byte.