Merge pull request #958: [proxima-direct-io-cassandra] unsynchronize CassandraWriter

Author: je-ik

direct/core/src/main/java/cz/o2/proxima/direct/core/DirectDataOperator.java

@@ -15,6 +15,7 @@
  */
 package cz.o2.proxima.direct.core;
 
+import cz.o2.proxima.core.annotations.DeclaredThreadSafe;
 import cz.o2.proxima.core.annotations.Internal;
 import cz.o2.proxima.core.functional.Factory;
 import cz.o2.proxima.core.functional.UnaryFunction;
@@ -360,7 +361,11 @@ private void cacheOrRetrieveWriterFor(DirectAttributeFamilyDescriptor af) {
       // store writer of this family to all attributes
       for (AttributeDescriptor<?> a : af.getAttributes()) {
         if (a.getTransactionMode() == TransactionMode.NONE) {
-          writers.put(a, OnlineAttributeWriters.synchronizedWriter(familyWriter));
+          if (familyWriter.getClass().getAnnotation(DeclaredThreadSafe.class) == null) {
+            writers.put(a, OnlineAttributeWriters.synchronizedWriter(familyWriter));
+          } else {
+            writers.put(a, familyWriter);
+          }
         } else {
           writers.put(a, maybeTransactionalWriter);
         }

direct/core/src/main/java/cz/o2/proxima/direct/core/OnlineAttributeWriters.java

@@ -41,7 +41,7 @@ public synchronized Factory<? extends OnlineAttributeWriter> asFactory() {
       }
 
       @Override
-      public synchronized URI getUri() {
+      public URI getUri() {
         return delegate.getUri();
       }
 

direct/io-cassandra/src/main/java/cz/o2/proxima/direct/io/cassandra/CassandraWriter.java

@@ -17,6 +17,7 @@
 
 import com.datastax.driver.core.BoundStatement;
 import com.datastax.driver.core.Session;
+import cz.o2.proxima.core.annotations.DeclaredThreadSafe;
 import cz.o2.proxima.core.storage.StreamElement;
 import cz.o2.proxima.direct.core.AbstractOnlineAttributeWriter;
 import cz.o2.proxima.direct.core.CommitCallback;
@@ -27,6 +28,7 @@
 
 /** A {@link OnlineAttributeWriter} implementation for Cassandra. */
 @Slf4j
+@DeclaredThreadSafe
 class CassandraWriter extends AbstractOnlineAttributeWriter implements OnlineAttributeWriter {
 
   private final CassandraDBAccessor accessor;
@@ -39,8 +41,7 @@ class CassandraWriter extends AbstractOnlineAttributeWriter implements OnlineAtt
   }
 
   @Override
-  public synchronized void write(StreamElement data, CommitCallback statusCallback) {
-
+  public void write(StreamElement data, CommitCallback statusCallback) {
     try {
       Session session = accessor.ensureSession();
       Optional<BoundStatement> cql = accessor.getCqlFactory().getWriteStatement(data, session);