apache/seatunnel · error · IOException

Failed to publish NATS JetStream message for subtask ${subta

Error message

Failed to publish NATS JetStream message for subtask ${subtaskIndex} to subject '${subject}'

What it means

This IOException is thrown by the sink writer's publish method when jetStream.publish fails for any reason (JetStreamApiException, IOException, or RuntimeException) while writing a message to the given subject. It wraps the underlying cause with subtask index and subject context so the failure is attributable in logs/checkpoint handling; being thrown from write(), it will fail the sink task and trigger the engine's retry/restart semantics.

Source

Thrown at seatunnel-connectors-v2/connector-nats-jetstream/src/main/java/org/apache/seatunnel/connectors/seatunnel/natsjetstream/sink/NatsJetStreamSinkWriter.java:197

        } catch (Exception e) {
            primaryFailure.addSuppressed(new IOException("Failed to close NATS connection", e));
        }
    }

    private void publish(PublishRequest publishRequest) throws IOException {
        PublishOptions.Builder optionsBuilder =
                PublishOptions.builder().streamTimeout(PUBLISH_TIMEOUT);
        if (publishRequest.messageId != null) {
            optionsBuilder.messageId(publishRequest.messageId);
        }
        try {
            jetStream.publish(
                    publishRequest.subject,
                    publishRequest.headers,
                    publishRequest.payload,
                    optionsBuilder.build());
        } catch (JetStreamApiException e) {
            throw publishFailure(publishRequest.subject, e);
        } catch (IOException | RuntimeException e) {
            throw publishFailure(publishRequest.subject, e);
        }
    }

    private IOException publishFailure(String subject, Exception cause) {
        return new IOException(
                String.format(
                        "Failed to publish NATS JetStream message for subtask %d to subject '%s'",
                        subtaskIndex, subject),
                cause);
    }

    private static boolean isNotBlank(String value) {
        return value != null && !value.trim().isEmpty();
    }

    static NatsJetStreamConnectorException invalidRecord(String fieldName, String message) {

View on GitHub (pinned to cf67b549a7)

Solutions

Verify the subject is covered by an existing JetStream stream (nats stream info / subject mapping)
Check NATS server connectivity and credentials; test with nats pub to the same subject
Compare payload size against the stream's max_msg_size and compress/trim payloads
Retry the job — transient network errors may resolve; add sink retry/resilience options if available
Inspect the wrapped cause in logs for the specific JetStream API error code

Example fix

// before
jetStream.publish(subject, payload); // fails: no stream covers 'events'
// after
// create/bind the stream first:
// nats str add EVENTS --subjects "events.>"
jetStream.publish("events.data", payload);

Defensive patterns

Strategy: try-catch

Validate before calling

// pre-flight: ensure stream covers subject and server is reachable
try (Connection c = Nats.connect("nats://host:4222")) {
    JetStreamManagement jsm = c.jetStreamManagement();
    StreamInfo si = jsm.getStreamForSubject(subject);
    System.out.println("Stream " + si.getConfig().getName() + " covers " + subject);
}

Try / catch

try {
    writer.write(row);
} catch (IOException e) {
    if (e.getCause() instanceof JetStreamApiException) {
        // inspect API error code; non-retryable (e.g. no stream) -> fail fast / fix config
    } else {
        // transient I/O -> rely on engine restart or backoff-retry
    }
    throw e; // let checkpointing record the failure
}

Prevention

Create JetStream streams covering all publish subjects before running jobs
Monitor NATS server health and set appropriate connection timeouts/reconnect settings
Keep payloads under the stream's max_msg_size (default 1MB)
Ensure publish permissions exist for the connecting NATS account
Test subjects with nats CLI before production deployment

When it happens

Trigger: NATS server returns a JetStream API error (e.g. no responders, stream not found, message too large, exceeded max message size or publish limits); network I/O errors to the NATS server; any RuntimeException from the publish path.

Common situations: Stream or subject not bound to a JetStream stream (subject has no listeners/stream); NATS server down or unreachable; publishing payloads exceeding stream max_msg_size; authentication/authorization failures at publish time; transient network partitions during long-running jobs.

Understand the failure class

Background: "API request failed": what wrapped HTTP errors from external APIs mean and how to find the real cause — this error's family across 29 libraries.

Related errors

AI-assisted analysis of apache/seatunnel@cf67b549a7 (2026-09-10). Data as JSON: /api/errors/251dfc1dcdbea997. Report an issue: GitHub.