latest message commited except old message - using high level consumer

[anujkrp62]

i am new in kafka and want to commit kafka message explicitly that means first i will check the acknowledgement from mysql query (any operation) if it's successfully done then we will commit message otherwise it will be try same message again and again.

But here i set below configuration:

    $conf = new \RdKafka\Conf();
    $conf->set('group.id', $groupId);
    $conf->set('metadata.broker.list', $brokers);
    $conf->set('auto.offset.reset', 'earliest');
    $conf->set('enable.auto.commit', 'false');
    $conf->set('enable.auto.offset.store', 'false');
    $conf->set('auto.commit.interval.ms', 999);
    $conf->set('enable.partition.eof', 'true');
    $consumer = new \RdKafka\KafkaConsumer($conf);

commiting message by using commit().

case RD_KAFKA_RESP_ERR_NO_ERROR:
$topic_name = $message->topic_name;
$timestamp = $message->timestamp;
$payload = $message->payload;
$message_offset = $message->offset;
$queryExecutionStatus = $this->sinkConnector->injectKafka($topic_name, $payload, $timestamp);

$this->logger->info($topic_name." query execution status",[$queryExecutionStatus]);

if ($queryExecutionStatus == 1) {
	if (array_key_exists($topic_name, $search_array) ) {
	    $search_array[$topic_name] = $search_array[$topic_name] + 1;
	} else {
	    // set initial value of the counter
	    $search_array[$topic_name]=1;
	}
	$this->logger->info("manually commited offset of topic ".$topic_name, [$message_offset]);
	$consumer->commit($message);
} else {
	$this->logger->warning("Warning: Message not inserted/updated ".$topic_name, [$message_offset]);
}  

break;

Please help me how we can do this.