WEBVTT 00:04.270 --> 00:08.770 Now, let's get started, though, with consumer configuration, and for that, we will create new glass 00:08.770 --> 00:16.540 in our product, please navigate to SLC Main and Java and change our folder, create new glass and let's 00:16.780 --> 00:18.190 give a name to this glass. 00:18.190 --> 00:20.010 Simply a consumer like this. 00:20.350 --> 00:21.420 Let's create the glass. 00:21.430 --> 00:24.820 And here let's create quickly main method for that. 00:24.820 --> 00:27.640 Simply diamine and press that wonderful. 00:27.760 --> 00:33.060 And now let's create basic consumer that will consume messages from particular topic. 00:33.070 --> 00:37.200 And at the moment we have actually two topics strings and the numbers. 00:37.210 --> 00:41.500 And we're able to get messages from any of those topics before doing that. 00:41.530 --> 00:47.210 Let's get back to the communication and find the how are you actually able to create a consumer? 00:47.470 --> 00:50.920 Please go to capture dot org documentation back again. 00:51.310 --> 00:58.860 Scroll down to Consumer API and let's observe what we need in order to start the consumer with Java. 00:59.140 --> 01:06.970 And as you see here, we need to use the same Mavin dependency as we actually use them in producer API. 01:07.120 --> 01:10.750 It is coffeecake lines in Abargil, that Kafka package. 01:10.900 --> 01:12.450 So same package here. 01:12.730 --> 01:19.470 We have already imported it to our project and you can verify that here in form that XML file here was 01:19.480 --> 01:25.020 this dependency and that's the way we are good to go and we are able to start the creation of consumer. 01:25.480 --> 01:29.860 Let's get back to documentation and let's open up in a new type of this. 01:29.860 --> 01:36.520 Java docs lean open link in a new tab and here you'll see details for Kafka consumer class. 01:36.820 --> 01:42.250 Let's go down and find the example that we could start actually with. 01:42.520 --> 01:43.600 Let me scroll down. 01:43.600 --> 01:49.930 And here was a basic example with automatic offset committing, you know, from previous sections that 01:49.930 --> 01:57.670 the consumers commit messages that they are receiving and you are able to enable all to commit or you 01:57.670 --> 02:00.370 are able to commit the method just manually. 02:00.670 --> 02:07.420 And for example, if on a consumer or site you perform some actions with messages, for example, you 02:07.420 --> 02:14.950 install them into a database or send to another service or something else, and you need to get confirmation 02:14.950 --> 02:20.260 of that operation, for example, that you have successfully written messages into the database. 02:20.620 --> 02:27.190 Then you could enable manual committing and send confirmation to brokers only when all operations on 02:27.190 --> 02:32.890 consumer side were successfully completed, for example, when records were successfully returned to 02:32.890 --> 02:33.580 database. 02:34.000 --> 02:39.610 And with such manual, you may be sure that the old messages in the topic were successfully returned 02:39.610 --> 02:40.360 to database. 02:40.690 --> 02:46.450 But for now, in the first example, let's actually use all the competing and you could enable it to 02:46.780 --> 02:49.840 the appropriate to enable us to commit and set up to do. 02:50.080 --> 02:57.130 And also with the next set in the interval, you are able to specify how often the consumer will send 02:57.130 --> 02:59.590 out to commit messages to brokers. 02:59.740 --> 03:04.330 And in this example, they will be sent every second one thousand milliseconds. 03:04.720 --> 03:08.090 Also, we need to use the key and while you're there, serializable. 03:08.350 --> 03:15.970 And before I have explained you that every producer actually uses a specific serializer and Dinara example, 03:15.970 --> 03:23.380 we use the students realize our FAKIE and Waialua, and that means that on consumer side you need to 03:23.380 --> 03:25.870 use corresponding their serializer. 03:25.870 --> 03:32.560 If you want to read messages from a specific topic where this producer has produced messages to and 03:32.560 --> 03:37.510 in this case, we need to use corresponding serializer for strings. 03:37.720 --> 03:41.830 And they are Goldust doing the serializer like this. 03:42.250 --> 03:47.230 And also, of course, in order to create the consumer, you need to specify bootstrap severals. 03:47.410 --> 03:49.450 And this option is the same. 03:49.450 --> 03:54.550 As for producers, we need simply list all brokers in the class. 03:54.940 --> 03:59.080 Of course, again, you are able to specify just a single several like in this example. 03:59.260 --> 04:03.430 But it's not recommended if you have an entire class or with multiple zeros. 04:03.550 --> 04:09.340 In such case, you should specify all zeroth in a class for redundancy grade. 04:09.340 --> 04:15.940 And afterwards, after setting properties, you need to create a consumer from a consumer class and 04:15.940 --> 04:17.180 to pass your properties. 04:17.350 --> 04:23.980 Similarly, we have created the coproducer like that and afterwards we will subscribe to specific topic. 04:24.100 --> 04:29.330 In this example, you'll see that we subscribe to a couple of topics and borrow. 04:29.380 --> 04:35.460 And basically in our example, we will subscribe just to one topic and afterwards we start infinite 04:35.470 --> 04:42.340 the while loop and pull consumer and try to get new messages every 100 milliseconds. 04:42.520 --> 04:46.930 And if some messages are received in this example, we bring them to Tamino. 04:47.170 --> 04:53.260 Please notice also that in this example there is group ID parameter at present and it is set to test 04:53.260 --> 04:56.170 itself actually consumer group ID. 04:56.320 --> 05:02.560 We will try to create consumer without group ID, but afterwards you'll see that the group ID is actually 05:02.560 --> 05:03.310 required and. 05:03.390 --> 05:10.230 We will set it like this great, let's go to our consumer configuration and let's create properties. 05:11.510 --> 05:12.260 Properties. 05:13.260 --> 05:13.800 Brewpubs. 05:14.670 --> 05:18.350 Equals sign new properties and next. 05:18.430 --> 05:24.780 Let's set up a broker or at least give you the serializer and enable us to commit with specific how 05:24.780 --> 05:25.760 to commit inter-modal. 05:26.070 --> 05:32.190 And let me actually go to producer and copy those three lines from here like this. 05:33.030 --> 05:37.020 Go here to consumer based this property properties here. 05:37.620 --> 05:39.920 This first line will remain unchanged. 05:39.930 --> 05:41.850 Same list of brokers. 05:42.090 --> 05:48.680 And here we need to replace Serializer with their serializer like this for key and value. 05:48.900 --> 05:55.340 And also here we will use their serializer and here we'll be doing the serializer as well. 05:55.800 --> 05:56.200 Great. 05:56.340 --> 05:58.560 And afterwards, let's enable us to commit. 05:58.560 --> 06:04.620 And let me actually go here to this example and copy those two properties from here in order to speed 06:04.620 --> 06:05.430 up this process. 06:05.640 --> 06:11.580 And actually, in example, set property method is used instead of let me replace it with put here and 06:12.000 --> 06:13.050 put here as well. 06:13.450 --> 06:13.860 Great. 06:14.070 --> 06:16.920 Here we have all necessary properties for this first example. 06:17.160 --> 06:22.350 And next, let's declare our topic variable like this and set it equal to Lombroso. 06:22.350 --> 06:24.960 We were the first read from no storbeck like this. 06:25.320 --> 06:28.050 And afterwards, let's create new consumer. 06:28.970 --> 06:36.410 COFCO consumer here will be the types of parameters to industry, let's give a name to our consumer 06:36.410 --> 06:43.790 or simply consumer, and let's create an instance of consumer class and health properties. 06:43.790 --> 06:44.960 Variable as argument. 06:45.200 --> 06:50.480 Great, our consumer was created and the next DELYTH subscribe to one of the topics and we have specified 06:50.480 --> 06:52.030 already topic here. 06:52.340 --> 07:00.020 And for that let's use consumer dot subscribe method and here as argument, let's pass formatted as 07:00.020 --> 07:01.700 least array of topics. 07:01.910 --> 07:06.740 And for that let's use a method of iRace class with type arrays. 07:06.920 --> 07:09.440 Don't forget to press enter in order to import it. 07:09.680 --> 07:13.880 And afterwards, let's use this method as least like this. 07:14.120 --> 07:17.840 And here let's pass our topic as argument. 07:18.170 --> 07:23.990 Actually, instead of a single topic, let's adjust to variable name to topics here and make it as array 07:23.990 --> 07:25.160 of strings like this. 07:25.400 --> 07:28.790 And here let's create an array of strings like soap. 07:29.090 --> 07:34.100 And with such configuration in place, you will be able to add additional topics here to this array 07:34.100 --> 07:36.460 if you want to create afterwards. 07:36.480 --> 07:43.400 Let's actually start our infinite loop and let's start consumption from the topic specified here in 07:43.400 --> 07:44.740 this topic array. 07:44.990 --> 07:47.690 Let's not make a quick break and go on after the balls.