[CAMEL-18627] camel-kafka: Kafka resume throws null pointer exception if no partition offset exists

• Type: Bug
• Status: Resolved
• Priority: Trivial
• Resolution: Fixed
• Affects Version/s: 3.18.2, 3.19.0
• Fix Version/s: 3.18.3, 3.20.0
• Component/s: camel-kafka
• Estimated Complexity: Novice
• Assignee: Otavio Rodolfo Piske
• Reporter: rupam
• Created: 19/Oct/22 16:22
• Updated: 20/Oct/22 09:14
• Resolved: 20/Oct/22 09:14

Description

 

In the updateTaskState() method, if the consumer is resumed and it had no previously stored offset details, then it throws a null pointer exception when it tries to retrieve ((OffsetAndMetadata) v).offset() information

 

     private void updateTaskState() {
    switch (state) {
      case PAUSE_REQUESTED:
        LOG.info("Pausing the consumer as a response to a pause request");
        consumer.pause(consumer.assignment());
        state = State.PAUSED;
        break;
      case RESUME_REQUESTED:
        LOG.info("Resuming the consumer as a response to a resume request");
        if (consumer.committed(consumer.assignment()) != null) {
          consumer.committed(consumer.assignment()).forEach((k, v) -> {
                final TopicPartition tp = ((TopicPartition) k);
                LOG.info("Resuming from offset {} for the topic {}  with partition {}",
                    ((OffsetAndMetadata) v).offset(), tp.topic(), tp.partition());

                consumer.seek(tp, ((OffsetAndMetadata) v).offset());
              }
          );
        }

        consumer.resume(consumer.assignment());
        state = State.RUNNING;
        break;
      default:
        break;
    } }
 {