public static void main(String... args) throws Exception {

// TODO(developer): Replace these variables before running the sample.

String projectId = "your-project-id";

String subscriptionId = "your-subscription-id";

subscribeWithErrorListenerExample(projectId, subscriptionId);

}

public static void subscribeWithErrorListenerExample(String projectId, String subscriptionId) {

ProjectSubscriptionName subscriptionName =

ProjectSubscriptionName.of(projectId, subscriptionId);

// Instantiate an asynchronous message receiver.

MessageReceiver receiver =

(PubsubMessage message, AckReplyConsumer consumer) -> {

// Handle incoming message, then ack the received message.

System.out.println("Id: " + message.getMessageId());

System.out.println("Data: " + message.getData().toStringUtf8());

consumer.ack();

};

Subscriber subscriber = null;

try {

// Provides an executor service for processing messages.

ExecutorProvider executorProvider =

InstantiatingExecutorProvider.newBuilder().setExecutorThreadCount(4).build();

subscriber =

Subscriber.newBuilder(subscriptionName, receiver)

.setExecutorProvider(executorProvider)

.build();

// Listen for unrecoverable failures. Rebuild a subscriber and restart subscribing

// when the current subscriber encounters permanent errors.

subscriber.addListener(

new Subscriber.Listener() {

public void failed(Subscriber.State from, Throwable failure) {

System.out.println(failure.getStackTrace());

if (!executorProvider.getExecutor().isShutdown()) {

subscribeWithErrorListenerExample(projectId, subscriptionId);

}

}

},

MoreExecutors.directExecutor());

// Start the subscriber.

subscriber.startAsync().awaitRunning();

System.out.printf("Listening for messages on %s:\n", subscriptionName.toString());

// Allow the subscriber to run for 30s unless an unrecoverable error occurs.

subscriber.awaitTerminated(30, TimeUnit.SECONDS);

} catch (TimeoutException timeoutException) {

// Shut down the subscriber after 30s. Stop receiving messages.

subscriber.stopAsync();

}

}