public static long messageCount(String bootstrapServers, String topicName, int partition) {
    Properties props = new Properties();
    String consumerId = UUID.randomUUID().toString();
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
    props.put(ConsumerConfig.GROUP_ID_CONFIG, "spark-consumer-group-"+consumerId);
    props.put(ConsumerConfig.CLIENT_ID_CONFIG, "spark-consumer-"+consumerId);
    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, IntegerDeserializer.class.getName());
    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ExternalizableDeserializer.class.getName());

    KafkaConsumer<Integer, Externalizable> consumer = new KafkaConsumer<Integer, Externalizable>(props);

    TopicPartition topicPartition = new TopicPartition(topicName, partition);
    List<TopicPartition> partitionList = Arrays.asList(topicPartition);
    consumer.assign(partitionList);
    consumer.seekToEnd(partitionList);
    long nextOffset = consumer.position(topicPartition);

    consumer.seekToBeginning(partitionList);
    long firstOffset = consumer.position(topicPartition);

    consumer.close();

    return nextOffset - firstOffset;
}
 

public static void sendAbsStructMsg(QQAppInterface qqAppInterface, Parcelable sessionInfo, Externalizable absStructMsg) {
    if (qqAppInterface == null) throw new NullPointerException("qqAppInterface == null");
    if (sessionInfo == null) throw new NullPointerException("sessionInfo == null");
    if (absStructMsg == null) throw new NullPointerException("absStructMsg == null");
    Method send = null;
    for (Method m : DexKit.doFindClass(DexKit.C_FACADE).getMethods()) {
        if (m.getReturnType().equals(void.class)) {
            Class<?>[] clz = m.getParameterTypes();
            if (clz.length != 3) continue;
            if (clz[0].equals(QQAppInterface.class) && clz[1].equals(_SessionInfo()) && clz[2].isInstance(absStructMsg)) {
                send = m;
                break;
            }
        }
    }
    try {
        send.invoke(null, qqAppInterface, sessionInfo, absStructMsg);
    } catch (Exception e) {
        log(e);
    }
}
 

public static <T extends Externalizable> T deepClone(T origin, ClassLoader classLoader, Map<String, Object> cloningResources) {
    if (origin == null) {
        return null;
    }
    try {
        ByteArrayOutputStream baos = new ByteArrayOutputStream();
        DroolsObjectOutputStream oos = new DroolsObjectOutputStream(baos);
        if ( cloningResources != null ) { cloningResources.forEach( (k, v) -> oos.addCustomExtensions(k, v) ); }
        oos.writeObject(origin);
        ByteArrayInputStream bais = new ByteArrayInputStream(baos.toByteArray());
        DroolsObjectInputStream ois = new DroolsObjectInputStream(bais, classLoader);
        if ( cloningResources != null ) { cloningResources.forEach( (k, v) -> ois.addCustomExtensions(k, v) ); }
        Object deepCopy = ois.readObject();
        return (T)deepCopy;
    } catch(IOException ioe) {
        throw new RuntimeException(ioe);
    } catch (ClassNotFoundException cnfe) {
        throw new RuntimeException(cnfe);
    }
}
 

public <V> DataSet<ExecRow> readKafkaTopic(String topicName, OperationContext context) throws StandardException {
    Properties props = new Properties();
    String consumerGroupId = "spark-consumer-dss-sdsp";
    String bootstrapServers = SIDriver.driver().getConfiguration().getKafkaBootstrapServers();
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
    props.put(ConsumerConfig.GROUP_ID_CONFIG, consumerGroupId);
    props.put(ConsumerConfig.CLIENT_ID_CONFIG, consumerGroupId+"-"+UUID.randomUUID());
    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, IntegerDeserializer.class.getName());
    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ExternalizableDeserializer.class.getName());

    KafkaConsumer<Integer, Externalizable> consumer = new KafkaConsumer<Integer, Externalizable>(props);
    List ps = consumer.partitionsFor(topicName);
    List<Integer> partitions = new ArrayList<>(ps.size());
    for (int i = 0; i < ps.size(); ++i) {
        partitions.add(i);
    }
    consumer.close();

    SparkDataSet rdd = new SparkDataSet(SpliceSpark.getContext().parallelize(partitions, partitions.size()));
    return rdd.flatMap(new KafkaReadFunction(context, topicName, bootstrapServers));
}
 

/**
 *
 */
protected void readExternalizable(String className, Object object) throws ClassNotFoundException, IOException
{
    if (object instanceof Externalizable)
    {
        if (isDebug)
            trace.startExternalizableObject(className, objectTable.size() - 1);

        ((Externalizable)object).readExternal(this);
    }
    else
    {
        //Class '{className}' must implement java.io.Externalizable to receive client IExternalizable instances.
        SerializationException ex = new SerializationException();
        ex.setMessage(10305, new Object[] {object.getClass().getName()});
        throw ex;
    }
}
 

public static byte[] toCompressedBytes(@Nonnull final Externalizable obj,
        @Nonnull final CompressionAlgorithm algo, final boolean bin2txt) throws IOException {
    FastMultiByteArrayOutputStream bos = new FastMultiByteArrayOutputStream();
    OutputStream out = null;
    FinishableOutputStream dos = null;
    try {
        out = bin2txt ? new Base91OutputStream(bos) : bos;
        dos = CompressionStreamFactory.createOutputStream(out, algo);
        toStream(obj, dos);
        dos.finish(); // flush is called
        return bos.toByteArray_clear();
    } finally {
        IOUtils.closeQuietly(dos);
        IOUtils.closeQuietly(out);
    }
}
 

@SneakyThrows
public Object deserialize(AmfReader reader) {
    Object result = cls.newInstance();

    if (result instanceof Externalizable) {
        ((Externalizable) result).readExternal(reader);
        return result;
    }


    for (Map.Entry<String, Object> field : reader.readAmf0KeyValuePairs().entrySet()) {
        setField(result, field.getKey(), field.getValue());
    }

    return result;
}
 

private static boolean usesCustomSerialization(Class<?> cls) {
	try {
		cls.getDeclaredMethod("writeObject", ObjectOutputStream.class);
		return true;
	} catch (NoSuchMethodException ignored) {}

	return Externalizable.class.isAssignableFrom(cls);
}
 

@Override
public Externalizable read(Kryo kryo, Input input, Class<Externalizable> type) {
    try {
        Externalizable e = kryo.newInstance(type);
        KryoObjectInput koi = new KryoObjectInput(input,kryo);
        e.readExternal(koi);
        return e;
    } catch (IOException | ClassNotFoundException e1) {
        throw new RuntimeException(e1);
    }
}
 

public static byte[] toBytes(Externalizable o) throws IOException
{
    ByteArrayOutputStream baos = new ByteArrayOutputStream();
    ObjectOutputStream out = new ObjectOutputStream(baos);
    o.writeExternal(out);
    out.flush();
    return baos.toByteArray();
}
 

@Test
public void testIsCacheSafe() {
	ClassLoader childLoader1 = new ClassLoader(classLoader) {};
	ClassLoader childLoader2 = new ClassLoader(classLoader) {};
	ClassLoader childLoader3 = new ClassLoader(classLoader) {
		@Override
		public Class<?> loadClass(String name) throws ClassNotFoundException {
			return childLoader1.loadClass(name);
		}
	};
	Class<?> composite = ClassUtils.createCompositeInterface(
			new Class<?>[] {Serializable.class, Externalizable.class}, childLoader1);

	assertTrue(ClassUtils.isCacheSafe(String.class, null));
	assertTrue(ClassUtils.isCacheSafe(String.class, classLoader));
	assertTrue(ClassUtils.isCacheSafe(String.class, childLoader1));
	assertTrue(ClassUtils.isCacheSafe(String.class, childLoader2));
	assertTrue(ClassUtils.isCacheSafe(String.class, childLoader3));
	assertFalse(ClassUtils.isCacheSafe(InnerClass.class, null));
	assertTrue(ClassUtils.isCacheSafe(InnerClass.class, classLoader));
	assertTrue(ClassUtils.isCacheSafe(InnerClass.class, childLoader1));
	assertTrue(ClassUtils.isCacheSafe(InnerClass.class, childLoader2));
	assertTrue(ClassUtils.isCacheSafe(InnerClass.class, childLoader3));
	assertFalse(ClassUtils.isCacheSafe(composite, null));
	assertFalse(ClassUtils.isCacheSafe(composite, classLoader));
	assertTrue(ClassUtils.isCacheSafe(composite, childLoader1));
	assertFalse(ClassUtils.isCacheSafe(composite, childLoader2));
	assertTrue(ClassUtils.isCacheSafe(composite, childLoader3));
}
 

/**
 * Initialize a class descriptor from the specified class.
 * @param clazz the class from which to initialize.
 * @param serializing whether we are serializing or deserializing.
 * @throws Exception if any error occurs.
 */
void fillIn(final Class<?> clazz, final boolean serializing) throws Exception {
  populated = true;
  this.clazz = clazz;
  primitive = clazz.isPrimitive();
  enumType = clazz.isEnum();
  if (!primitive && !enumType) {
    externalizable = Externalizable.class.isAssignableFrom(clazz);
    //hasReadWriteObject = handleReadOrWriteObjectMethod(true) && handleReadOrWriteObjectMethod(false);
    hasReadWriteObject = handleReadOrWriteObjectMethod(serializing);
    array = clazz.isArray();
    if (!array) fields = SerializationReflectionHelper.getPersistentDeclaredFields(clazz);
  }
  if (signature == null) signature = SerializationReflectionHelper.getSignatureFromType(clazz);
}
 

private void externalize(final Externalizable original, final Externalizable copy) throws IOException, ClassNotFoundException {
    final ByteArrayOutputStream baos = new ByteArrayOutputStream();
    final ObjectOutputStream out = new ObjectOutputStream(baos);

    original.writeExternal(out);
    out.close();

    final ByteArrayInputStream bais = new ByteArrayInputStream(baos.toByteArray());
    final ObjectInputStream in = new ObjectInputStream(bais);

    copy.readExternal(in);
}
 

/**
 * Returns an accessible no-arg constructor for an externalizable class to be
 * initialized using a public no-argument constructor.
 *
 * @param cl the class to instantiate
 * @return A no-arg constructor for the class; returns {@code null} if
 *     the class does not implement {@link java.io.Externalizable}
 */
public final Constructor<?> newConstructorForExternalization(Class<?> cl) {
    if (!Externalizable.class.isAssignableFrom(cl)) {
        return null;
    }
    try {
        Constructor<?> cons = cl.getConstructor();
        cons.setAccessible(true);
        return cons;
    } catch (NoSuchMethodException ex) {
        return null;
    }
}
 

/**
 * Method to declare various Java types to be handles as JSON array.
 * 
 * @param clazz the type
 * @return <code>true</code> if handles as array
 * @since 1.4
 */
protected boolean isArray(Class clazz) {
    return clazz != null && (clazz.isArray()
        || Collection.class.isAssignableFrom(clazz)
        || Externalizable.class.isAssignableFrom(clazz)
        || Map.class.isAssignableFrom(clazz)
        || Map.Entry.class.isAssignableFrom(clazz));
}
 

/**
 * Returns an accessible no-arg constructor for an externalizable class to be
 * initialized using a public no-argument constructor.
 *
 * @param cl the class to instantiate
 * @return A no-arg constructor for the class; returns {@code null} if
 *     the class does not implement {@link java.io.Externalizable}
 */
public final Constructor<?> newConstructorForExternalization(Class<?> cl) {
    if (!Externalizable.class.isAssignableFrom(cl)) {
        return null;
    }
    try {
        Constructor<?> cons = cl.getConstructor();
        cons.setAccessible(true);
        return cons;
    } catch (NoSuchMethodException ex) {
        return null;
    }
}
 

public final Constructor<?> newConstructorForExternalization(Class<?> cl) {
    if (!Externalizable.class.isAssignableFrom(cl)) {
        return null;
    }
    try {
        Constructor<?> cons = cl.getConstructor();
        cons.setAccessible(true);
        return cons;
    } catch (NoSuchMethodException ex) {
        return null;
    }
}
 

private JApiJavaObjectSerializationCompatibility.JApiJavaObjectSerializationChangeStatus checkChangesForInterfaces(JApiClass jApiClass, JApiJavaObjectSerializationCompatibility.JApiJavaObjectSerializationChangeStatus state) {
	boolean serializableAdded = false;
	boolean serializableRemoved = false;
	boolean serializableUnchanged = false;
	boolean externalizableAdded = false;
	boolean externalizableRemoved = false;
	for (JApiImplementedInterface implementedInterface : jApiClass.getInterfaces()) {
		if (Serializable.class.getCanonicalName().equals(implementedInterface.getFullyQualifiedName())) {
			if (implementedInterface.getChangeStatus() == JApiChangeStatus.NEW) {
				serializableAdded = true;
			} else if (implementedInterface.getChangeStatus() == JApiChangeStatus.REMOVED) {
				serializableRemoved = true;
			} else if (implementedInterface.getChangeStatus() == JApiChangeStatus.UNCHANGED) {
				serializableUnchanged = true;
			}
		}
		if (Externalizable.class.getCanonicalName().equals(implementedInterface.getFullyQualifiedName())) {
			if (implementedInterface.getChangeStatus() == JApiChangeStatus.NEW) {
				externalizableAdded = true;
			} else if (implementedInterface.getChangeStatus() == JApiChangeStatus.REMOVED) {
				externalizableRemoved = true;
			}
		}
	}
	if (serializableRemoved) {
		state = JApiJavaObjectSerializationCompatibility.JApiJavaObjectSerializationChangeStatus.SERIALIZABLE_INCOMPATIBLE_SERIALIZABLE_REMOVED;
	}
	if (externalizableRemoved) {
		state = JApiJavaObjectSerializationCompatibility.JApiJavaObjectSerializationChangeStatus.SERIALIZABLE_INCOMPATIBLE_EXTERNALIZABLE_REMOVED;
	}
	if ((serializableRemoved || serializableUnchanged || serializableAdded) && externalizableAdded) {
		state = JApiJavaObjectSerializationCompatibility.JApiJavaObjectSerializationChangeStatus.SERIALIZABLE_INCOMPATIBLE_CHANGED_FROM_SERIALIZABLE_TO_EXTERNALIZABLE;
	}
	if ((serializableUnchanged || serializableAdded) && externalizableRemoved) {
		state = JApiJavaObjectSerializationCompatibility.JApiJavaObjectSerializationChangeStatus.SERIALIZABLE_INCOMPATIBLE_CHANGED_FROM_EXTERNALIZABLE_TO_SERIALIZABLE;
	}
	return state;
}
 

public final Constructor<?> newConstructorForExternalization(Class<?> cl) {
    if (!Externalizable.class.isAssignableFrom(cl)) {
        return null;
    }
    try {
        Constructor<?> cons = cl.getConstructor();
        cons.setAccessible(true);
        return cons;
    } catch (NoSuchMethodException ex) {
        return null;
    }
}
 

/**
 * Returns an accessible no-arg constructor for an externalizable class to be
 * initialized using a public no-argument constructor.
 *
 * @param cl the class to instantiate
 * @return A no-arg constructor for the class; returns {@code null} if
 *     the class does not implement {@link java.io.Externalizable}
 */
public final Constructor<?> newConstructorForExternalization(Class<?> cl) {
    if (!Externalizable.class.isAssignableFrom(cl)) {
        return null;
    }
    try {
        Constructor<?> cons = cl.getConstructor();
        cons.setAccessible(true);
        return cons;
    } catch (NoSuchMethodException ex) {
        return null;
    }
}
 

@Test
public void correctlyChecksIsInstanceOf()
		throws JClassAlreadyExistsException {

	final JClass arrayList = codeModel.ref("java.util.ArrayList");
	Assert.assertTrue(JClassUtils.isInstanceOf(arrayList, Collection.class));
	final JDefinedClass subArrayList = codeModel._class("SubArrayList");
	subArrayList._extends(arrayList);
	Assert.assertTrue(JClassUtils.isInstanceOf(subArrayList,
			Collection.class));

	final JClass subArrayListOfObjects = subArrayList.narrow(Object.class);
	Assert.assertTrue(JClassUtils.isInstanceOf(subArrayListOfObjects,
			Collection.class));

	final JDefinedClass subExternalizable = codeModel
			._class("SubExternalizable");
	subExternalizable._implements(Externalizable.class);
	Assert.assertTrue(JClassUtils.isInstanceOf(subExternalizable,
			Externalizable.class));

	subArrayList._implements(subExternalizable);
	Assert.assertTrue(JClassUtils.isInstanceOf(subArrayList,
			Externalizable.class));

	Assert.assertFalse(JClassUtils.isInstanceOf(codeModel.NULL,
			Collection.class));

}
 

/**
 * @throws Exception If failed.
 */
@Test
public void testSerializationTopicCreatedByStrinAndLong() throws Exception {
    for (Marshaller marsh : getMarshallers()) {
        info("Test GridTopic externalization [marshaller=" + marsh + ']');

        for (GridTopic topic : GridTopic.values()) {
            Externalizable msgOut = (Externalizable)topic.topic(A_STRING, A_LONG);

            assertEquals(msgOut, GridTestIoUtils.externalize(msgOut, marsh));
        }
    }
}
 

@Override
public void write(Kryo kryo, Output output, Externalizable object) {
    KryoObjectOutput koo = new KryoObjectOutput(output, kryo);
    try {
        object.writeExternal(koo);
    } catch (IOException e) {
        //shouldn't happen
        throw new RuntimeException(e);
    }
}
 

/**
 * @throws Exception If failed.
 */
@Test
public void testSerializationTopicCreatedByStringAndUUIDAndLong() throws Exception {
    for (Marshaller marsh : getMarshallers()) {
        info("Test GridTopic externalization [marshaller=" + marsh + ']');

        for (GridTopic topic : GridTopic.values()) {
            Externalizable msgOut = (Externalizable)topic.topic(A_STRING, AN_UUID, A_LONG);

            assertEquals(msgOut, GridTestIoUtils.externalize(msgOut, marsh));
        }
    }
}
 

/**
 * Returns an accessible no-arg constructor for an externalizable class to be
 * initialized using a public no-argument constructor.
 *
 * @param cl the class to instantiate
 * @return A no-arg constructor for the class; returns {@code null} if
 *     the class does not implement {@link java.io.Externalizable}
 */
public final Constructor<?> newConstructorForExternalization(Class<?> cl) {
    if (!Externalizable.class.isAssignableFrom(cl)) {
        return null;
    }
    try {
        Constructor<?> cons = cl.getConstructor();
        cons.setAccessible(true);
        return cons;
    } catch (NoSuchMethodException ex) {
        return null;
    }
}
 

public static byte[] toCompressedBytes(@Nonnull final Externalizable obj) throws IOException {
    FastMultiByteArrayOutputStream bos = new FastMultiByteArrayOutputStream();
    DeflaterOutputStream dos = new DeflaterOutputStream(bos);
    try {
        toStream(obj, dos);
        dos.finish();
        dos.flush();
        return bos.toByteArray_clear();
    } finally {
        IOUtils.closeQuietly(dos);
    }
}
 

public static void readCompressedObject(@Nonnull final byte[] src,
        @Nonnull final Externalizable dst) throws IOException, ClassNotFoundException {
    FastByteArrayInputStream bis = new FastByteArrayInputStream(src);
    final InflaterInputStream iis = new InflaterInputStream(bis);
    try {
        readObject(iis, dst);
    } finally {
        IOUtils.closeQuietly(iis);
    }
}
 

public static void readCompressedObject(@Nonnull final byte[] src, final int len,
        @Nonnull final Externalizable dst, @Nonnull final CompressionAlgorithm algo,
        final boolean bin2txt) throws IOException, ClassNotFoundException {
    FastByteArrayInputStream bis = new FastByteArrayInputStream(src, len);
    InputStream in = null;
    InputStream compressedStream = null;
    try {
        in = bin2txt ? new Base91InputStream(bis) : bis;
        compressedStream = CompressionStreamFactory.createInputStream(in, algo);
        readObject(compressedStream, dst);
    } finally {
        IOUtils.closeQuietly(compressedStream);
        IOUtils.closeQuietly(in);
    }
}
 

/**
 * @throws Exception If failed.
 */
@Test
public void testSerializationTopicCreatedByString() throws Exception {
    for (Marshaller marsh : getMarshallers()) {
        info("Test GridTopic externalization [marshaller=" + marsh + ']');

        for (GridTopic topic : GridTopic.values()) {
            Externalizable msgOut = (Externalizable)topic.topic(A_STRING);

            assertEquals(msgOut, GridTestIoUtils.externalize(msgOut, marsh));
        }
    }
}
 

@Test
public void testGet() {
    InterfaceMap<Object> map = new InterfaceMap<Object>();
    A a = new A();
    B b = new B();
    map.put(a, a);
    map.put(b, b);
    assertSame(a, map.get(Runnable.class));
    assertSame(b, map.get(Externalizable.class));
    assertSame(b, map.get(Comparable.class));
}