Introduction 

When CDAP system services are down, or when dependent services like HBase, HDFS, YARN, Kafka, ZooKeeper, Kerberos, etc are down, user programs and CDAP services must remain running, though they may be in a degraded mode where some operations will fail.

Goals

Define the expected behavior of CDAP programmatic APIs during underlying service failures

Define the expected behavior of CDAP RESTful APIs during underlying service failures

User Stories 

• As an application developer, I want my Service, Flow, Worker, and SparkStreaming programs to remain running when underlying services experience downtime
• As an application developer, I want my Service, Flow, Worker, and SparkStreaming programs to recover to a normal state when underlying service downtime ends
• As an application developer, I want to be able to configure how long my batch programs (Workflow, MapReduce, Spark batch) should retry before eventually failing when underlying services experience downtime
• As a system administrator, I want CDAP system services to remain running when underlying services experience downtime
• As a system administrator, I want CDAP RESTful APIs to return with http code 503 when required underlying services are unavailable
• As a system administrator, I want CDAP system services to recover to a normal state when underlying service downtime ends
• As a system administrator, I want CDAP program run history recording to be unaffected by failover
• As a system administrator, I want all CDAP system services to be highly available

Design (Work in Progress)

Here we will list out different types of service unavailabilities that may occur, and describe the expected behavior for different types of programs in those situations.

RS-002: Client Resiliency

From a user program's perspective, any method that involves a remote call can throw a ServiceUnavailableException, which extends RetryableException. The program can catch and handle the exception. If it doesn't, CDAP will handle it depending on the context.

CDAP APIs that involve a remote call are:

• DatasetContext.getDataset() (if the cache is not hit)
• ServiceDiscoverer.getServiceURL()
• SecureStore.listSecureData()
• SecureStore.getSecureData()
• Transactional.execute()
• Admin.datasetExists()
• Admin.getDatasetType()
• Admin.getDatasetProperties()
• Admin.createDataset()
• Admin.updateDataset()
• Admin.dropDataset()
• Admin.truncateDataset()
• StreamWriter.write() (only in Workers)
• StreamWriter.writeFile() (only in Workers)
• StreamBatchWriter (only in Workers)
• Table methods
• Location methods
• PartitionedFileSet methods

If uncaught by the program, CDAP will handle the exception depending on the context it occurs in:

• CDAP APIs that involve a remote call and that are called in a long transaction (inside a Mapper, Reducer, or Spark), or not called inside any transaction (methods where explicit transactions are used and the code is not in a TxRunnable), will be retried a configurable amount of time before throwing a ServiceUnavailableException.
• Programs running Transactional.execute(TxRunnable) will throw a TransactionFailureException with the ServiceUnavailableException as its cause. Transactional can be used in:
  • initialize methods using explicit transactions.
  • destroy methods using explicit transactions.
  • Service endpoint methods using explicit transactions.
  • CustomAction.run().
  • Worker.run().
  • Spark client.
• Programs running their initialize() method will retry a configurable amount of time before failing the program run
• Programs running their destroy() method will retry a configurable amount of time before failing the program run
• Service programs executing an endpoint method with an implicit transaction will return a 503 error code
• Flowlet programs executing their processEvent method with an implicit transaction will indefinitely retry the process method with its event
• Programs that are starting up and that use the `@UseDataSet` annotation will try to get and inject the Dataset for a configurable amount of time before failing the program run

This documents how these APIs should behave. It does not document how they currently behave. JIRAs will be opened for any differences in behavior and fixed.

We may want to consider including utility methods in the api for retrying operations if they throw a ServiceUnavailableException.

Configurable Retries

Lifecycle Methods

Consider the following code for a lifecycle method:

public class MyMapReduce extends AbstractMapReduce {

  @TransactionPolicy(TransactionControl.IMPLICIT)
  @Override
  public void destroy() {
    MapReduceContext context = getContext();
    String programName = context.getSpecification().getName();
    PartitionedFileSet output = context.getDataset("output");
    Set<PartitionKey> newOutputPartitions = getOutputPartitions();
    for (PartitionKey key : newOutputPartitions) {
      output.addPartition(key, getPartitionPath(key));
    }
  }

This method cannot be retried by CDAP because it will add metadata multiple times. It can, however, be written in an idempotent way that would make it safe to be retried:

@TransactionPolicy(TransactionControl.IMPLICIT)
@Override
public void destroy() {
  MapReduceContext context = getContext();
  String programName = context.getSpecification().getName();
  PartitionedFileSet output = context.getDataset("output");
  Set<PartitionKey> newOutputPartitions = getOutputPartitions();
  for (PartitionKey key : newOutputPartitions) {
    PartitionDetail detail = output.getPartition(key);
    if (detail != null) {
      output.addPartition(key, getPartitionPath(key));
    }
  }
}

For methods like this, it would be desirable to have the platform retry the method, especially since a service outage can easily last longer than a short transaction timeout. Since there is no way for CDAP to know what methods are idempotent, it is up to the user to configure the amount of time to retry. This can be done by setting a preference or runtime argument "remote.client.retry.timeout". A value of 0 means no retries. Another option would be to add a RetryPolicy annotation for the method. For example:

@TransactionPolicy(control = TransactionControl.IMPLICIT)
@RetryPolicy(strategybackoff = RetryStrategyRetryBackoff.FIXED_DELAY, timeout = 180000)
@Override
public void destroy() {
  MapReduceContext context = getContext();
  String programName = context.getSpecification().getName();
  PartitionedFileSet output = context.getDataset("output");
  Set<PartitionKey> newOutputPartitions = getOutputPartitions();
  for (PartitionKey key : newOutputPartitions) {
    PartitionDetail detail = output.getPartition(key);
    if (detail != null) {
      output.addPartition(key, getPartitionPath(key));
    }
  }
}

Now consider some similar code that has been changed to use a transaction per partition in order to avoid transaction timeouts when there are many output partitions:

@TransactionPolicy(TransactionControl.EXPLICIT)
@Override
public void destroy() {
  MapReduceContext context = getContext();
  final String programName = context.getSpecification().getName();
  PartitionedFileSet output;
  long startTime = System.currentTimeMillis();
  while (System.currentTimeMillis() - startTime < 180000) {
    try {
      output = context.getDataset("output");  // remote call
      break;
    } catch (RetryableException e) {
      TimeUnit.SECONDS.sleep(1);
    }
  }
  if (output == null) {
    throw new RuntimeException("Unable to get output dataset after 3 minutes");
  }

  Set<PartitionKey> newOutputPartitions = getOutputPartitions();
  for (final PartitionKey key : newOutputPartitions) {
    long startTime = System.currentTimeMillis();
    boolean succeeded = false;
    while (System.currentTimeMillis() - startTime < 180000) {
      try {
        context.execute(new TxRunnable() { // remote call
          @Override
          public void run(DatasetContext context) throws Exception {
            output.addPartition(key, getPartitionPath(key)); // remote call
          }
        });
        succeeded = true;
        break;
      } catch (RetryableTransactionFailureException e) {
        TimeUnit.SECONDS.sleep(1);
      } catch (TransactionFailureException e) {
        throw new RuntimeException(e);
      }
    }
    if (!succeeded) {
      throw new RuntimeException("Unable to add metadata after 3 minutes");
    }
  }
}

The destroy() method is no longer idempotent because it starts multiple transactions, which means you would not want to configure CDAP to retry this method. Instead, the developer is forced to add their own retry logic around every remote call. This, however, makes the code very messy, and requires similar logic to be present in every single user program. In these situations, it would be desirable for CDAP to handle the retries for the getDataset() call, and also for the execute() call.  We could extend the annotation idea to apply to all remote calls within a method. Then the code would look like:

@TransactionPolicy(TransactionControl.EXPLICIT)
@RetryPolicy(strategybackoff = RetryStrategyRetryBackoff.FIXED_DELAY, timeout = 180000)
@Override
public void destroy() {
  MapReduceContext context = getContext();
  final String programName = context.getSpecification().getName();
  final PartitionedFileSet output = context.getDataset("output");  // remote call

  Set<PartitionKey> newOutputPartitions = getOutputPartitions();
  for (final PartitionKey key : newOutputPartitions) {
    try {
      context.execute(new TxRunnable() { // remote call
        @Override
        public void run(DatasetContext context) throws Exception {
          output.addPartition(key, getPartitionPath(key)); // remote call
        }
      }
    } catch (TransactionFailureException e) {
      throw new RuntimeException(e);
    }
  }
}

The retry policy would then apply to the getDataset() call and to the execute() call. It would not, however, apply to the addPartition() call, though TxRunnable's run() method could also be annotated if desired. This does make things a little confusing, since it is hard to tell what the retry policy applies to. Another option would be to add methods for each remote call to take a RetryStrategy as an argument. For example:

@TransactionPolicy(TransactionControl.EXPLICIT)
@Override
public void destroy() {
  MapReduceContext context = getContext();
  final String programName = context.getSpecification().getName();
  final RetryPolicy retryPolicy = RetryPolicies.timeLimit(3, TimeUnit.MINUTES, RetryPolicies.fixedDelayfixedBackoff(1, TimeUnit.SECONDS));
  final PartitionedFileSet output = context.getDataset("output", retryPolicy);  // remote call
  Set<PartitionKey> newOutputPartitions = getOutputPartitions();
  for (final PartitionKey key : newOutputPartitions) {
    try {
      context.execute( // remote call
        new TxRunnable() {
          @Override
          public void run(DatasetContext context) throws Exception {
            output.addPartition(key, getPartitionPath(key)); // remote call
          }
        },
        30, retryPolicy);
    } catch (TransactionFailureException e) {
      throw new RuntimeException(e);
    }
  }
}

This would have some additional benefits of indicating to users which methods involve remote calls and which ones do not, and eliminating any ambiguity about what will be retried. However, this is a little more verbose and could make it more difficult to evolve the API as more parameters are added to various method calls.

CDAP System Service Failures

These are various services that CDAP runs, both inside YARN and on edge nodes. This section documents what service degradations are expected if various CDAP system services are not available.

Transaction

Batch Programs (MapReduce, Spark, Workflow)

• transaction start will be retried for a configurable amount of time before the run is failed
• transaction commit will be retried for a configurable amount of time before the run is failed

Dataset

• Programs that use the @UseDataSet annotation and that are starting up and that will fail their YARN container. The container will be retried up to a configurable number of times.
• Dataset REST APIs will return 503
• Deploy application REST API will fail if the application creates any datasets and return a 503
• TODO: check if other REST APIs will fail
• Explore queries on datasets will fail
• See Programmatic APIs section. Applies to:
  • DatasetContext.getDataset() (if the cache is not hit)
  • ServiceDiscoverer.getServiceURL()
  • SecureStore.listSecureData()
  • SecureStore.getSecureData()
  • Admin.datasetExists()
  • Admin.getDatasetType()
  • Admin.getDatasetProperties()
  • Admin.createDataset()
  • Admin.updateDataset()
  • Admin.dropDataset()
  • Admin.truncateDataset()

Streams

• Stream REST APIs will return 503
• Deploy application REST API will fail if the application creates any streams and return a 503
• Notifications on stream size will not be sent (shouldn't be triggered anyway since there is no way to write to a stream when this happens)
• StreamWriter methods will throw a ServiceUnavailableException. The developer can catch and handle the exception. If it is not caught and handled, the platform will handle it.
  • StreamWriter is only available in a Worker. Uncaught exceptions will fail the run
• Old Stream files on HDFS will not be cleaned up until the service is restored
• See Programmatic APIs section. Applies to:
  • StreamWriter.write() (only in Workers)
  • StreamWriter.writeFile() (only in Workers)
  • StreamBatchWriter (only in Workers)

Explore

• Explore REST APIs will return with 503
• Queries that are in progress will be lost
• Queries that are in progress of writing data are undefined
• See Programmatic APIs section. Applies to the following methods if the dataset referenced is explorable:
  • Admin.createDataset()
  • Admin.updateDataset()
  • Admin.dropDataset()
  • Admin.truncateDataset()
  • PartitionedFileSet.addPartition()
  • PartitionedFileSet.dropPartition()

Metrics

• REST APIs to fetch metrics and logs will fail with 503

Metrics Processor

• Metrics emitted after the outage will not be returned by the REST APIs to fetch metrics
• If the outage lasts longer than Kafka retention, any unprocessed metrics older than the retention will be lost

Log Saver

• Logs emitted after the outage will not be returned by the REST APIs to fetch logs
• If the outage lasts longer than Kafka retention, any log messages older than the retention will be lost

Master

Consider the follow mapreduce code taken from the Purchase example:

Code Block
@Override public void initialize() throws Exception { MapReduceContext context = getContext(); Job job = context.getHadoopJob(); job.setReducerClass(PerUserReducer.class); context.addInput(Input.ofDataset("purchases"), PurchaseMapper.class); context.addOutput(Output.ofDataset("history")); }

In this example, the MapReduce job reads from an ObjectMappedTable called "purchases" and writes to a custom dataset called "history". This means that all the code handling retries and timeouts is located in the datasets and not in this program. Today, with these particular datasets, read operations eventually boil down to HBase scan calls, with any exception simply being propagated up as a runtime exception. This means the resiliency is actually dependent on the hbase client settings in hbase-site.xml. In order to configure retries and timeouts from job to job, the developer would have to know that HBase is being used behind the scenes, and to set various hbase settings in the job Configuration.

Code Block

@Override public void initialize() throws Exception { MapReduceContext context = getContext(); Job job = context.getHadoopJob(); job.setReducerClass(PerUserReducer.class); context.addInput(Input.ofDataset("purchases"), PurchaseMapper.class); context.addOutput(Output.ofDataset("history")); Configuration config = job.getConfiguration(); config.setInt("hbase.client.operation.timeout", 3 * 60 * 60 * 1000); }

Requiring this breaks abstraction barriers and also requires strong hbase knowledge. One way to address this is to allow setting a retry policy for the job.

Code Block

@Override public void initialize() throws Exception { MapReduceContext context = getContext(); Job job = context.getHadoopJob(); job.setReducerClass(PerUserReducer.class); context.addInput(Input.ofDataset("purchases"), PurchaseMapper.class); context.addOutput(Output.ofDataset("history")); context.setRetryPolicy(RetryPolicies.timeLimit(3, TimeUnit.HOURS, RetryPolicies.fixedBackoff(1, TimeUnit.MINUTE))); }

This will tell the CDAP MapReduce framework to retry any BatchReadable or BatchWritable method that throws a RetryableException for 3 hours. In addition, the datasets would need to make sure that this exception is thrown on operations that can be retried. This would also apply to any remote calls that happen inside the Mapper or Reducer. For example:

Code Block

public static class PerUserReducer extends Reducer<Text, Purchase, String, PurchaseHistory> implements ProgramLifecycle<MapReduceContext> { private static final Logger LOG = LoggerFactory.getLogger(PerUserReducer.class); @UseDataSet("frequentCustomers") // remote call private KeyValueTable frequentCustomers; private URL userProfileServiceURL; @Override public void initialize(MapReduceContext context) throws Exception { userProfileServiceURL = context.getServiceURL(UserProfileServiceHandler.SERVICE_NAME); // remote call } public void reduce(Text customer, Iterable<Purchase> values, Context context) throws IOException, InterruptedException { UserProfile userProfile = null; try { URL url = new URL(userProfileServiceURL, UserProfileServiceHandler.USER_ENDPOINT + "/" + customer.toString()); HttpRequest request = HttpRequest.get(url).build(); HttpResponse response = HttpRequests.execute(request); if (response.getResponseCode() != HttpURLConnection.HTTP_NO_CONTENT) { userProfile = new Gson().fromJson(response.getResponseBodyAsString(), UserProfile.class); } } catch (Exception e) { LOG.warn("Error accessing user profile.", e); } PurchaseHistory purchases = new PurchaseHistory(customer.toString(), userProfile); int numPurchases = 0; for (Purchase val : values) { purchases.add(new Purchase(val)); numPurchases++; } if (numPurchases > 10) { frequentCustomers.write(customer.toString(), String.valueOf(numPurchases)); } context.write(customer.toString(), purchases); } }

If context.getServiceURL() throws a RetryableException, it would be retried for 3 hours. Similarly, if flushing the frequentCustomers table throws a RetryableException, it would retry the flush for 3 hours as well. Similarly, when the Reducer is being instantiated, and the frequentCustomers field is being injected (involves a .getDataset() call), it will retry for 3 hours if the dataset service is unavailable.

Spark would work in a similar way, with dataset methods used in .fromDataset() and saveAsDataset() being retried if they throw a RetryableException. 

Setting a retry policy at the workflow level would apply to all programs in the workflow if the individual programs don't specify one.

CDAP System Service Failures

These are various services that CDAP runs, both inside YARN and on edge nodes. This section documents what service degradations are expected if various CDAP system services are not available.

Transaction

• Transactions cannot be started or committed.
• In contexts where a long transaction is used (MapReduce, Spark), 
  • transaction start will be retried for a configurable amount of time before the run is failed
  • transaction commit will be retried for a configurable amount of time before the run is failed
• In methods where a short transaction is used:
  • any transaction failure in a Service method will cause a 503 to be returned
  • transaction start failures in a flowlet will be retried indefinitely
  • transaction commit failures in a flowlet will cause the process method to be retried
  • transaction start failures before an initialize() or destroy() method will be retried for a configurable amount of time before causing the program to fail
  • transaction commit failures after an initialize() or destroy() method will cause the method to throw a ServiceUnavailableException. If uncaught, the program will fail
  • transaction start failures in Transactional.execute() will retry transaction start a configurable amount of time before throwing a TransactionFailureException with the ServiceUnavailableException as its cause
  • transaction commit failures in Transactional.execute() will throw a TransactionFailureException with the ServiceUnavailableException as its cause
• CDAP REST APIs will return a 503
• Metrics Processor will be unable to process metrics
• Log Saver will be unable to save logs

Dataset

• Programs that use the @UseDataSet annotation and that are starting up and that will fail their YARN container. The container will be retried up to a configurable number of times.
• Dataset REST APIs will return 503
• Deploy application REST API will fail if the application creates any datasets and return a 503
• TODO: check if other REST APIs will fail
• Explore queries on datasets will fail
• See Programmatic APIs section. Applies to:
  • DatasetContext.getDataset() (if the cache is not hit)
  • ServiceDiscoverer.getServiceURL()
  • SecureStore.listSecureData()
  • SecureStore.getSecureData()
  • Admin.datasetExists()
  • Admin.getDatasetType()
  • Admin.getDatasetProperties()
  • Admin.createDataset()
  • Admin.updateDataset()
  • Admin.dropDataset()
  • Admin.truncateDataset()

Streams

• Stream REST APIs will return 503
• Deploy application REST API will fail if the application creates any streams and return a 503
• Notifications on stream size will not be sent (shouldn't be triggered anyway since there is no way to write to a stream when this happens)
• StreamWriter methods will throw a ServiceUnavailableException. The developer can catch and handle the exception. If it is not caught and handled, the platform will handle it.
  • StreamWriter is only available in a Worker. Uncaught exceptions will fail the run
• Old Stream files on HDFS will not be cleaned up until the service is restored
• See Programmatic APIs section. Applies to:
  • StreamWriter.write() (only in Workers)
  • StreamWriter.writeFile() (only in Workers)
  • StreamBatchWriter (only in Workers)

Explore

• Explore REST APIs will return with 503
• Queries that are in progress will be lost
• Queries that are in progress of writing data are undefined
• See Programmatic APIs section. Applies to the following methods if the dataset referenced is explorable:
  • Admin.createDataset()
  • Admin.updateDataset()
  • Admin.dropDataset()
  • Admin.truncateDataset()
  • PartitionedFileSet.addPartition()
  • PartitionedFileSet.dropPartition()

Metrics

• REST APIs to fetch metrics and logs will fail with 503

Metrics Processor

• Metrics emitted after the outage will not be returned by the REST APIs to fetch metrics
• If the outage lasts longer than Kafka retention, any unprocessed metrics older than the retention will be lost

Log Saver

• Logs emitted after the outage will not be returned by the REST APIs to fetch logs
• If the outage lasts longer than Kafka retention, any log messages older than the retention will be lost

Master

• Namespace, Metadata, Preference, Configuration, Security, Artifact, and Lifecycle CDAP REST APIs will fail with 503 
• Scheduled workflow runs will be missed
• Program state changes will not be recorded. Program state may be incorrect right after service is restored, but will eventually be corrected
• Program run records will not be updated. Run records may be incorrect right after service is restored, but will eventually be corrected
• Programs that are starting but have not yet received their YARN containers will fail
• cluster metrics like memory available and used will not be collected for the time period of the downtime

Auth

• CDAP users will be unable to get a security token

Router

• All CDAP REST APIs will fail
  
  

TODO: What service does SecureStore use?

Hadoop Infrastructure

This section documents expected degradation when various Hadoop systems are down. Note that an outage in any one of these services will likely have an impact on the CDAP system services as well.

HBase

• DatasetService should return 503, so all degradations in DatasetService will apply here
• Program state will not be recorded. Once service is restored, state will eventually be corrected
  • This should not cause user programs that finish during the outage to fail
• Program run records will not be updated. Once service is restored, state will eventually be corrected
• CDAP REST APIs that lookup, create, modify, or delete a CDAP entity will fail with a 503
  • TODO: are there any APIs that don't fall under this category? fetch logs maybe?
• Metrics still in Kafka will not be processed. If the outage lasts longer than Kafka retention, unprocessed metrics will be lost
• Log messages still in Kafka will not be saved. If the outage lasts longer than Kafka retention, log messages will be lost
• Programs will be unable to start. Scheduled program runs will fail
• See Programmatic APIs section. Applies to:
  • DatasetContext.getDataset() (if the cache is not hit)
  • Admin.datasetExists()
  • Admin.getDatasetType()
  • Admin.getDatasetProperties()
  • Admin.createDataset()
  • Admin.updateDataset()
  • Admin.dropDataset()
  • Admin.truncateDataset()
  • PartitionedFileSet methods
  • Table get and scan methods
  • Due to the usage of BufferingTable, Table put and increment methods won't throw exceptions. HBase unavailability will only be seen by the CDAP platform when a transaction is committed.
• TODO: investigate effect on HBase replication 

Notes:

'normal' HBase failures can take a while to recover from. For example, if an HBase regionserver it can take minutes for the master to notice and reassign the regions. This suggest that the default retry timeout should be pretty high.

Timeouts and retries for Tables may be difficult to control, as HTable and HBaseAdmin have their own logic and settings around retries and timeouts based on hbase-site.xml. HTable has a .setOperationTimeout() that it seems to use for most RPC calls it makes, and which defaults to the value of 'hbase.client.operation.timeout' (which in turn defaults to max long value in some versions of hbase). HBaseAdmin also uses this timeout for some operations, like creating a namespace, but for other operations it uses the 'hbase.client.retries.number' setting along with 'hbase.client.pause', and sometimes multiplied by 'hbase.client.retries.longer.multiplier'. The defaults and logic differ between hbase versions.

In the CDAP system services, it may make sense to use smaller timeouts and retries so that operations always return within 10 seconds or so instead of 20 minutes.

HDFS

• Transaction snapshots will not be saved. If the transaction server dies, some transaction state will be lost
• HBase relies on HDFS and will be unavailable. Everything under HBase applies here.
• All REST APIs will return 503
• See programmatic APIs section. Applies to all APIs.

YARN

• All REST APIs will return 503
• CDAP system services should be re-launched once YARN returns
• User programs that were running when YARN died should be re-launched once YARN returns
• TODO: what happens to running java processes? Do they keep running as zombies?
• If YARN is down longer than the Kafka retention, unprocessed metrics and unsaved log messages will be lost

Kafka

• Programs that emit logs and metrics will buffer their data for some time. After some configurable timeout, metrics and logs will be dropped
• Stream size notifications will not be sent
• Recent, unsaved log messages will not be return by the CDAP REST log endpoints
• If the outage lasts longer than the Kafka retention, log messages and metrics will be lost

ZooKeeper

• All CDAP REST APIs will return 503, since CDAP services will be undiscoverable
• See programmatic APIs section. All APIs apply, since CDAP services will be undiscoverable
• Any HBase regionserver failures will go unnoticed. If this happens, HBase requests to regions belonging to dead regionservers will fail. See the HBase section for expected behavior in these situations

Kerberos

• TODO: investigate what happens here. Probably won't be able to start programs and some might expire and die, in which case we should continuously relaunch?

Configurable Timeouts

We mention that several operations will be retried for a configurable amount of time before programs start to fail. This timeout should be configurable down to the program run level through preferences and runtime arguments. For example, a MapReduce program that usually runs for a week may want to use a timeout of a few hours, whereas a program that usually runs for just ten minutes may want to use a timeout of a couple minutes.

One option is to introduce a config property 'remote.service.timeout' that defaults to something like 3 minutes. Users can then scope the property through preferences and runtime arguments to apply across a namespace, app, program, or program run.  For flows, a prefix (flowlet.[flowletname].) can be added to narrow the scope to a specific flowlet. Workflow components can also be scoped as described by http://docs.cdap.io/cdap/current/en/admin-manual/operations/preferences.html#scoped-runtime-arguments-for-workflow.the CDAP system services as well.

HBase

• DatasetService should return 503, so all degradations in DatasetService will apply here
• Program state will not be recorded. Once service is restored, state will eventually be corrected
  • This should not cause user programs that finish during the outage to fail
• Program run records will not be updated. Once service is restored, state will eventually be corrected
• CDAP REST APIs that lookup, create, modify, or delete a CDAP entity will fail with a 503
  • TODO: are there any APIs that don't fall under this category? fetch logs maybe?
• Metrics still in Kafka will not be processed. If the outage lasts longer than Kafka retention, unprocessed metrics will be lost
• Log messages still in Kafka will not be saved. If the outage lasts longer than Kafka retention, log messages will be lost
• Programs will be unable to start. Scheduled program runs will fail
• See Programmatic APIs section. Applies to:
  • DatasetContext.getDataset() (if the cache is not hit)
  • Admin.datasetExists()
  • Admin.getDatasetType()
  • Admin.getDatasetProperties()
  • Admin.createDataset()
  • Admin.updateDataset()
  • Admin.dropDataset()
  • Admin.truncateDataset()
  • PartitionedFileSet methods
  • Table get and scan methods
  • Due to the usage of BufferingTable, Table put and increment methods won't throw exceptions. HBase unavailability will only be seen by the CDAP platform when a transaction is committed.
• TODO: investigate effect on HBase replication 

Notes:

'normal' HBase failures can take a while to recover from. For example, if an HBase regionserver it can take minutes for the master to notice and reassign the regions. This suggest that the default retry timeout should be pretty high.

Timeouts and retries for Tables may be difficult to control, as HTable and HBaseAdmin have their own logic and settings around retries and timeouts based on hbase-site.xml. HTable has a .setOperationTimeout() that it seems to use for most RPC calls it makes, and which defaults to the value of 'hbase.client.operation.timeout' (which in turn defaults to max long value in some versions of hbase). HBaseAdmin also uses this timeout for some operations, like creating a namespace, but for other operations it uses the 'hbase.client.retries.number' setting along with 'hbase.client.pause', and sometimes multiplied by 'hbase.client.retries.longer.multiplier'. The defaults and logic differ between hbase versions.

In the CDAP system services, it may make sense to use smaller timeouts and retries so that operations always return within 10 seconds or so instead of 20 minutes.

HDFS

• Transaction snapshots will not be saved. If the transaction server dies, some transaction state will be lost
• HBase relies on HDFS and will be unavailable. Everything under HBase applies here.
• All REST APIs will return 503
• See programmatic APIs section. Applies to all APIs.

YARN

• All REST APIs will return 503
• CDAP system services should be re-launched once YARN returns
• User programs that were running when YARN died should be re-launched once YARN returns
• TODO: what happens to running java processes? Do they keep running as zombies?
• If YARN is down longer than the Kafka retention, unprocessed metrics and unsaved log messages will be lost

Kafka

• Programs that emit logs and metrics will buffer their data for some time. After some configurable timeout, metrics and logs will be dropped
• Stream size notifications will not be sent
• Recent, unsaved log messages will not be return by the CDAP REST log endpoints
• If the outage lasts longer than the Kafka retention, log messages and metrics will be lost

ZooKeeper

• All CDAP REST APIs will return 503, since CDAP services will be undiscoverable
• See programmatic APIs section. All APIs apply, since CDAP services will be undiscoverable
• Any HBase regionserver failures will go unnoticed. If this happens, HBase requests to regions belonging to dead regionservers will fail. See the HBase section for expected behavior in these situations

Kerberos

• TODO: investigate what happens here. Probably won't be able to start programs and some might expire and die, in which case we should continuously relaunch?

RESTful APIs

The CDAP REST APIs should always respond, even if that response is a 503.

Many internal changes will be required to ensure that there is error handling in place in case any remote call fails. For example, many internal services create Tables to store metadata. These services must not assume that table creation was successful. Instead, if it failed, they must keep trying to create that table, and in the meantime throw ServiceUnavailableException in their methods that try and use that Table. 

Internal services should also not require underlying infrastructure in order to start up or continue running. For example, the CDAP master today will try and create the default HBase namespace during startup, and will hang for a long time if HBase is down. 

RS-003: System Services HA

Most system services are already scalable to multiple instances (explore is not). Need to verify that if one instance dies, it is re-launched correctly. At a high level, making the Explore service HA amounts to moving any in-memory state to a dataset. Like any explore change, it will probably end up being a significant amount of work. 

Program state transition is also handled in memory in app fabric, (we add a Listener to the program controller), which means state transition logic will not be run if programs are started with one master, then the master fails over before the run finishes. The run record corrector will eventually set the state to 'failed', but that may not match the state it would normally transition to ('completed', 'killed', etc.). CDAP-7496

API changes

New Programmatic APIs

There will be a new ServiceUnavailable exception that relevant methods will throw. It will be a RuntimeException.

Deprecated Programmatic APIs

No programmatic APIs will be deprecated

New REST APIs

No new REST APIs will be introduced

Deprecated REST API

No REST APIs will be deprecated

CLI Impact or Changes

• None

UI Impact or Changes

• None

Security Impact 

CDAP should make sure tokens are handled correctly if Kerberos ever does down

Impact on Infrastructure Outages 

Should make CDAP resilient to infrastructure outages

Test Scenarios

It is difficult to test all possible scenarios. A few are listed below and more will be added. It would be good if we could run the long running tests in combination with a chaos monkey.

Releases

Release 4.1.0

Release 4.2.0

Related Work

Future work