/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.spark.streaming.aliyun.logservice;

import com.aliyun.openservices.log.exception.LogException;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import java.util.concurrent.Callable;

final class RetryUtil {
  private static final Log LOG = LogFactory.getLog(RetryUtil.class);

  private static final long maxBackoff = 10000;
  private static final long initialBackoff = 1000;
  private static final int maxRetry = 10;

  static <T> T call(Callable<T> callable) throws Exception {
    long backoff = initialBackoff;
    int retries = 0;
    do {
      try {
        return callable.call();
      } catch (LogException ex) {
        if ("ConsumerGroupAlreadyExist".equals(ex.GetErrorCode())) {
          throw ex;
        }
        if (ex.GetHttpCode() < 500) {
          if (retries >= maxRetry) {
            LOG.error("reconnect to log-service exceed max retry times[" + maxRetry + "].");
            throw ex;
          }
          ++retries;
        } else {
          // always retry on internal server error
          retries = 0;
        }
      } catch (Exception ex) {
        if (retries >= maxRetry) {
          LOG.error("reconnect to log-service exceed max retry times[" + maxRetry + "].");
          throw ex;
        }
        ++retries;
      }
      Thread.sleep(backoff);
      backoff = Math.min(backoff * 2, maxBackoff);
    } while (true);
  }
}

@@ -995,16 +995,64 @@
 ctory;%0A%0A
+import java.util.Arrays;%0Aimport java.util.List;%0A
 import j
@@ -1318,16 +1318,286 @@
  = 10;%0A%0A
+  private static final List%3CString%3E UNRECOVERABLE_ERROR_CODES = Arrays.asList(%0A          %22ProjectNotExist%22,%0A          %22LogStoreNotExist%22,%0A          %22ConsumerGroupAlreadyExist%22,%0A          %22ConsumerGroupNotExist%22,%0A          %22InvalidCursor%22,%0A          %22Unauthorized%22%0A  );%0A%0A
   static
@@ -1808,41 +1808,41 @@
 if (
-%22ConsumerGroupAlreadyExist%22.equal
+UNRECOVERABLE_ERROR_CODES.contain
 s(ex
@@ -1982,88 +1982,29 @@
     
-LOG.error(%22reconnect to log-service exceed max retry times%5B%22 + maxRetry + %22%5D.%22);
+throw ex;%0A          %7D
 %0A   
@@ -2006,34 +2006,33 @@
 %7D%0A          
-  throw ex
+++retries
 ;%0A          
@@ -2021,35 +2021,40 @@
 etries;%0A        
-  %7D
+%7D else %7B
 %0A          ++ret
@@ -2044,33 +2044,35 @@
 %7B%0A          
-++
 retries
+ = 0
 ;%0A        %7D 
@@ -2074,99 +2074,122 @@
    %7D
- else %7B%0A          // always retry on internal server error%0A          retries = 0;%0A        %7D
+%0A        LOG.warn(%22Connecting log-service failed: %22 + ex.getMessage() + %22, retrying %22 + retries + %22/%22 + maxRetry);
 %0A   
@@ -2265,31 +2265,56 @@
     
-LOG.error(%22rec
+throw ex;%0A        %7D%0A        LOG.warn(%22C
 onnect
- to
+ing
  log
@@ -2326,82 +2326,80 @@
 ice 
-exceed max retry times%5B%22 + maxRetry + %22%5D.%22);%0A          throw ex;%0A        %7D
+failed: %22 + ex.getMessage() + %22, retrying %22 + retries + %22/%22 + maxRetry);
 %0A   

Add retry error code blacklist (#377)