spark/serializer/DummySerializerInstance.java

0001 /*
0002  * Licensed to the Apache Software Foundation (ASF) under one or more
0003  * contributor license agreements.  See the NOTICE file distributed with
0004  * this work for additional information regarding copyright ownership.
0005  * The ASF licenses this file to You under the Apache License, Version 2.0
0006  * (the "License"); you may not use this file except in compliance with
0007  * the License.  You may obtain a copy of the License at
0008  *
0009  *    http://www.apache.org/licenses/LICENSE-2.0
0010  *
0011  * Unless required by applicable law or agreed to in writing, software
0012  * distributed under the License is distributed on an "AS IS" BASIS,
0013  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
0014  * See the License for the specific language governing permissions and
0015  * limitations under the License.
0016  */
0017
0018 package org.apache.spark.serializer;
0019
0020 import java.io.IOException;
0021 import java.io.InputStream;
0022 import java.io.OutputStream;
0023 import java.nio.ByteBuffer;
0024
0025 import scala.reflect.ClassTag;
0026
0027 import org.apache.spark.annotation.Private;
0028 import org.apache.spark.unsafe.Platform;
0029
0030 /**
0031  * Unfortunately, we need a serializer instance in order to construct a DiskBlockObjectWriter.
0032  * Our shuffle write path doesn't actually use this serializer (since we end up calling the
0033  * `write() OutputStream methods), but DiskBlockObjectWriter still calls some methods on it. To work
0034  * around this, we pass a dummy no-op serializer.
0035  */
0036 @Private
0037 public final class DummySerializerInstance extends SerializerInstance {
0038
0039   public static final DummySerializerInstance INSTANCE = new DummySerializerInstance();
0040
0041   private DummySerializerInstance() { }
0042
0043   @Override
0044   public SerializationStream serializeStream(final OutputStream s) {
0045     return new SerializationStream() {
0046       @Override
0047       public void flush() {
0048         // Need to implement this because DiskObjectWriter uses it to flush the compression stream
0049         try {
0050           s.flush();
0051         } catch (IOException e) {
0052           Platform.throwException(e);
0053         }
0054       }
0055
0056       @Override
0057       public <T> SerializationStream writeObject(T t, ClassTag<T> ev1) {
0058         throw new UnsupportedOperationException();
0059       }
0060
0061       @Override
0062       public void close() {
0063         // Need to implement this because DiskObjectWriter uses it to close the compression stream
0064         try {
0065           s.close();
0066         } catch (IOException e) {
0067           Platform.throwException(e);
0068         }
0069       }
0070     };
0071   }
0072
0073   @Override
0074   public <T> ByteBuffer serialize(T t, ClassTag<T> ev1) {
0075     throw new UnsupportedOperationException();
0076   }
0077
0078   @Override
0079   public DeserializationStream deserializeStream(InputStream s) {
0080     throw new UnsupportedOperationException();
0081   }
0082
0083   @Override
0084   public <T> T deserialize(ByteBuffer bytes, ClassLoader loader, ClassTag<T> ev1) {
0085     throw new UnsupportedOperationException();
0086   }
0087
0088   @Override
0089   public <T> T deserialize(ByteBuffer bytes, ClassTag<T> ev1) {
0090     throw new UnsupportedOperationException();
0091   }
0092 }