/**
 * Copyright 2015 Cerner Corporation.
 *
 * Licensed 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.kitesdk.apps.spark.spi.kryo;

import com.esotericsoftware.kryo.Kryo;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.Lists;
import org.apache.avro.specific.SpecificRecord;
import org.apache.spark.serializer.KryoRegistrator;
import org.kitesdk.apps.AppException;

import java.util.List;

public class KryoAvroRegistrator implements KryoRegistrator {

  /**
   * System property identifying the Avro classes to be registered
   * with Spark.
   */
  public static final String KITE_AVRO_CLASSES = "kite.avro.classes";

  public void registerClasses(Kryo kryo) {

    String classesString = System.getProperty(KITE_AVRO_CLASSES);

    if (classesString == null || classesString.isEmpty()) {
      throw new AppException("Property " + KITE_AVRO_CLASSES + " not set.");

    }

    String[] classes = classesString.split(",");

    for(String className: classes) {

      Class cls = null;
      try {
        cls = Thread.currentThread().getContextClassLoader().loadClass(className);
      } catch (ClassNotFoundException e) {
        throw new AppException(e);
      }

      kryo.register(cls, new KryoAvroSerializer(cls));
    }
  }
}

@@ -678,152 +678,8 @@
 yo;%0A
-import com.google.common.annotations.VisibleForTesting;%0Aimport com.google.common.collect.Lists;%0Aimport org.apache.avro.specific.SpecificRecord;%0A
 impo
@@ -768,30 +768,63 @@
 on;%0A
-%0A
 import 
-java.util.List
+org.slf4j.Logger;%0Aimport org.slf4j.LoggerFactory
 ;%0A%0Ap
@@ -885,16 +885,87 @@
 ator %7B%0A%0A
+  Logger LOGGER = LoggerFactory.getLogger(KryoAvroRegistrator.class);%0A%0A
   /**%0A  
@@ -1304,41 +1304,53 @@
     
-throw new AppException(%22P
+LOGGER.info(%22No Avro classes set in p
 roperty 
 %22 + 
@@ -1349,11 +1349,13 @@
 rty 
-%22 +
+%7B%7D.%22,
  KIT
@@ -1372,33 +1372,93 @@
 SSES
- + %22 not set.%22
+);%0A    %7D else %7B%0A%0A      LOGGER.info(%22Registring Avro classes %7B%7D.%22, classesString
 );%0A%0A  
-  %7D%0A%0A
     
@@ -1499,24 +1499,26 @@
 plit(%22,%22);%0A%0A
+  
     for(Stri
@@ -1549,16 +1549,18 @@
 %0A%0A      
+  
 Class cl
@@ -1569,16 +1569,18 @@
 = null;%0A
+  
       tr
@@ -1591,16 +1591,18 @@
         
+  
 cls = Th
@@ -1674,16 +1674,18 @@
 ;%0A      
+  
 %7D catch 
@@ -1721,16 +1721,18 @@
         
+ 
+ 
 throw ne
@@ -1756,16 +1756,18 @@
 ;%0A      
+  
 %7D%0A%0A     
@@ -1767,16 +1767,18 @@
 %0A%0A      
+ 
+ 
 kryo.reg
@@ -1818,16 +1818,24 @@
 (cls));%0A
+      %7D%0A
     %7D%0A  

Replace exception with logging to make Spark jobs usuable in unit tests.