From a91518c97beca7f108b4c085721391e0cefbb7a5 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Mon, 14 Dec 2015 18:15:26 -0500 Subject: [PATCH] INT-3907: Downgrade to Kryo-2.22 JIRA: https://jira.spring.io/browse/INT-3907 The IO-1.1.x is based on the Kryo-2.22 and can't be upgraded to 3.0. Therefore we should downgrade. * Change `PojoCodec` to use `StdInstantiatorStrategy` directly, because the `DefaultInstantiatorStrategy` logic is as a core code of `Kryo`. * Introduce `org.springframework.integration.codec.kryo.pool` package and copy/paste `com.esotericsoftware.kryo.pool` classes, since they have been introduced since Kryo-3.0. * That copy/paste seemed to me the simplest fix, since the `KryoPool` logic is encapsulated in the `AbstractKryoCodec` --- build.gradle | 4 +- .../codec/kryo/AbstractKryoCodec.java | 6 +- .../integration/codec/kryo/PojoCodec.java | 5 +- .../codec/kryo/pool/KryoCallback.java | 34 ++++ .../codec/kryo/pool/KryoFactory.java | 32 ++++ .../integration/codec/kryo/pool/KryoPool.java | 143 ++++++++++++++++ .../codec/kryo/pool/KryoPoolQueueImpl.java | 74 +++++++++ .../codec/kryo/pool/SoftReferenceQueue.java | 156 ++++++++++++++++++ .../codec/kryo/pool/package-info.java | 5 + 9 files changed, 451 insertions(+), 8 deletions(-) create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/codec/kryo/pool/KryoCallback.java create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/codec/kryo/pool/KryoFactory.java create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/codec/kryo/pool/KryoPool.java create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/codec/kryo/pool/KryoPoolQueueImpl.java create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/codec/kryo/pool/SoftReferenceQueue.java create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/codec/kryo/pool/package-info.java diff --git a/build.gradle b/build.gradle index 94360fb1d0..35026281e8 100644 --- a/build.gradle +++ b/build.gradle @@ -106,7 +106,7 @@ subprojects { subproject -> jsonpathVersion = '0.9.1' junitVersion = '4.11' jythonVersion = '2.5.3' - kryoShadedVersion = '3.0.3' + kryoVersion = '2.22' log4jVersion = '1.2.17' mockitoVersion = '1.9.5' mysqlVersion = '5.1.29' @@ -268,7 +268,7 @@ project('spring-integration-core') { compile("com.fasterxml.jackson.core:jackson-databind:$jackson2Version", optional) compile("com.jayway.jsonpath:json-path:$jsonpathVersion", optional) compile("io.fastjson:boon:$boonVersion", optional) - compile("com.esotericsoftware:kryo-shaded:$kryoShadedVersion", optional) + compile("com.esotericsoftware.kryo:kryo:$kryoVersion", optional) testCompile ("org.aspectj:aspectjweaver:$aspectjVersion") testCompile ("net.openhft:chronicle:$chronicleVersion") diff --git a/spring-integration-core/src/main/java/org/springframework/integration/codec/kryo/AbstractKryoCodec.java b/spring-integration-core/src/main/java/org/springframework/integration/codec/kryo/AbstractKryoCodec.java index 887d60752e..2a4f348bd2 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/codec/kryo/AbstractKryoCodec.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/codec/kryo/AbstractKryoCodec.java @@ -22,14 +22,14 @@ import java.io.InputStream; import java.io.OutputStream; import org.springframework.integration.codec.Codec; +import org.springframework.integration.codec.kryo.pool.KryoCallback; +import org.springframework.integration.codec.kryo.pool.KryoFactory; +import org.springframework.integration.codec.kryo.pool.KryoPool; import org.springframework.util.Assert; import com.esotericsoftware.kryo.Kryo; import com.esotericsoftware.kryo.io.Input; import com.esotericsoftware.kryo.io.Output; -import com.esotericsoftware.kryo.pool.KryoCallback; -import com.esotericsoftware.kryo.pool.KryoFactory; -import com.esotericsoftware.kryo.pool.KryoPool; /** * Base class for {@link Codec}s using {@link Kryo}. diff --git a/spring-integration-core/src/main/java/org/springframework/integration/codec/kryo/PojoCodec.java b/spring-integration-core/src/main/java/org/springframework/integration/codec/kryo/PojoCodec.java index 969673d5a1..28869ca341 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/codec/kryo/PojoCodec.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/codec/kryo/PojoCodec.java @@ -19,13 +19,12 @@ package org.springframework.integration.codec.kryo; import java.util.Collections; import java.util.List; -import org.objenesis.strategy.StdInstantiatorStrategy; - import org.springframework.util.CollectionUtils; import com.esotericsoftware.kryo.Kryo; import com.esotericsoftware.kryo.io.Input; import com.esotericsoftware.kryo.io.Output; +import com.esotericsoftware.shaded.org.objenesis.strategy.StdInstantiatorStrategy; /** * Kryo Codec that can encode and decode arbitrary types. Classes and associated @@ -99,7 +98,7 @@ public class PojoCodec extends AbstractKryoCodec { @Override protected void configureKryoInstance(Kryo kryo) { - kryo.setInstantiatorStrategy(new Kryo.DefaultInstantiatorStrategy(new StdInstantiatorStrategy())); + kryo.setInstantiatorStrategy(new StdInstantiatorStrategy()); if (this.kryoRegistrar != null) { this.kryoRegistrar.registerTypes(kryo); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/codec/kryo/pool/KryoCallback.java b/spring-integration-core/src/main/java/org/springframework/integration/codec/kryo/pool/KryoCallback.java new file mode 100644 index 0000000000..c58e8f3a3a --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/codec/kryo/pool/KryoCallback.java @@ -0,0 +1,34 @@ +/* + * Copyright 2015 the original author or authors. + * 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.springframework.integration.codec.kryo.pool; + +import com.esotericsoftware.kryo.Kryo; + +/** + * Callback to run with a provided kryo instance. + *

+ * Copied from Kryo 3.0 to facilitate dropping back to 2.22 for Spring IO + * compatibility. + * + * @author Martin Grotzke + * + * @param The type of the result of the interaction with kryo. + */ +public interface KryoCallback { + + T execute(Kryo kryo); + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/codec/kryo/pool/KryoFactory.java b/spring-integration-core/src/main/java/org/springframework/integration/codec/kryo/pool/KryoFactory.java new file mode 100644 index 0000000000..6b45fef9cf --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/codec/kryo/pool/KryoFactory.java @@ -0,0 +1,32 @@ +/* + * Copyright 2015 the original author or authors. + * 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.springframework.integration.codec.kryo.pool; + +import com.esotericsoftware.kryo.Kryo; + +/** + * Factory to create new configured instances of {@link Kryo}. + *

+ * Copied from Kryo 3.0 to facilitate dropping back to 2.22 for Spring IO + * compatibility. + * + * @author Martin Grotzke + */ +public interface KryoFactory { + + Kryo create(); + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/codec/kryo/pool/KryoPool.java b/spring-integration-core/src/main/java/org/springframework/integration/codec/kryo/pool/KryoPool.java new file mode 100644 index 0000000000..41067ec884 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/codec/kryo/pool/KryoPool.java @@ -0,0 +1,143 @@ +/* + * Copyright 2015 the original author or authors. + * 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.springframework.integration.codec.kryo.pool; + +import java.lang.ref.SoftReference; +import java.util.Queue; +import java.util.concurrent.ConcurrentLinkedQueue; + +import com.esotericsoftware.kryo.Kryo; + +/** + * A simple pool interface for {@link Kryo} instances. Use the {@link KryoPool.Builder} to + * construct a pool instance. + *

+ * Copied from Kryo 3.0 to facilitate dropping back to 2.22 for Spring IO + * compatibility. + *

+ * Usage: + * 

+ * {@code
+ *
+ * KryoFactory factory = new KryoFactory() {
+ *   public Kryo create () {
+ *     Kryo kryo = new Kryo();
+ *     // configure kryo instance, customize settings
+ *     return kryo;
+ *   }
+ * };
+ * // Simple pool, you might also activate SoftReferences to fight OOMEs.
+ * KryoPool pool = new KryoPool.Builder(factory).build();
+ * Kryo kryo = pool.borrow();
+ * // do s.th. with kryo here, and afterwards release it
+ * pool.release(kryo);
+ *
+ * // or use a callback to work with kryo (pool.run borrows+releases for you)
+ * String value = pool.run(new KryoCallback() {
+ *   public String execute(Kryo kryo) {
+ *     return kryo.readObject(input, String.class);
+ *   }
+ * });
+ *
+ * }
+ *
+ * @author Martin Grotzke + */ +public interface KryoPool { + + /** + * Takes a {@link Kryo} instance from the pool or creates a new one + * (using the factory) if the pool is empty. + * @return the {@link Kryo} instance. + */ + Kryo borrow(); + + /** + * Returns the given {@link Kryo} instance to the pool. + * @param kryo the {@link Kryo} instance to release. + */ + void release(Kryo kryo); + + /** + * Runs the provided {@link KryoCallback} with a {@link Kryo} instance + * from the pool (borrow/release around {@link KryoCallback#execute(Kryo)}). + * @param callback the {@link KryoCallback} to call. + * @param The type of the result of the interaction with kryo. + * @return the result of the interaction with {@link Kryo}. + */ + T run(KryoCallback callback); + + /** + * Builder for a {@link KryoPool} instance, constructs a {@link KryoPoolQueueImpl} instance. + */ + class Builder { + + private final KryoFactory factory; + + private Queue queue = new ConcurrentLinkedQueue(); + + private boolean softReferences; + + public Builder(KryoFactory factory) { + if (factory == null) { + throw new IllegalArgumentException("factory must not be null"); + } + this.factory = factory; + } + + /** + * Use the given queue for pooling kryo instances (by default a {@link ConcurrentLinkedQueue} + * is used). + * @param queue the queue instance to use. + * @return the Builder instance. + */ + public Builder queue(Queue queue) { + if (queue == null) { + throw new IllegalArgumentException("queue must not be null"); + } + this.queue = queue; + return this; + } + + /** + * Use {@link SoftReference}s for pooled {@link Kryo} instances, so that + * instances may be garbage collected when there's memory demand (by default + * disabled). + * @return the Builder instance. + */ + public Builder softReferences() { + softReferences = true; + return this; + } + + /** + * Build the pool. + * @return the KryoPool instance. + */ + public KryoPool build() { + Queue q = softReferences ? new SoftReferenceQueue(queue) : queue; + return new KryoPoolQueueImpl(factory, q); + } + + @Override + public String toString() { + return getClass().getName() + "[queue.class=" + queue.getClass() + + ", softReferences=" + softReferences + "]"; + } + + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/codec/kryo/pool/KryoPoolQueueImpl.java b/spring-integration-core/src/main/java/org/springframework/integration/codec/kryo/pool/KryoPoolQueueImpl.java new file mode 100644 index 0000000000..d615379fce --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/codec/kryo/pool/KryoPoolQueueImpl.java @@ -0,0 +1,74 @@ +/* + * Copyright 2015 the original author or authors. + * 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.springframework.integration.codec.kryo.pool; + +import java.util.Queue; + +import com.esotericsoftware.kryo.Kryo; + +/** + * A simple {@link Queue} based {@link KryoPool} implementation, should be built + * using the KryoPool.Builder. + *

+ * Copied from Kryo 3.0 to facilitate dropping back to 2.22 for Spring IO + * compatibility. + * @author Martin Grotzke + */ +class KryoPoolQueueImpl implements KryoPool { + + private final Queue queue; + + private final KryoFactory factory; + + KryoPoolQueueImpl(KryoFactory factory, Queue queue) { + this.factory = factory; + this.queue = queue; + } + + public int size() { + return queue.size(); + } + + @Override + public Kryo borrow() { + Kryo res; + if ((res = queue.poll()) != null) { + return res; + } + return factory.create(); + } + + @Override + public void release(Kryo kryo) { + queue.offer(kryo); + } + + @Override + public T run(KryoCallback callback) { + Kryo kryo = borrow(); + try { + return callback.execute(kryo); + } + finally { + release(kryo); + } + } + + public void clear() { + queue.clear(); + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/codec/kryo/pool/SoftReferenceQueue.java b/spring-integration-core/src/main/java/org/springframework/integration/codec/kryo/pool/SoftReferenceQueue.java new file mode 100644 index 0000000000..a863e57f29 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/codec/kryo/pool/SoftReferenceQueue.java @@ -0,0 +1,156 @@ +/* + * Copyright 2015 the original author or authors. + * 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.springframework.integration.codec.kryo.pool; + +import java.lang.ref.SoftReference; +import java.util.Collection; +import java.util.Iterator; +import java.util.Queue; + +import com.esotericsoftware.kryo.Kryo; + +/** + * Internally uses {@link SoftReference}s for queued Kryo instances, + * most importantly adjusts the {@link Queue#poll() poll} + * behavior so that gc'ed Kryo instances are skipped. + * Most other methods are unsupported. + *

+ * Copied from Kryo 3.0 to facilitate dropping back to 2.22 for Spring IO + * compatibility. + * @author Martin Grotzke + */ +class SoftReferenceQueue implements Queue { + + private final Queue> delegate; + + @SuppressWarnings("unchecked") + public SoftReferenceQueue(Queue delegate) { + this.delegate = (Queue>) delegate; + } + + @Override + public Kryo poll() { + Kryo res; + SoftReference ref; + while ((ref = delegate.poll()) != null) { + if ((res = ref.get()) != null) { + return res; + } + } + return null; + } + + @Override + public boolean offer(Kryo e) { + return delegate.offer(new SoftReference(e)); + } + + @Override + public boolean add(Kryo e) { + return delegate.add(new SoftReference(e)); + } + + @Override + public int size() { + return delegate.size(); + } + + @Override + public boolean isEmpty() { + return delegate.isEmpty(); + } + + @Override + public boolean contains(Object o) { + return delegate.contains(o); + } + + @Override + public void clear() { + delegate.clear(); + } + + @Override + public boolean equals(Object o) { + return delegate.equals(o); + } + + @Override + public int hashCode() { + return delegate.hashCode(); + } + + @Override + public String toString() { + return getClass().getSimpleName() + super.toString(); + } + + @Override + public Iterator iterator() { + throw new UnsupportedOperationException(); + } + + @Override + public Kryo remove() { + throw new UnsupportedOperationException(); + } + + @Override + public Object[] toArray() { + throw new UnsupportedOperationException(); + } + + @Override + public Kryo element() { + throw new UnsupportedOperationException(); + } + + @Override + public Kryo peek() { + throw new UnsupportedOperationException(); + } + + @Override + public T[] toArray(T[] a) { + throw new UnsupportedOperationException(); + } + + @Override + public boolean remove(Object o) { + throw new UnsupportedOperationException(); + } + + @Override + public boolean containsAll(Collection c) { + throw new UnsupportedOperationException(); + } + + @Override + public boolean addAll(Collection c) { + throw new UnsupportedOperationException(); + } + + @Override + public boolean removeAll(Collection c) { + throw new UnsupportedOperationException(); + } + + @Override + public boolean retainAll(Collection c) { + throw new UnsupportedOperationException(); + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/codec/kryo/pool/package-info.java b/spring-integration-core/src/main/java/org/springframework/integration/codec/kryo/pool/package-info.java new file mode 100644 index 0000000000..240fd0f979 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/codec/kryo/pool/package-info.java @@ -0,0 +1,5 @@ +/** + * The Kryo Pool classes. + * Copy/paste from Kryo 3.0 + */ +package org.springframework.integration.codec.kryo.pool;