@Override
  public Collection<StoreDescriptor> getStoreDescriptors() {
    String rocksDBStoreFactory = "org.apache.samza.storage.kv.RocksDbKeyValueStorageEngineFactory";
    String leftStoreName = getLeftOpId();
    String rightStoreName = getRightOpId();
    Map<String, String> leftStoreCustomProps = ImmutableMap.of(
        String.format("stores.%s.rocksdb.ttl.ms", leftStoreName), Long.toString(ttlMs),
        String.format("stores.%s.changelog.kafka.cleanup.policy", leftStoreName), "delete",
        String.format("stores.%s.changelog.kafka.retention.ms", leftStoreName), Long.toString(ttlMs));
    Map<String, String> rightStoreCustomProps = ImmutableMap.of(
        String.format("stores.%s.rocksdb.ttl.ms", rightStoreName), Long.toString(ttlMs),
        String.format("stores.%s.changelog.kafka.cleanup.policy", rightStoreName), "delete",
        String.format("stores.%s.changelog.kafka.retention.ms", rightStoreName), Long.toString(ttlMs));

    return Arrays.asList(
        new StoreDescriptor(leftStoreName, rocksDBStoreFactory, this.keySerde, this.messageSerde,
            leftStoreName, leftStoreCustomProps),
        new StoreDescriptor(rightStoreName, rocksDBStoreFactory, this.keySerde, this.otherMessageSerde,
            rightStoreName, rightStoreCustomProps));
  }
Default constructor for a {@link JoinOperatorSpec}.

@param leftInputOpSpec  the operator spec for the stream on the left side of the join
@param rightInputOpSpec  the operator spec for the stream on the right side of the join
@param joinFn  the user-defined join function to get join keys and results
@param ttlMs  the ttl in ms for retaining messages in each stream
@param opId  the unique ID for this operator
samza-core/src/main/java/org/apache/samza/operators/spec/JoinOperatorSpec.java