Troubleshooting

Question 1

Initialize transaction

Answer 1

producer = new KafkaProducer<>(conf);

producer.initTransactions();

Question 2

Start transaction

Answer 2

producer.beginTransaction();

Question 3

Wrap up transaction

Answer 3

producer. sendOffsetsToTransaction(getConsumedOffsets(consumer), consumer.groupMetadata());
producer. commitTransaction();

Question 4

wait for async calculations to complete

Answer 4

Question 5

Print Stream in one instruction

Answer 5

stream.print(Printed.toSysOut());

Question 6

Consume events sent to topic. Create a tumbling window with a size of 10 seconds without a grace period. Count the records, and print the KTable to the standard output.

Answer 6

Question 7

Configure timeouts

Answer 7

props. put(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, 60000);
props. put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, 150);

Question 8

Configure idempotence

Answer 8

props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, false);

Question 9

Configure retries

Answer 9

props.put(ProducerConfig.RETRIES_CONFIG, 0);

Question 10

Initiate topology test driver

Answer 10

testDriver = new TopologyTestDriver(tracker.buildTopology());

Question 11

Create input test topic

Answer 11

vehicleRegisteredTopic = testDriver.createInputTopic(topic, serializer1, serializer2);

Question 12

Create test output topic

Answer 12

vehicleStatusTopic = testDriver.createOutputTopic(topic,deserializer1,deserializer2);

Question 13

Create test store

Answer 13

vehiclesStore = testDriver.getKeyValueStore(“vehicles-store”);

Question 14

Wrap up test with topology test driver

Answer 14

@AfterEach

public void tearDown() {

testDriver.close();

}

Question 15

test that VehicleRegistered events create vehicles in the store

Answer 15

Question 16

Test events sent to output topic

Answer 16

Question 17

Implements exception test case

Answer 17

Question 18

implement VehicleRegistered test case

Answer 18