When to Extend Visibility

Some tasks take longer than expected. Instead of setting an excessively long initial visibility timeout, you can extend the timeout mid-processing to prevent the message from being redelivered.

Extending the timeout before it lapses keeps a slow task hidden from other consumers until the consumer acknowledges.

Steps

Receive with Initial Visibility

extend_visibility.go

resp, err := client.PollQueue(ctx, &kubemq.PollRequest{
    Channel:            "orders",
    MaxItems:           1,
    WaitTimeoutSeconds: 5,
    VisibilitySeconds:  60,
})
if err != nil {
    log.Fatal(err)
}

extend_visibility.py

response = client.receive_queue_messages(
    channel="orders",
    max_messages=1,
    wait_timeout_in_seconds=5,
    visibility_seconds=60,
)

extend_visibility.ts

const messages = await client.receiveQueueMessages({
  channel: 'orders',
  maxMessages: 1,
  waitTimeoutSeconds: 5,
  visibilitySeconds: 60,
});

ExtendVisibility.java

ReceiveQueueMessagesResponse response = client.receiveQueueMessages(
    ReceiveQueueMessagesRequest.builder()
        .channel("orders")
        .maxMessages(1)
        .waitTimeoutSeconds(5)
        .visibilitySeconds(60)
        .build());

ExtendVisibility.cs

var response = await client.ReceiveQueueMessagesAsync(new ReceiveQueueMessagesRequest
{
    Channel = "orders",
    MaxMessages = 1,
    WaitTimeoutSeconds = 5,
    VisibilitySeconds = 60,
});

ExtendVisibility.kt

val response = client.receiveQueueMessages(
    channel = "orders",
    maxMessages = 1,
    waitTimeoutSeconds = 5,
    visibilitySeconds = 60
)

extend_visibility.cpp

auto response = client.receiveQueueMessages("orders", 1, 5, false, 60);

extend_visibility.rs

// Poll the queue with explicit ack control (auto_ack = false).
let (response, mut receiver) = client
    .poll_queue(PollRequest {
        channel: "orders".to_string(),
        max_items: 1,
        wait_timeout_seconds: 5,
        auto_ack: false,
    })
    .await?;

extend_visibility.rb

receiver = client.create_downstream_receiver
response = receiver.poll(
  KubeMQ::Queues::QueuePollRequest.new(
    channel: 'orders',
    max_items: 1,
    wait_timeout: 5,
    auto_ack: false
  )
)

extend_visibility.exs

{:ok, poll} =
  KubeMQ.Client.poll_queue(client,
    channel: "orders",
    max_items: 1,
    wait_timeout: 5_000,
    auto_ack: false
  )

Process and Extend When Needed

Before the visibility timeout expires, extend it to get more processing time.

for _, m := range resp.Messages {
    fmt.Printf("Processing: %s\n", string(m.Message.Body))

    // Phase 1: Quick validation
    time.Sleep(10 * time.Second)

    // Need more time — extend by 60 more seconds
    if err := m.ExtendVisibility(60); err != nil {
        log.Printf("Failed to extend visibility: %v", err)
    }
    fmt.Println("Visibility extended by 60 seconds")

    // Phase 2: Heavy processing
    time.Sleep(40 * time.Second)
}
resp.AckAll()
fmt.Println("Done — all messages acknowledged")

for msg in response.messages:
    print(f"Processing: {msg.body.decode('utf-8')}")

    # Phase 1: Quick validation
    time.sleep(10)

    # Need more time — extend by 60 more seconds
    msg.extend_visibility(60)
    print("Visibility extended by 60 seconds")

    # Phase 2: Heavy processing
    time.sleep(40)
    msg.ack()

print("Done — all messages acknowledged")

for (const msg of messages) {
  console.log('Processing:', new TextDecoder().decode(msg.body));

  // Phase 1: Quick validation
  await new Promise((r) => setTimeout(r, 10000));

  // Need more time — extend by 60 more seconds
  await msg.extendVisibility(60);
  console.log('Visibility extended by 60 seconds');

  // Phase 2: Heavy processing
  await new Promise((r) => setTimeout(r, 40000));
  await msg.ack();
}
console.log('Done — all messages acknowledged');

for (QueueMessageReceived msg : response.getMessages()) {
    System.out.println("Processing: " + new String(msg.getBody()));

    Thread.sleep(10_000);

    msg.extendVisibility(60);
    System.out.println("Visibility extended by 60 seconds");

    Thread.sleep(40_000);
    msg.ack();
}
System.out.println("Done — all messages acknowledged");

foreach (var msg in response.Messages)
{
    Console.WriteLine($"Processing: {Encoding.UTF8.GetString(msg.Body.Span)}");

    await Task.Delay(10_000);

    await msg.ExtendVisibilityAsync(60);
    Console.WriteLine("Visibility extended by 60 seconds");

    await Task.Delay(40_000);
    await msg.AckAsync();
}
Console.WriteLine("Done — all messages acknowledged");

for (msg in response.messages) {
    println("Processing: ${String(msg.body)}")

    Thread.sleep(10_000)

    msg.extendVisibility(60)
    println("Visibility extended by 60 seconds")

    Thread.sleep(40_000)
    msg.ack()
}
println("Done — all messages acknowledged")

for (const auto& msg : response.messages) {
    std::cout << "Processing: " << msg.body << std::endl;

    std::this_thread::sleep_for(std::chrono::seconds(10));

    msg.extendVisibility(60);
    std::cout << "Visibility extended by 60 seconds" << std::endl;

    std::this_thread::sleep_for(std::chrono::seconds(40));
    msg.ack();
}
std::cout << "Done — all messages acknowledged" << std::endl;

// Extending an in-flight message's visibility is not yet available in the Rust SDK —
// the downstream message exposes ack(), nack(), and re_queue() only.
// Set a generous initial visibility on the poll, then ack once processing completes.
for msg in &response.messages {
    println!("Processing: {}", String::from_utf8_lossy(&msg.message.body));

    // Phase 1: Quick validation
    tokio::time::sleep(std::time::Duration::from_secs(10)).await;

    // Phase 2: Heavy processing
    tokio::time::sleep(std::time::Duration::from_secs(40)).await;

    msg.ack().await?;
}
println!("Done — all messages acknowledged");
receiver.close().await?;

# Extending an in-flight message's visibility is not yet available in the Ruby SDK —
# each polled message exposes ack and nack, while requeue is settled at the
# response level via response.requeue_all(channel:).
# Set a generous initial visibility on the poll, then ack once processing completes.
response.messages.each do |msg|
  puts "Processing: #{msg.body}"

  # Phase 1: Quick validation
  sleep(10)

  # Phase 2: Heavy processing
  sleep(40)

  msg.ack
end
puts 'Done — all messages acknowledged'
receiver.close

# Extending an in-flight message's visibility is not yet available in the Elixir SDK —
# PollResponse exposes ack_all/1, nack_all/1, and requeue_all/2 only.
# Set a generous initial visibility on the poll, then ack once processing completes.
Enum.each(poll.messages, fn msg ->
  IO.puts("Processing: #{msg.body}")

  # Phase 1: Quick validation
  Process.sleep(10_000)

  # Phase 2: Heavy processing
  Process.sleep(40_000)
end)

{:ok, _} = KubeMQ.PollResponse.ack_all(poll)
IO.puts("Done — all messages acknowledged")

Always extend visibility before the current timeout expires. Once the timeout lapses, the message may be delivered to another consumer, leading to duplicate processing.

Extend Visibility Timeout

When to Extend Visibility

Steps

Receive with Initial Visibility

Process and Extend When Needed

Next Steps

Visibility Timeout

Queue Reference

On this page