Repository: weaviate/weaviate
Pull request #3084 (2023-06-01 23:43:53)

Adjust the backup and restoration process to accommodate multi-tenant classes
### What's being changed:

- validation specific to adding a new class abstracted out of generic validation func, to accommodate class restoration from a backup
- multi-tenant single node and cluster backup journey tests added

### Review checklist

- [ ] Documentation has been updated, if necessary. Link to changed documentation:
- [ ] Chaos pipeline run or not necessary. Link to pipeline:
- [ ] All new code is covered by tests where it is reasonable.
- [ ] Performance tests have been run or not necessary.



Patch:
diff --git a/adapters/handlers/rest/embedded_spec.go b/adapters/handlers/rest/embedded_spec.go
index b5be94fd81..3b576ec379 100644
--- a/adapters/handlers/rest/embedded_spec.go
+++ b/adapters/handlers/rest/embedded_spec.go
@@ -437,6 +437,9 @@ func init() {
           },
           {
             "$ref": "#/parameters/CommonConsistencyLevelParameterQuery"
+          },
+          {
+            "$ref": "#/parameters/CommonTenantKeyParameterQuery"
           }
         ],
         "responses": {
@@ -502,6 +505,9 @@ func init() {
           },
           {
             "$ref": "#/parameters/CommonConsistencyLevelParameterQuery"
+          },
+          {
+            "$ref": "#/parameters/CommonTenantKeyParameterQuery"
           }
         ],
         "responses": {
@@ -4865,6 +4871,12 @@ func init() {
             "description": "Determines how many replicas must acknowledge a request before it is considered successful",
             "name": "consistency_level",
             "in": "query"
+          },
+          {
+            "type": "string",
+            "description": "Specifies the tenant in a request targeting a multi-tenant class",
+            "name": "tenant_key",
+            "in": "query"
           }
         ],
         "responses": {
@@ -4933,6 +4945,12 @@ func init() {
             "description": "Determines how many replicas must acknowledge a request before it is considered successful",
             "name": "consistency_level",
             "in": "query"
+          },
+          {
+            "type": "string",
+            "description": "Specifies the tenant in a request targeting a multi-tenant class",
+            "name": "tenant_key",
+            "in": "query"
           }
         ],
         "responses": {
diff --git a/adapters/handlers/rest/handlers_batch_objects.go b/adapters/handlers/rest/handlers_batch_objects.go
index e431514890..2aa3bba866 100644
--- a/adapters/handlers/rest/handlers_batch_objects.go
+++ b/adapters/handlers/rest/handlers_batch_objects.go
@@ -12,12 +12,14 @@
 package rest
 
 import (
+	"errors"
+
 	middleware "github.com/go-openapi/runtime/middleware"
 	"github.com/go-openapi/strfmt"
 	"github.com/weaviate/weaviate/adapters/handlers/rest/operations"
 	"github.com/weaviate/weaviate/adapters/handlers/rest/operations/batch"
 	"github.com/weaviate/weaviate/entities/models"
-	"github.com/weaviate/weaviate/usecases/auth/authorization/errors"
+	autherrs "github.com/weaviate/weaviate/usecases/auth/authorization/errors"
 	"github.com/weaviate/weaviate/usecases/objects"
 )
 
@@ -34,11 +36,13 @@ func (h *batchObjectHandlers) addObjects(params batch.BatchObjectsCreateParams,
 			WithPayload(errPayloadFromSingleErr(err))
 	}
 
-	objs, err := h.manager.AddObjects(params.HTTPRequest.Context(),
-		principal, params.Body.Objects, params.Body.Fields, repl)
+	tenantKey := getTenantKey(params.TenantKey)
+
+	objs, err := h.manager.AddObjects(params.HTTPRequest.Context(), principal,
+		params.Body.Objects, params.Body.Fields, repl, tenantKey)
 	if err != nil {
 		switch err.(type) {
-		case errors.Forbidden:
+		case autherrs.Forbidden:
 			return batch.NewBatchObjectsCreateForbidden().
 				WithPayload(errPayloadFromSingleErr(err))
 		case objects.ErrInvalidUserInput:
@@ -89,7 +93,7 @@ func (h *batchObjectHandlers) addReferences(params batch.BatchReferencesCreatePa
 		params.HTTPRequest.Context(), principal, params.Body, repl, tenantKey)
 	if err != nil {
 		switch err.(type) {
-		case errors.Forbidden:
+		case autherrs.Forbidden:
 			return batch.NewBatchReferencesCreateForbidden().
 				WithPayload(errPayloadFromSingleErr(err))
 		case objects.ErrInvalidUserInput:
@@ -141,17 +145,18 @@ func (h *batchObjectHandlers) deleteObjects(params batch.BatchObjectsDeleteParam
 			WithPayload(errPayloadFromSingleErr(err))
 	}
 
-	res, err := h.manager.DeleteObjects(params.HTTPRequest.Context(),
-		principal, params.Body.Match, params.Body.DryRun, params.Body.Output, repl)
+	tenantKey := getTenantKey(params.TenantKey)
+
+	res, err := h.manager.DeleteObjects(params.HTTPRequest.Context(), principal,
+		params.Body.Match, params.Body.DryRun, params.Body.Output, repl, tenantKey)
 	if err != nil {
-		switch err.(type) {
-		case errors.Forbidden:
-			return batch.NewBatchObjectsDeleteForbidden().
-				WithPayload(errPayloadFromSingleErr(err))
-		case objects.ErrInvalidUserInput:
+		if errors.As(err, &objects.ErrInvalidUserInput{}) {
 			return batch.NewBatchObjectsDeleteUnprocessableEntity().
 				WithPayload(errPayloadFromSingleErr(err))
-		default:
+		} else if errors.As(err, &autherrs.Forbidden{}) {
+			return batch.NewBatchObjectsDeleteForbidden().
+				WithPayload(errPayloadFromSingleErr(err))
+		} else {
 			return batch.NewBatchObjectsDeleteInternalServerError().
 				WithPayload(errPayloadFromSingleErr(err))
 		}
diff --git a/adapters/handlers/rest/handlers_objects.go b/adapters/handlers/rest/handlers_objects.go
index 62dbf9db9b..3e7ed7a76f 100644
--- a/adapters/handlers/rest/handlers_objects.go
+++ b/adapters/handlers/rest/handlers_objects.go
@@ -13,6 +13,7 @@ package rest
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"strings"
 
@@ -24,7 +25,7 @@ import (
 	"github.com/weaviate/weaviate/entities/additional"
 	"github.com/weaviate/weaviate/entities/models"
 	"github.com/weaviate/weaviate/entities/schema/crossref"
-	"github.com/weaviate/weaviate/usecases/auth/authorization/errors"
+	autherrs "github.com/weaviate/weaviate/usecases/auth/authorization/errors"
 	"github.com/weaviate/weaviate/usecases/config"
 	uco "github.com/weaviate/weaviate/usecases/objects"
 	"github.com/weaviate/weaviate/usecases/replica"
@@ -85,14 +86,13 @@ func (h *objectHandlers) addObject(params objects.ObjectsCreateParams,
 	object, err := h.manager.AddObject(params.HTTPRequest.Context(),
 		principal, params.Body, repl, tenantKey)
 	if err != nil {
-		switch err.(type) {
-		case errors.Forbidden:
-			return objects.NewObjectsCreateForbidden().
-				WithPayload(errPayloadFromSingleErr(err))
-		case uco.ErrInvalidUserInput:
+		if errors.As(err, &uco.ErrInvalidUserInput{}) {
 			return objects.NewObjectsCreateUnprocessableEntity().
 				WithPayload(errPayloadFromSingleErr(err))
-		default:
+		} else if errors.As(err, &autherrs.Forbidden{}) {
+			return objects.NewObjectsCreateForbidden().
+				WithPayload(errPayloadFromSingleErr(err))
+		} else {
 			return objects.NewObjectsCreateInternalServerError().
 				WithPayload(errPayloadFromSingleErr(err))
 		}
@@ -112,7 +112,7 @@ func (h *objectHandlers) validateObject(params objects.ObjectsValidateParams,
 	err := h.manager.ValidateObject(params.HTTPRequest.Context(), principal, params.Body, nil)
 	if err != nil {
 		switch err.(type) {
-		case errors.Forbidden:
+		case autherrs.Forbidden:
 			return objects.NewObjectsValidateForbidden().
 				WithPayload(errPayloadFromSingleErr(err))
 		case uco.ErrInvalidUserInput:
@@ -165,7 +165,7 @@ func (h *objectHandlers) getObject(params objects.ObjectsClassGetParams,
 		params.ClassName, params.ID, additional, replProps, tenantKey)
 	if err != nil {
 		switch err.(type) {
-		case errors.Forbidden:
+		case autherrs.Forbidden:
 			return objects.NewObjectsClassGetForbidden().
 				WithPayload(errPayloadFromSingleErr(err))
 		case uco.ErrNotFound:
@@ -202,7 +202,7 @@ func (h *objectHandlers) getObjects(params objects.ObjectsListParams,
 		params.Offset, params.Limit, params.Sort, params.Order, params.After, additional)
 	if err != nil {
 		switch err.(type) {
-		case errors.Forbidden:
+		case autherrs.Forbidden:
 			return objects.NewObjectsListForbidden().
 				WithPayload(errPayloadFromSingleErr(err))
 		default:
@@ -291,7 +291,7 @@ func (h *objectHandlers) deleteObject(params objects.ObjectsClassDeleteParams,
 		principal, params.ClassName, params.ID, repl, tenantKey)
 	if err != nil {
 		switch err.(type) {
-		case errors.Forbidden:
+		case autherrs.Forbidden:
 			return objects.NewObjectsClassDeleteForbidden().
 				WithPayload(errPayloadFromSingleErr(err))
 		case uco.ErrNotFound:
@@ -319,16 +319,13 @@ func (h *objectHandlers) updateObject(params objects.ObjectsClassPutParams,
 	object, err := h.manager.UpdateObject(params.HTTPRequest.Context(),
 		principal, params.ClassName, params.ID, params.Body, repl, tenantKey)
 	if err != nil {
-		switch err.(type) {
-		case errors.Forbidden:
-			return objects.NewObjectsClassPutForbidden().
-				WithPayload(errPayloadFromSingleErr(err))
-		case uco.ErrInvalidUserInput:
+		if errors.As(err, &uco.ErrInvalidUserInput{}) {
 			return objects.NewObjectsClassPutUnprocessableEntity().
 				WithPayload(errPayloadFromSingleErr(err))
-		case uco.ErrNotFound:
-			return objects.NewObjectsClassDeleteNotFound()
-		default:
+		} else if errors.As(err, &autherrs.Forbidden{}) {
+			return objects.NewObjectsClassPutForbidden().
+				WithPayload(errPayloadFromSingleErr(err))
+		} else {
 			return objects.NewObjectsClassPutInternalServerError().
 				WithPayload(errPayloadFromSingleErr(err))
 		}
diff --git a/adapters/handlers/rest/operations/batch/batch_objects_create_parameters.go b/adapters/handlers/rest/operations/batch/batch_objects_create_parameters.go
index 92606465e6..a10e3705a4 100644
--- a/adapters/handlers/rest/operations/batch/batch_objects_create_parameters.go
+++ b/adapters/handlers/rest/operations/batch/batch_objects_create_parameters.go
@@ -53,6 +53,10 @@ type BatchObjectsCreateParams struct {
 	  In: query
 	*/
 	ConsistencyLevel *string
+	/*Specifies the tenant in a request targeting a multi-tenant class
+	  In: query
+	*/
+	TenantKey *string
 }
 
 // BindRequest both binds and validates a request, it assumes that complex things implement a Validatable(strfmt.Registry) error interface
@@ -98,6 +102,11 @@ func (o *BatchObjectsCreateParams) BindRequest(r *http.Request, route *middlewar
 	if err := o.bindConsistencyLevel(qConsistencyLevel, qhkConsistencyLevel, route.Formats); err != nil {
 		res = append(res, err)
 	}
+
+	qTenantKey, qhkTenantKey, _ := qs.GetOK("tenant_key")
+	if err := o.bindTenantKey(qTenantKey, qhkTenantKey, route.Formats); err != nil {
+		res = append(res, err)
+	}
 	if len(res) > 0 {
 		return errors.CompositeValidationError(res...)
 	}
@@ -121,3 +130,21 @@ func (o *BatchObjectsCreateParams) bindConsistencyLevel(rawData []string, hasKey
 
 	return nil
 }
+
+// bindTenantKey binds and validates parameter TenantKey from query.
+func (o *BatchObjectsCreateParams) bindTenantKey(rawData []string, hasKey bool, formats strfmt.Registry) error {
+	var raw string
+	if len(rawData) > 0 {
+		raw = rawData[len(rawData)-1]
+	}
+
+	// Required: false
+	// AllowEmptyValue: false
+
+	if raw == "" { // empty values pass all other validations
+		return nil
+	}
+	o.TenantKey = &raw
+
+	return nil
+}
diff --git a/adapters/handlers/rest/operations/batch/batch_objects_create_urlbuilder.go b/adapters/handlers/rest/operations/batch/batch_objects_create_urlbuilder.go
index fba49c20e0..c167590cfb 100644
--- a/adapters/handlers/rest/operations/batch/batch_objects_create_urlbuilder.go
+++ b/adapters/handlers/rest/operations/batch/batch_objects_create_urlbuilder.go
@@ -25,6 +25,7 @@ import (
 // BatchObjectsCreateURL generates an URL for the batch objects create operation
 type BatchObjectsCreateURL struct {
 	ConsistencyLevel *string
+	TenantKey        *string
 
 	_basePath string
 	// avoid unkeyed usage
@@ -68,6 +69,14 @@ func (o *BatchObjectsCreateURL) Build() (*url.URL, error) {
 		qs.Set("consistency_level", consistencyLevelQ)
 	}
 
+	var tenantKeyQ string
+	if o.TenantKey != nil {
+		tenantKeyQ = *o.TenantKey
+	}
+	if tenantKeyQ != "" {
+		qs.Set("tenant_key", tenantKeyQ)
+	}
+
 	_result.RawQuery = qs.Encode()
 
 	return &_result, nil
diff --git a/adapters/handlers/rest/operations/batch/batch_objects_delete_parameters.go b/adapters/handlers/rest/operations/batch/batch_objects_delete_parameters.go
index d7615c1b19..bd87195e08 100644
--- a/adapters/handlers/rest/operations/batch/batch_objects_delete_parameters.go
+++ b/adapters/handlers/rest/operations/batch/batch_objects_delete_parameters.go
@@ -55,6 +55,10 @@ type BatchObjectsDeleteParams struct {
 	  In: query
 	*/
 	ConsistencyLevel *string
+	/*Specifies the tenant in a request targeting a multi-tenant class
+	  In: query
+	*/
+	TenantKey *string
 }
 
 // BindRequest both binds and validates a request, it assumes that complex things implement a Validatable(strfmt.Registry) error interface
@@ -100,6 +104,11 @@ func (o *BatchObjectsDeleteParams) BindRequest(r *http.Request, route *middlewar
 	if err := o.bindConsistencyLevel(qConsistencyLevel, qhkConsistencyLevel, route.Formats); err != nil {
 		res = append(res, err)
 	}
+
+	qTenantKey, qhkTenantKey, _ := qs.GetOK("tenant_key")
+	if err := o.bindTenantKey(qTenantKey, qhkTenantKey, route.Formats); err != nil {
+		res = append(res, err)
+	}
 	if len(res) > 0 {
 		return errors.CompositeValidationError(res...)
 	}
@@ -123,3 +132,21 @@ func (o *BatchObjectsDeleteParams) bindConsistencyLevel(rawData []string, hasKey
 
 	return nil
 }
+
+// bindTenantKey binds and validates parameter TenantKey from query.
+func (o *BatchObjectsDeleteParams) bindTenantKey(rawData []string, hasKey bool, formats strfmt.Registry) error {
+	var raw string
+	if len(rawData) > 0 {
+		raw = rawData[len(rawData)-1]
+	}
+
+	// Required: false
+	// AllowEmptyValue: false
+
+	if raw == "" { // empty values pass all other validations
+		return nil
+	}
+	o.TenantKey = &raw
+
+	return nil
+}
diff --git a/adapters/handlers/rest/operations/batch/batch_objects_delete_urlbuilder.go b/adapters/handlers/rest/operations/batch/batch_objects_delete_urlbuilder.go
index a8161f4995..2b4c7d8812 100644
--- a/adapters/handlers/rest/operations/batch/batch_objects_delete_urlbuilder.go
+++ b/adapters/handlers/rest/operations/batch/batch_objects_delete_urlbuilder.go
@@ -25,6 +25,7 @@ import (
 // BatchObjectsDeleteURL generates an URL for the batch objects delete operation
 type BatchObjectsDeleteURL struct {
 	ConsistencyLevel *string
+	TenantKey        *string
 
 	_basePath string
 	// avoid unkeyed usage
@@ -68,6 +69,14 @@ func (o *BatchObjectsDeleteURL) Build() (*url.URL, error) {
 		qs.Set("consistency_level", consistencyLevelQ)
 	}
 
+	var tenantKeyQ string
+	if o.TenantKey != nil {
+		tenantKeyQ = *o.TenantKey
+	}
+	if tenantKeyQ != "" {
+		qs.Set("tenant_key", tenantKeyQ)
+	}
+
 	_result.RawQuery = qs.Encode()
 
 	return &_result, nil
diff --git a/adapters/repos/db/batch.go b/adapters/repos/db/batch.go
index 5580b07b3b..a2b83cb776 100644
--- a/adapters/repos/db/batch.go
+++ b/adapters/repos/db/batch.go
@@ -28,7 +28,7 @@ type batchQueue struct {
 }
 
 func (db *DB) BatchPutObjects(ctx context.Context, objs objects.BatchObjects,
-	repl *additional.ReplicationProperties,
+	repl *additional.ReplicationProperties, tenantKey string,
 ) (objects.BatchObjects, error) {
 	objectByClass := make(map[string]batchQueue)
 	indexByClass := make(map[string]*Index)
@@ -63,7 +63,7 @@ func (db *DB) BatchPutObjects(ctx context.Context, objs objects.BatchObjects,
 		if !ok {
 			continue
 		}
-		errs := index.putObjectBatch(ctx, queue.objects, repl)
+		errs := index.putObjectBatch(ctx, queue.objects, repl, tenantKey)
 		index.dropIndex.RUnlock()
 		for i, err := range errs {
 			if err != nil {
@@ -121,15 +121,18 @@ func (db *DB) AddBatchReferences(ctx context.Context, references objects.BatchRe
 }
 
 func (db *DB) BatchDeleteObjects(ctx context.Context, params objects.BatchDeleteParams,
-	repl *additional.ReplicationProperties,
+	repl *additional.ReplicationProperties, tenantKey string,
 ) (objects.BatchDeleteResult, error) {
 	// get index for a given class
 	idx := db.GetIndex(params.ClassName)
 	if idx == nil {
 		return objects.BatchDeleteResult{}, errors.Errorf("cannot find index for class %v", params.ClassName)
 	}
+	if err := idx.validateMultiTenancy(tenantKey, nil); err != nil {
+		return objects.BatchDeleteResult{}, err
+	}
 	// find all DocIDs in all shards that match the filter
-	shardDocIDs, err := idx.findDocIDs(ctx, params.Filters)
+	shardDocIDs, err := idx.findDocIDs(ctx, params.Filters, tenantKey)
 	if err != nil {
 		return objects.BatchDeleteResult{}, errors.Wrapf(err, "cannot find objects")
 	}
diff --git a/adapters/repos/db/crud.go b/adapters/repos/db/crud.go
index 5338ba68ba..1b6deb9616 100644
--- a/adapters/repos/db/crud.go
+++ b/adapters/repos/db/crud.go
@@ -40,7 +40,7 @@ func (db *DB) PutObject(ctx context.Context, obj *models.Object,
 	}
 
 	if err := idx.putObject(ctx, object, repl, tenantKey); err != nil {
-		return errors.Wrapf(err, "import into index %s", idx.ID())
+		return fmt.Errorf("import into index %s: %w", idx.ID(), err)
 	}
 
 	return nil
diff --git a/adapters/repos/db/index.go b/adapters/repos/db/index.go
index 4437c669eb..f1f99d13d3 100644
--- a/adapters/repos/db/index.go
+++ b/adapters/repos/db/index.go
@@ -309,22 +309,6 @@ func (i *Index) shardFromUUID(in strfmt.UUID) (string, error) {
 		PhysicalShard(uuidBytes), nil
 }
 
-// shardFromObject returns the name of shard to which o belongs
-func (i *Index) shardFromObject(o *storobj.Object) (string, error) {
-	if i.tenantKey == "" {
-		return i.shardFromUUID(o.ID())
-	}
-	keyVal := objects.ParseTenantKeyFromObject(i.tenantKey, &o.Object)
-	if keyVal == "" {
-		return "", fmt.Errorf("tenant key %q is required", i.tenantKey)
-	}
-	ss := i.getSchema.ShardingState(i.Config.ClassName.String())
-	if name := ss.Shard(keyVal, string(o.ID())); name != "" {
-		return name, nil
-	}
-	return "", fmt.Errorf("no tenant found with this key %q", keyVal)
-}
-
 func (i *Index) determineObjectShard(id strfmt.UUID, tenantKey string) (string, error) {
 	if tenantKey != "" {
 		return i.shardFromTenantKey(tenantKey, id)
@@ -343,7 +327,7 @@ func (i *Index) shardFromTenantKey(tenantKey string, id strfmt.UUID) (string, er
 func (i *Index) putObject(ctx context.Context, object *storobj.Object,
 	replProps *additional.ReplicationProperties, tenantKey string,
 ) error {
-	if err := i.validateMultiTenancy(tenantKey); err != nil {
+	if err := i.validateMultiTenancy(tenantKey, &object.Object); err != nil {
 		return err
 	}
 
@@ -502,7 +486,7 @@ func parseAsStringToTime(in interface{}) (time.Time, error) {
 // return value []error gives the error for the index with the positions
 // matching the inputs
 func (i *Index) putObjectBatch(ctx context.Context, objects []*storobj.Object,
-	replProps *additional.ReplicationProperties,
+	replProps *additional.ReplicationProperties, tenantKey string,
 ) []error {
 	i.backupStateLock.RLock()
 	defer i.backupStateLock.RUnlock()
@@ -511,11 +495,25 @@ func (i *Index) putObjectBatch(ctx context.Context, objects []*storobj.Object,
 		pos     []int
 	}
 
+	if err := i.validateMultiTenancy(tenantKey, nil); err != nil {
+		return []error{err}
+	}
+
 	byShard := map[string]objsAndPos{}
 	out := make([]error, len(objects))
 
+	className := string(i.Config.ClassName)
+	ss := i.getSchema.ShardingState(className)
+	if ss == nil {
+		err := fmt.Errorf("could not find sharding state for class %q", className)
+		for i := 0; i < len(out); i++ {
+			out[i] = err
+		}
+		return out
+	}
+
 	for pos, obj := range objects {
-		shardName, err := i.shardFromObject(obj)
+		shardName, err := i.determineObjectShard(obj.ID(), tenantKey)
 		if err != nil {
 			out[pos] = err
 			continue
@@ -606,7 +604,7 @@ func (i *Index) addReferencesBatch(ctx context.Context, refs objects.BatchRefere
 		pos  []int
 	}
 
-	if err := i.validateMultiTenancy(tenantKey); err != nil {
+	if err := i.validateMultiTenancy(tenantKey, nil); err != nil {
 		return []error{err}
 	}
 
@@ -667,7 +665,7 @@ func (i *Index) objectByID(ctx context.Context, id strfmt.UUID,
 	props search.SelectProperties, addl additional.Properties,
 	replProps *additional.ReplicationProperties, tenantKey string,
 ) (*storobj.Object, error) {
-	if err := i.validateMultiTenancy(tenantKey); err != nil {
+	if err := i.validateMultiTenancy(tenantKey, nil); err != nil {
 		return nil, err
 	}
 
@@ -820,7 +818,7 @@ func wrapIDsInMulti(in []strfmt.UUID) []multi.Identifier {
 func (i *Index) exists(ctx context.Context, id strfmt.UUID,
 	replProps *additional.ReplicationProperties, tenantKey string,
 ) (bool, error) {
-	if err := i.validateMultiTenancy(tenantKey); err != nil {
+	if err := i.validateMultiTenancy(tenantKey, nil); err != nil {
 		return false, err
 	}
 
@@ -1234,7 +1232,7 @@ func (i *Index) deleteObject(ctx context.Context, id strfmt.UUID,
 	i.backupStateLock.RLock()
 	defer i.backupStateLock.RUnlock()
 
-	if err := i.validateMultiTenancy(tenantKey); err != nil {
+	if err := i.validateMultiTenancy(tenantKey, nil); err != nil {
 		return err
 	}
 
@@ -1295,7 +1293,7 @@ func (i *Index) mergeObject(ctx context.Context, merge objects.MergeDocument,
 	i.backupStateLock.RLock()
 	defer i.backupStateLock.RUnlock()
 
-	if err := i.validateMultiTenancy(tenantKey); err != nil {
+	if err := i.validateMultiTenancy(tenantKey, nil); err != nil {
 		return err
 	}
 
@@ -1495,13 +1493,22 @@ func (i *Index) notifyReady() {
 }
 
 func (i *Index) findDocIDs(ctx context.Context,
-	filters *filters.LocalFilter,
+	filters *filters.LocalFilter, tenantKey string,
 ) (map[string][]uint64, error) {
 	before := time.Now()
 	defer i.metrics.BatchDelete(before, "filter_total")
 
 	shardState := i.getSchema.ShardingState(i.Config.ClassName.String())
-	shardNames := shardState.AllPhysicalShards()
+	var shardNames []string
+
+	// If this index is multi-tenant-enabled, we are only
+	// interested in deleting objects from the target
+	// tenant's shard
+	if tenantKey != "" {
+		shardNames = []string{tenantKey}
+	} else {
+		shardNames = shardState.AllPhysicalShards()
+	}
 
 	results := make(map[string][]uint64)
 	for _, shardName := range shardNames {
@@ -1516,7 +1523,7 @@ func (i *Index) findDocIDs(ctx context.Context,
 			res, err = shard.findDocIDs(ctx, filters)
 		}
 		if err != nil {
-			return nil, errors.Wrapf(err, "shard %s", shardName)
+			return nil, fmt.Errorf("find matching doc ids in shard %q: %w", shardName, err)
 		}
 
 		results[shardName] = res
@@ -1643,10 +1650,18 @@ func (i *Index) addNewShard(ctx context.Context,
 	return nil
 }
 
-func (i *Index) validateMultiTenancy(tenantKey string) error {
+func (i *Index) validateMultiTenancy(tenantKey string, object *models.Object) error {
 	if i.tenantKey != "" && tenantKey == "" {
-		return fmt.Errorf("class %q has multi-tenancy enabled, tenant_key %q required",
+		return objects.NewErrInvalidUserInput("class %q has multi-tenancy enabled, tenant_key %q required",
 			i.Config.ClassName, i.tenantKey)
 	}
+	if object != nil {
+		parsedTenantKey := objects.ParseTenantKeyFromObject(i.tenantKey, object)
+		if parsedTenantKey != tenantKey {
+			return objects.NewErrInvalidUserInput(
+				"tenant_key query param value %q conflicts with object body value %q for class %q tenant key %q",
+				tenantKey, parsedTenantKey, i.Config.ClassName, i.tenantKey)
+		}
+	}
 	return nil
 }
diff --git a/client/batch/batch_objects_create_parameters.go b/client/batch/batch_objects_create_parameters.go
index 83c646a394..e7b07bc17c 100644
--- a/client/batch/batch_objects_create_parameters.go
+++ b/client/batch/batch_objects_create_parameters.go
@@ -81,6 +81,12 @@ type BatchObjectsCreateParams struct {
 	*/
 	ConsistencyLevel *string
 
+	/* TenantKey.
+
+	   Specifies the tenant in a request targeting a multi-tenant class
+	*/
+	TenantKey *string
+
 	timeout    time.Duration
 	Context    context.Context
 	HTTPClient *http.Client
@@ -156,6 +162,17 @@ func (o *BatchObjectsCreateParams) SetConsistencyLevel(consistencyLevel *string)
 	o.ConsistencyLevel = consistencyLevel
 }
 
+// WithTenantKey adds the tenantKey to the batch objects create params
+func (o *BatchObjectsCreateParams) WithTenantKey(tenantKey *string) *BatchObjectsCreateParams {
+	o.SetTenantKey(tenantKey)
+	return o
+}
+
+// SetTenantKey adds the tenantKey to the batch objects create params
+func (o *BatchObjectsCreateParams) SetTenantKey(tenantKey *string) {
+	o.TenantKey = tenantKey
+}
+
 // WriteToRequest writes these params to a swagger request
 func (o *BatchObjectsCreateParams) WriteToRequest(r runtime.ClientRequest, reg strfmt.Registry) error {
 
@@ -184,6 +201,23 @@ func (o *BatchObjectsCreateParams) WriteToRequest(r runtime.ClientRequest, reg s
 		}
 	}
 
+	if o.TenantKey != nil {
+
+		// query param tenant_key
+		var qrTenantKey string
+
+		if o.TenantKey != nil {
+			qrTenantKey = *o.TenantKey
+		}
+		qTenantKey := qrTenantKey
+		if qTenantKey != "" {
+
+			if err := r.SetQueryParam("tenant_key", qTenantKey); err != nil {
+				return err
+			}
+		}
+	}
+
 	if len(res) > 0 {
 		return errors.CompositeValidationError(res...)
 	}
diff --git a/client/batch/batch_objects_delete_parameters.go b/client/batch/batch_objects_delete_parameters.go
index 85da3afa71..8be2ac9796 100644
--- a/client/batch/batch_objects_delete_parameters.go
+++ b/client/batch/batch_objects_delete_parameters.go
@@ -83,6 +83,12 @@ type BatchObjectsDeleteParams struct {
 	*/
 	ConsistencyLevel *string
 
+	/* TenantKey.
+
+	   Specifies the tenant in a request targeting a multi-tenant class
+	*/
+	TenantKey *string
+
 	timeout    time.Duration
 	Context    context.Context
 	HTTPClient *http.Client
@@ -158,6 +164,17 @@ func (o *BatchObjectsDeleteParams) SetConsistencyLevel(consistencyLevel *string)
 	o.ConsistencyLevel = consistencyLevel
 }
 
+// WithTenantKey adds the tenantKey to the batch objects delete params
+func (o *BatchObjectsDeleteParams) WithTenantKey(tenantKey *string) *BatchObjectsDeleteParams {
+	o.SetTenantKey(tenantKey)
+	return o
+}
+
+// SetTenantKey adds the tenantKey to the batch objects delete params
+func (o *BatchObjectsDeleteParams) SetTenantKey(tenantKey *string) {
+	o.TenantKey = tenantKey
+}
+
 // WriteToRequest writes these params to a swagger request
 func (o *BatchObjectsDeleteParams) WriteToRequest(r runtime.ClientRequest, reg strfmt.Registry) error {
 
@@ -188,6 +205,23 @@ func (o *BatchObjectsDeleteParams) WriteToRequest(r runtime.ClientRequest, reg s
 		}
 	}
 
+	if o.TenantKey != nil {
+
+		// query param tenant_key
+		var qrTenantKey string
+
+		if o.TenantKey != nil {
+			qrTenantKey = *o.TenantKey
+		}
+		qTenantKey := qrTenantKey
+		if qTenantKey != "" {
+
+			if err := r.SetQueryParam("tenant_key", qTenantKey); err != nil {
+				return err
+			}
+		}
+	}
+
 	if len(res) > 0 {
 		return errors.CompositeValidationError(res...)
 	}
diff --git a/openapi-specs/schema.json b/openapi-specs/schema.json
index 7464826b26..52fc8cce5f 100644
--- a/openapi-specs/schema.json
+++ b/openapi-specs/schema.json
@@ -3176,6 +3176,9 @@
           },
           {
             "$ref": "#/parameters/CommonConsistencyLevelParameterQuery"
+          },
+          {
+            "$ref": "#/parameters/CommonTenantKeyParameterQuery"
           }
         ],
         "responses": {
@@ -3241,6 +3244,9 @@
           },
           {
             "$ref": "#/parameters/CommonConsistencyLevelParameterQuery"
+          },
+          {
+            "$ref": "#/parameters/CommonTenantKeyParameterQuery"
           }
         ],
         "responses": {
diff --git a/usecases/classification/classifier.go b/usecases/classification/classifier.go
index 7db0ad3adf..56cd15f07b 100644
--- a/usecases/classification/classifier.go
+++ b/usecases/classification/classifier.go
@@ -112,7 +112,7 @@ type VectorRepo interface {
 type vectorRepo interface {
 	VectorRepo
 	BatchPutObjects(ctx context.Context, objects objects.BatchObjects,
-		repl *additional.ReplicationProperties) (objects.BatchObjects, error)
+		repl *additional.ReplicationProperties, tenantKey string) (objects.BatchObjects, error)
 }
 
 // NeighborRef is the result of an aggregation of the ref properties of k
diff --git a/usecases/classification/writer.go b/usecases/classification/writer.go
index 3cfcdff76c..12a78a250f 100644
--- a/usecases/classification/writer.go
+++ b/usecases/classification/writer.go
@@ -123,7 +123,7 @@ func (r *batchWriter) saveObjects(items objects.BatchObjects) {
 	defer cancel()
 
 	if len(items) > 0 {
-		saved, err := r.vectorRepo.BatchPutObjects(ctx, items, nil)
+		saved, err := r.vectorRepo.BatchPutObjects(ctx, items, nil, "")
 		if err != nil {
 			r.ec.Add(err)
 		}
diff --git a/usecases/objects/add.go b/usecases/objects/add.go
index be528ad996..cfc9bea4c8 100644
--- a/usecases/objects/add.go
+++ b/usecases/objects/add.go
@@ -97,7 +97,7 @@ func (m *Manager) addObjectToConnectorAndSchema(ctx context.Context, principal *
 		return nil, NewErrInvalidUserInput("invalid object: %v", err)
 	}
 
-	err = m.validateObjectAndNormalizeNames(ctx, principal, object, repl, tenantKey)
+	err = m.validateObjectAndNormalizeNames(ctx, principal, repl, object, nil, tenantKey)
 	if err != nil {
 		return nil, NewErrInvalidUserInput("invalid object: %v", err)
 	}
@@ -119,24 +119,41 @@ func (m *Manager) addObjectToConnectorAndSchema(ctx context.Context, principal *
 
 	err = m.vectorRepo.PutObject(ctx, object, object.Vector, repl, tenantKey)
 	if err != nil {
-		return nil, fmt.Errorf("put object: %s", err)
+		return nil, fmt.Errorf("put object: %w", err)
 	}
 
 	return object, nil
 }
 
-func (m *Manager) validateObjectAndNormalizeNames(ctx context.Context, principal *models.Principal,
-	object *models.Object, repl *additional.ReplicationProperties, tenantKey string,
+func (m *Manager) validateObjectAndNormalizeNames(ctx context.Context,
+	principal *models.Principal, repl *additional.ReplicationProperties,
+	incoming *models.Object, existing *models.Object, tenantKey string,
 ) error {
-	// Validate schema given in body with the weaviate schema
-	if _, err := uuid.Parse(object.ID.String()); err != nil {
+	class, err := m.validateSchema(ctx, principal, incoming)
+	if err != nil {
 		return err
 	}
 
-	class, err := m.schemaManager.GetClass(ctx, principal, object.Class)
+	return validation.New(m.vectorRepo.Exists, m.config, repl, tenantKey).
+		Object(ctx, class, incoming, existing)
+}
+
+func (m *Manager) validateSchema(ctx context.Context,
+	principal *models.Principal, obj *models.Object,
+) (*models.Class, error) {
+	// Validate schema given in body with the weaviate schema
+	if _, err := uuid.Parse(obj.ID.String()); err != nil {
+		return nil, err
+	}
+
+	class, err := m.schemaManager.GetClass(ctx, principal, obj.Class)
 	if err != nil {
-		return err
+		return nil, err
+	}
+
+	if class == nil {
+		return nil, fmt.Errorf("class %q not found in schema", obj.Class)
 	}
 
-	return validation.New(m.vectorRepo.Exists, m.config, repl, tenantKey).Object(ctx, object, class)
+	return class, nil
 }
diff --git a/usecases/objects/batch_add.go b/usecases/objects/batch_add.go
index 94cc02cd9f..5432655a15 100644
--- a/usecases/objects/batch_add.go
+++ b/usecases/objects/batch_add.go
@@ -28,6 +28,7 @@ import (
 // AddObjects Class Instances in batch to the connected DB
 func (b *BatchManager) AddObjects(ctx context.Context, principal *models.Principal,
 	objects []*models.Object, fields []*string, repl *additional.ReplicationProperties,
+	tenantKey string,
 ) (BatchObjects, error) {
 	err := b.authorizer.Authorize(principal, "create", "batch/objects")
 	if err != nil {
@@ -45,18 +46,19 @@ func (b *BatchManager) AddObjects(ctx context.Context, principal *models.Princip
 	defer b.metrics.BatchOp("total_uc_level", before.UnixNano())
 	defer b.metrics.BatchDec()
 
-	return b.addObjects(ctx, principal, objects, fields, repl)
+	return b.addObjects(ctx, principal, objects, fields, repl, tenantKey)
 }
 
 func (b *BatchManager) addObjects(ctx context.Context, principal *models.Principal,
 	classes []*models.Object, fields []*string, repl *additional.ReplicationProperties,
+	tenantKey string,
 ) (BatchObjects, error) {
 	beforePreProcessing := time.Now()
 	if err := b.validateObjectForm(classes); err != nil {
 		return nil, NewErrInvalidUserInput("invalid param 'objects': %v", err)
 	}
 
-	batchObjects := b.validateObjectsConcurrently(ctx, principal, classes, fields, repl)
+	batchObjects := b.validateObjectsConcurrently(ctx, principal, classes, fields, repl, tenantKey)
 	b.metrics.BatchOp("total_preprocessing", beforePreProcessing.UnixNano())
 
 	var (
@@ -66,7 +68,7 @@ func (b *BatchManager) addObjects(ctx context.Context, principal *models.Princip
 
 	beforePersistence := time.Now()
 	defer b.metrics.BatchOp("total_persistence_level", beforePersistence.UnixNano())
-	if res, err = b.vectorRepo.BatchPutObjects(ctx, batchObjects, repl); err != nil {
+	if res, err = b.vectorRepo.BatchPutObjects(ctx, batchObjects, repl, tenantKey); err != nil {
 		return nil, NewErrInternal("batch objects: %#v", err)
 	}
 
@@ -83,6 +85,7 @@ func (b *BatchManager) validateObjectForm(classes []*models.Object) error {
 
 func (b *BatchManager) validateObjectsConcurrently(ctx context.Context, principal *models.Principal,
 	classes []*models.Object, fields []*string, repl *additional.ReplicationProperties,
+	tenantKey string,
 ) BatchObjects {
 	fieldsToKeep := determineResponseFields(fields)
 	c := make(chan BatchObject, len(classes))
@@ -92,8 +95,7 @@ func (b *BatchManager) validateObjectsConcurrently(ctx context.Context, principa
 	// Generate a goroutine for each separate request
 	for i, object := range classes {
 		wg.Add(1)
-		go b.validateObject(ctx, principal, wg,
-			object, i, &c, fieldsToKeep, repl)
+		go b.validateObject(ctx, principal, wg, object, i, &c, fieldsToKeep, repl, tenantKey)
 	}
 
 	wg.Wait()
@@ -103,7 +105,7 @@ func (b *BatchManager) validateObjectsConcurrently(ctx context.Context, principa
 
 func (b *BatchManager) validateObject(ctx context.Context, principal *models.Principal,
 	wg *sync.WaitGroup, concept *models.Object, originalIndex int, resultsC *chan BatchObject,
-	fieldsToKeep map[string]struct{}, repl *additional.ReplicationProperties,
+	fieldsToKeep map[string]struct{}, repl *additional.ReplicationProperties, tenantKey string,
 ) {
 	defer wg.Done()
 
@@ -154,13 +156,11 @@ func (b *BatchManager) validateObject(ctx context.Context, principal *models.Pri
 	if class == nil {
 		ec.Add(fmt.Errorf("class '%s' not present in schema", object.Class))
 	} else {
-		var tenantKey string
-		if class.MultiTenancyConfig != nil {
-			tenantKey = ParseTenantKeyFromObject(
-				class.MultiTenancyConfig.TenantKey, concept)
-		}
+		err = validateSingleBatchObjectTenantKey(class, concept, tenantKey, ec)
+		ec.Add(err)
 
-		err = validation.New(b.vectorRepo.Exists, b.config, repl, tenantKey).Object(ctx, object, class)
+		err = validation.New(b.vectorRepo.Exists, b.config, repl, tenantKey).
+			Object(ctx, class, object, nil)
 		ec.Add(err)
 
 		err = b.modulesProvider.UpdateVector(ctx, object, class, nil, b.findObject, b.logger)
diff --git a/usecases/objects/batch_delete.go b/usecases/objects/batch_delete.go
index f3228fd9ef..24b5a20140 100644
--- a/usecases/objects/batch_delete.go
+++ b/usecases/objects/batch_delete.go
@@ -31,7 +31,7 @@ const (
 // DeleteObjects deletes objects in batch based on the match filter
 func (b *BatchManager) DeleteObjects(ctx context.Context, principal *models.Principal,
 	match *models.BatchDeleteMatch, dryRun *bool, output *string,
-	repl *additional.ReplicationProperties,
+	repl *additional.ReplicationProperties, tenantKey string,
 ) (*BatchDeleteResponse, error) {
 	err := b.authorizer.Authorize(principal, "delete", "batch/objects")
 	if err != nil {
@@ -47,21 +47,21 @@ func (b *BatchManager) DeleteObjects(ctx context.Context, principal *models.Prin
 	b.metrics.BatchDeleteInc()
 	defer b.metrics.BatchDeleteDec()
 
-	return b.deleteObjects(ctx, principal, match, dryRun, output, repl)
+	return b.deleteObjects(ctx, principal, match, dryRun, output, repl, tenantKey)
 }
 
 func (b *BatchManager) deleteObjects(ctx context.Context, principal *models.Principal,
 	match *models.BatchDeleteMatch, dryRun *bool, output *string,
-	repl *additional.ReplicationProperties,
+	repl *additional.ReplicationProperties, tenantKey string,
 ) (*BatchDeleteResponse, error) {
 	params, err := b.validateBatchDelete(ctx, principal, match, dryRun, output)
 	if err != nil {
 		return nil, NewErrInvalidUserInput("validate: %v", err)
 	}
 
-	result, err := b.vectorRepo.BatchDeleteObjects(ctx, *params, repl)
+	result, err := b.vectorRepo.BatchDeleteObjects(ctx, *params, repl, tenantKey)
 	if err != nil {
-		return nil, NewErrInternal("batch delete objects: %#v", err)
+		return nil, fmt.Errorf("batch delete objects: %w", err)
 	}
 
 	return b.toResponse(match, params.Output, result)
diff --git a/usecases/objects/batch_manager.go b/usecases/objects/batch_manager.go
index 2be03a010e..01c31f6da8 100644
--- a/usecases/objects/batch_manager.go
+++ b/usecases/objects/batch_manager.go
@@ -41,9 +41,9 @@ type BatchVectorRepo interface {
 
 type batchRepoNew interface {
 	BatchPutObjects(ctx context.Context, objects BatchObjects,
-		repl *additional.ReplicationProperties) (BatchObjects, error)
+		repl *additional.ReplicationProperties, tenantKey string) (BatchObjects, error)
 	BatchDeleteObjects(ctx context.Context, params BatchDeleteParams,
-		repl *additional.ReplicationProperties) (BatchDeleteResult, error)
+		repl *additional.ReplicationProperties, tenantKey string) (BatchDeleteResult, error)
 	AddBatchReferences(ctx context.Context, references BatchReferences,
 		repl *additional.ReplicationProperties, tenantKey string) (BatchReferences, error)
 }
diff --git a/usecases/objects/merge.go b/usecases/objects/merge.go
index e1c6acceef..d9319112c9 100644
--- a/usecases/objects/merge.go
+++ b/usecases/objects/merge.go
@@ -51,6 +51,14 @@ func (m *Manager) MergeObject(ctx context.Context, principal *models.Principal,
 	m.metrics.MergeObjectInc()
 	defer m.metrics.MergeObjectDec()
 
+	obj, err := m.vectorRepo.Object(ctx, cls, id, nil, additional.Properties{}, repl, tenantKey)
+	if err != nil {
+		return &Error{"repo.object", StatusInternalServerError, err}
+	}
+	if obj == nil {
+		return &Error{"not found", StatusNotFound, err}
+	}
+
 	var propertiesToDelete []string
 	if updates.Properties != nil {
 		for key, val := range updates.Properties.(map[string]interface{}) {
@@ -60,20 +68,15 @@ func (m *Manager) MergeObject(ctx context.Context, principal *models.Principal,
 		}
 	}
 
-	if err := m.validateObjectAndNormalizeNames(ctx, principal, updates, repl, tenantKey); err != nil {
+	if err := m.validateObjectAndNormalizeNames(
+		ctx, principal, repl, updates, obj.Object(), tenantKey); err != nil {
 		return &Error{"bad request", StatusBadRequest, err}
 	}
 
 	if updates.Properties == nil {
 		updates.Properties = map[string]interface{}{}
 	}
-	obj, err := m.vectorRepo.Object(ctx, cls, id, nil, additional.Properties{}, repl, tenantKey)
-	if err != nil {
-		return &Error{"repo.object", StatusInternalServerError, err}
-	}
-	if obj == nil {
-		return &Error{"not found", StatusNotFound, err}
-	}
+
 	return m.patchObject(ctx, principal, obj, updates, repl, propertiesToDelete, tenantKey)
 }
 
diff --git a/usecases/objects/multi_tenancy.go b/usecases/objects/multi_tenancy.go
index 34ba665ca2..4aa822c3c2 100644
--- a/usecases/objects/multi_tenancy.go
+++ b/usecases/objects/multi_tenancy.go
@@ -11,7 +11,12 @@
 
 package objects
 
-import "github.com/weaviate/weaviate/entities/models"
+import (
+	"fmt"
+
+	"github.com/weaviate/weaviate/entities/errorcompounder"
+	"github.com/weaviate/weaviate/entities/models"
+)
 
 // ParseTenantKeyFromObject extract the value of the tenant key if it exists
 func ParseTenantKeyFromObject(tenantKeyName string, o *models.Object) string {
@@ -24,3 +29,20 @@ func ParseTenantKeyFromObject(tenantKeyName string, o *models.Object) string {
 	}
 	return ""
 }
+
+func validateSingleBatchObjectTenantKey(class *models.Class, obj *models.Object,
+	tk string, ec *errorcompounder.ErrorCompounder,
+) error {
+	var objTk string
+	if class.MultiTenancyConfig != nil {
+		if tk == "" {
+			return NewErrInvalidUserInput("class %q has multi-tenancy enabled, tenant_key %q required",
+				class.Class, class.MultiTenancyConfig.TenantKey)
+		}
+		objTk = ParseTenantKeyFromObject(class.MultiTenancyConfig.TenantKey, obj)
+		if objTk != tk {
+			ec.Add(fmt.Errorf("object does not belong to tenant %q", tk))
+		}
+	}
+	return nil
+}
diff --git a/usecases/objects/update.go b/usecases/objects/update.go
index 5bf05f3cd3..1c9dfc7179 100644
--- a/usecases/objects/update.go
+++ b/usecases/objects/update.go
@@ -68,7 +68,8 @@ func (m *Manager) updateObjectToConnectorAndSchema(ctx context.Context,
 		WithField("id", id).
 		Debug("received update kind request")
 
-	err = m.validateObjectAndNormalizeNames(ctx, principal, updates, repl, tenantKey)
+	err = m.validateObjectAndNormalizeNames(
+		ctx, principal, repl, updates, obj.Object(), tenantKey)
 	if err != nil {
 		return nil, NewErrInvalidUserInput("invalid object: %v", err)
 	}
@@ -91,7 +92,7 @@ func (m *Manager) updateObjectToConnectorAndSchema(ctx context.Context,
 
 	err = m.vectorRepo.PutObject(ctx, updates, updates.Vector, repl, tenantKey)
 	if err != nil {
-		return nil, NewErrInternal("put object: %v", err)
+		return nil, fmt.Errorf("put object: %w", err)
 	}
 
 	return updates, nil
diff --git a/usecases/objects/validate.go b/usecases/objects/validate.go
index e51ed08e01..facd314095 100644
--- a/usecases/objects/validate.go
+++ b/usecases/objects/validate.go
@@ -34,7 +34,7 @@ func (m *Manager) ValidateObject(ctx context.Context, principal *models.Principa
 	}
 	defer unlock()
 
-	err = m.validateObjectAndNormalizeNames(ctx, principal, obj, repl, "")
+	err = m.validateObjectAndNormalizeNames(ctx, principal, repl, obj, nil, "")
 	if err != nil {
 		return NewErrInvalidUserInput("invalid object: %v", err)
 	}
diff --git a/usecases/objects/validation/model_validation.go b/usecases/objects/validation/model_validation.go
index 29fcf56662..b9cbdad6ba 100644
--- a/usecases/objects/validation/model_validation.go
+++ b/usecases/objects/validation/model_validation.go
@@ -79,12 +79,14 @@ func New(exists exists, config *config.WeaviateConfig,
 	}
 }
 
-func (v *Validator) Object(ctx context.Context, object *models.Object, class *models.Class) error {
-	if err := validateClass(object.Class); err != nil {
+func (v *Validator) Object(ctx context.Context, class *models.Class,
+	incoming *models.Object, existing *models.Object,
+) error {
+	if err := validateClass(incoming.Class); err != nil {
 		return err
 	}
 
-	return v.properties(ctx, object, class)
+	return v.properties(ctx, class, incoming, existing)
 }
 
 func validateClass(class string) error {
diff --git a/usecases/objects/validation/properties_validation.go b/usecases/objects/validation/properties_validation.go
index e4e100eab3..bb8c6dcbd1 100644
--- a/usecases/objects/validation/properties_validation.go
+++ b/usecases/objects/validation/properties_validation.go
@@ -40,10 +40,12 @@ const (
 	ErrorMissingSingleRefType string = "class '%s' with property '%s' requires exactly 3 arguments: 'beacon', 'locationUrl' and 'type'. 'type' is missing, check your input schema"
 )
 
-func (v *Validator) properties(ctx context.Context, object interface{}, class *models.Class) error {
-	className := object.(*models.Object).Class
-	isp := object.(*models.Object).Properties
-	vectorWeights := object.(*models.Object).VectorWeights
+func (v *Validator) properties(ctx context.Context, class *models.Class,
+	incomingObject *models.Object, existingObject *models.Object,
+) error {
+	className := incomingObject.Class
+	isp := incomingObject.Properties
+	vectorWeights := incomingObject.VectorWeights
 
 	if vectorWeights != nil {
 		res, err := v.validateVectorWeights(vectorWeights)
@@ -67,6 +69,15 @@ func (v *Validator) properties(ctx context.Context, object interface{}, class *m
 			continue // nil values are removed and filtered out
 		}
 
+		mt := class.MultiTenancyConfig
+		if mt != nil && existingObject != nil && propertyKey == mt.TenantKey {
+			beforeKey := existingObject.Properties.(map[string]interface{})[mt.TenantKey]
+			afterKey := isp.(map[string]interface{})[mt.TenantKey]
+			if beforeKey != afterKey {
+				return fmt.Errorf("tenant key %q is immutable", mt.TenantKey)
+			}
+		}
+
 		// properties in the class are saved with lower case first letter
 		propertyKeyLowerCase := strings.ToLower(propertyKey[:1])
 		if len(propertyKey) > 1 {
@@ -85,8 +96,8 @@ func (v *Validator) properties(ctx context.Context, object interface{}, class *m
 		returnSchema[propertyKeyLowerCase] = data
 	}
 
-	object.(*models.Object).Properties = returnSchema
-	object.(*models.Object).VectorWeights = vectorWeights
+	incomingObject.Properties = returnSchema
+	incomingObject.VectorWeights = vectorWeights
 
 	return nil
 }
diff --git a/usecases/schema/add.go b/usecases/schema/add.go
index 1bbd2c5fee..bcbd8d1d6a 100644
--- a/usecases/schema/add.go
+++ b/usecases/schema/add.go
@@ -132,6 +132,10 @@ func (m *Manager) addClass(ctx context.Context, class *models.Class,
 	class.Class = schema.UppercaseClassName(class.Class)
 	class.Properties = schema.LowercaseAllPropertyNames(class.Properties)
 
+	if class.ShardingConfig != nil && class.MultiTenancyConfig != nil {
+		return nil, fmt.Errorf("cannot have both shardingConfig and multiTenancyConfig")
+	}
+
 	m.setClassDefaults(class)
 	err := m.validateCanAddClass(ctx, class, false)
 	if err != nil {
@@ -375,10 +379,6 @@ func (m *Manager) validateCanAddClass(
 		return err
 	}
 
-	if class.ShardingConfig != nil && class.MultiTenancyConfig != nil {
-		return fmt.Errorf("cannot have both shardingConfig and multiTenancyConfig")
-	}
-
 	// all is fine!
 	return nil
 }


Test patch:
diff --git a/adapters/repos/db/batch_integration_test.go b/adapters/repos/db/batch_integration_test.go
index ef7ecccd34..670aaca758 100644
--- a/adapters/repos/db/batch_integration_test.go
+++ b/adapters/repos/db/batch_integration_test.go
@@ -188,41 +188,40 @@ func TestBatchDeleteObjectsWithDimensions(t *testing.T) {
 }
 
 func delete2Objects(t *testing.T, repo *DB, className string) {
-	batchDeleteRes, err := repo.BatchDeleteObjects(context.Background(),
-		objects.BatchDeleteParams{
-			ClassName: "ThingForBatching",
-			Filters: &filters.LocalFilter{
-				Root: &filters.Clause{
-					Operator: filters.OperatorOr,
-					Operands: []filters.Clause{
-						{
-							Operator: filters.OperatorEqual,
-							On: &filters.Path{
-								Class:    "ThingForBatching",
-								Property: schema.PropertyName("id"),
-							},
-							Value: &filters.Value{
-								Value: "8d5a3aa2-3c8d-4589-9ae1-3f638f506003",
-								Type:  schema.DataTypeText,
-							},
+	batchDeleteRes, err := repo.BatchDeleteObjects(context.Background(), objects.BatchDeleteParams{
+		ClassName: "ThingForBatching",
+		Filters: &filters.LocalFilter{
+			Root: &filters.Clause{
+				Operator: filters.OperatorOr,
+				Operands: []filters.Clause{
+					{
+						Operator: filters.OperatorEqual,
+						On: &filters.Path{
+							Class:    "ThingForBatching",
+							Property: schema.PropertyName("id"),
 						},
-						{
-							Operator: filters.OperatorEqual,
-							On: &filters.Path{
-								Class:    "ThingForBatching",
-								Property: schema.PropertyName("id"),
-							},
-							Value: &filters.Value{
-								Value: "8d5a3aa2-3c8d-4589-9ae1-3f638f506004",
-								Type:  schema.DataTypeText,
-							},
+						Value: &filters.Value{
+							Value: "8d5a3aa2-3c8d-4589-9ae1-3f638f506003",
+							Type:  schema.DataTypeText,
+						},
+					},
+					{
+						Operator: filters.OperatorEqual,
+						On: &filters.Path{
+							Class:    "ThingForBatching",
+							Property: schema.PropertyName("id"),
+						},
+						Value: &filters.Value{
+							Value: "8d5a3aa2-3c8d-4589-9ae1-3f638f506004",
+							Type:  schema.DataTypeText,
 						},
 					},
 				},
 			},
-			DryRun: false,
-			Output: "verbose",
-		}, nil)
+		},
+		DryRun: false,
+		Output: "verbose",
+	}, nil, "")
 	require.Nil(t, err)
 	require.Equal(t, 2, len(batchDeleteRes.Objects), "Objects deleted")
 }
@@ -386,7 +385,7 @@ func testBatchImportObjectsNoVector(repo *DB) func(t *testing.T) {
 			}
 
 			t.Run("can import", func(t *testing.T) {
-				batchRes, err := repo.BatchPutObjects(context.Background(), batch, nil)
+				batchRes, err := repo.BatchPutObjects(context.Background(), batch, nil, "")
 				require.Nil(t, err)
 
 				assert.Nil(t, batchRes[0].Err)
@@ -422,7 +421,7 @@ func simpleInsertObjects(t *testing.T, repo *DB, class string, count int) {
 		}
 	}
 
-	repo.BatchPutObjects(context.Background(), batch, nil)
+	repo.BatchPutObjects(context.Background(), batch, nil, "")
 }
 
 func testBatchImportObjects(repo *DB) func(t *testing.T) {
@@ -471,7 +470,7 @@ func testBatchImportObjects(repo *DB) func(t *testing.T) {
 			}
 
 			t.Run("can import", func(t *testing.T) {
-				batchRes, err := repo.BatchPutObjects(context.Background(), batch, nil)
+				batchRes, err := repo.BatchPutObjects(context.Background(), batch, nil, "")
 				require.Nil(t, err)
 
 				assert.Nil(t, batchRes[0].Err)
@@ -555,7 +554,7 @@ func testBatchImportObjects(repo *DB) func(t *testing.T) {
 			}
 
 			t.Run("can import", func(t *testing.T) {
-				batchRes, err := repo.BatchPutObjects(context.Background(), batch, nil)
+				batchRes, err := repo.BatchPutObjects(context.Background(), batch, nil, "")
 				require.Nil(t, err, "there shouldn't be an overall error, only inividual ones")
 
 				t.Run("element errors are marked correctly", func(t *testing.T) {
@@ -616,7 +615,7 @@ func testBatchImportObjects(repo *DB) func(t *testing.T) {
 				}
 
 				t.Run("can import", func(t *testing.T) {
-					batchRes, err := repo.BatchPutObjects(context.Background(), batch, nil)
+					batchRes, err := repo.BatchPutObjects(context.Background(), batch, nil, "")
 					require.Nil(t, err)
 
 					assert.Nil(t, batchRes[0].Err)
@@ -701,7 +700,7 @@ func testBatchImportObjects(repo *DB) func(t *testing.T) {
 			}
 
 			t.Run("can import", func(t *testing.T) {
-				batchRes, err := repo.BatchPutObjects(context.Background(), batch, nil)
+				batchRes, err := repo.BatchPutObjects(context.Background(), batch, nil, "")
 				require.Nil(t, err, "there shouldn't be an overall error, only inividual ones")
 
 				t.Run("element errors are marked correctly", func(t *testing.T) {
@@ -758,7 +757,7 @@ func testBatchImportObjects(repo *DB) func(t *testing.T) {
 				ctx, cancel := context.WithTimeout(context.Background(), 1*time.Millisecond)
 				defer cancel()
 
-				batchRes, err := repo.BatchPutObjects(ctx, batch, nil)
+				batchRes, err := repo.BatchPutObjects(ctx, batch, nil, "")
 				require.Nil(t, err, "there shouldn't be an overall error, only inividual ones")
 
 				t.Run("some elements have error'd due to context", func(t *testing.T) {
@@ -814,7 +813,7 @@ func testBatchImportGeoObjects(repo *DB) func(t *testing.T) {
 					}
 				}
 
-				res, err := repo.BatchPutObjects(context.Background(), batch, nil)
+				res, err := repo.BatchPutObjects(context.Background(), batch, nil, "")
 				require.Nil(t, err)
 				assertAllItemsErrorFree(t, res)
 			}
@@ -896,7 +895,7 @@ func testBatchImportGeoObjects(repo *DB) func(t *testing.T) {
 					}
 				}
 
-				res, err := repo.BatchPutObjects(context.Background(), batch, nil)
+				res, err := repo.BatchPutObjects(context.Background(), batch, nil, "")
 				require.Nil(t, err)
 				assertAllItemsErrorFree(t, res)
 			}
@@ -977,7 +976,7 @@ func testBatchDeleteObjects(repo *DB) func(t *testing.T) {
 			beforeDelete := len(res)
 			require.True(t, beforeDelete > 0)
 			// dryRun == true, only test how many objects can be deleted
-			batchDeleteRes, err := repo.BatchDeleteObjects(context.Background(), getParams(true, "verbose"), nil)
+			batchDeleteRes, err := repo.BatchDeleteObjects(context.Background(), getParams(true, "verbose"), nil, "")
 			require.Nil(t, err)
 			require.Equal(t, int64(beforeDelete), batchDeleteRes.Matches)
 			require.Equal(t, beforeDelete, len(batchDeleteRes.Objects))
@@ -996,7 +995,7 @@ func testBatchDeleteObjects(repo *DB) func(t *testing.T) {
 			beforeDelete := len(res)
 			require.True(t, beforeDelete > 0)
 			// dryRun == true, only test how many objects can be deleted
-			batchDeleteRes, err := repo.BatchDeleteObjects(context.Background(), getParams(true, "minimal"), nil)
+			batchDeleteRes, err := repo.BatchDeleteObjects(context.Background(), getParams(true, "minimal"), nil, "")
 			require.Nil(t, err)
 			require.Equal(t, int64(beforeDelete), batchDeleteRes.Matches)
 			require.Equal(t, beforeDelete, len(batchDeleteRes.Objects))
@@ -1048,7 +1047,7 @@ func testBatchDeleteObjects(repo *DB) func(t *testing.T) {
 				},
 				DryRun: false,
 				Output: "verbose",
-			}, nil)
+			}, nil, "")
 			require.Nil(t, err)
 			require.Equal(t, int64(2), batchDeleteRes.Matches)
 			require.Equal(t, 2, len(batchDeleteRes.Objects))
@@ -1067,7 +1066,7 @@ func testBatchDeleteObjects(repo *DB) func(t *testing.T) {
 			beforeDelete := len(res)
 			require.True(t, beforeDelete > 0)
 			// dryRun == true, only test how many objects can be deleted
-			batchDeleteRes, err := repo.BatchDeleteObjects(context.Background(), getParams(false, "verbose"), nil)
+			batchDeleteRes, err := repo.BatchDeleteObjects(context.Background(), getParams(false, "verbose"), nil, "")
 			require.Nil(t, err)
 			require.Equal(t, int64(beforeDelete), batchDeleteRes.Matches)
 			require.Equal(t, beforeDelete, len(batchDeleteRes.Objects))
@@ -1110,7 +1109,7 @@ func testBatchDeleteObjectsJourney(repo *DB, queryMaximumResults int64) func(t *
 		}
 		t.Run("batch delete journey", func(t *testing.T) {
 			// delete objects to limit
-			batchDeleteRes, err := repo.BatchDeleteObjects(context.Background(), getParams(true, "verbose"), nil)
+			batchDeleteRes, err := repo.BatchDeleteObjects(context.Background(), getParams(true, "verbose"), nil, "")
 			require.Nil(t, err)
 			objectsMatches := batchDeleteRes.Matches
 
@@ -1119,14 +1118,14 @@ func testBatchDeleteObjectsJourney(repo *DB, queryMaximumResults int64) func(t *
 			deletedObjectsCount := 0
 			for {
 				// delete objects to limit
-				batchDeleteRes, err := repo.BatchDeleteObjects(context.Background(), getParams(false, "verbose"), nil)
+				batchDeleteRes, err := repo.BatchDeleteObjects(context.Background(), getParams(false, "verbose"), nil, "")
 				require.Nil(t, err)
 				matches, deleted := batchDeleteRes.Matches, len(batchDeleteRes.Objects)
 				require.Equal(t, leftToDelete, matches)
 				require.True(t, deleted > 0)
 				deletedObjectsCount += deleted
 
-				batchDeleteRes, err = repo.BatchDeleteObjects(context.Background(), getParams(true, "verbose"), nil)
+				batchDeleteRes, err = repo.BatchDeleteObjects(context.Background(), getParams(true, "verbose"), nil, "")
 				require.Nil(t, err)
 				leftToDelete = batchDeleteRes.Matches
 
diff --git a/adapters/repos/db/clusterintegrationtest/cluster_integration_test.go b/adapters/repos/db/clusterintegrationtest/cluster_integration_test.go
index b60b896922..1571284830 100644
--- a/adapters/repos/db/clusterintegrationtest/cluster_integration_test.go
+++ b/adapters/repos/db/clusterintegrationtest/cluster_integration_test.go
@@ -101,7 +101,7 @@ func testDistributed(t *testing.T, dirName string, batch bool) {
 			node := nodes[rand.Intn(len(nodes))]
 
 			batchObjs := dataAsBatch(data)
-			res, err := node.repo.BatchPutObjects(context.Background(), batchObjs, nil)
+			res, err := node.repo.BatchPutObjects(context.Background(), batchObjs, nil, "")
 			require.Nil(t, err)
 			for _, ind := range res {
 				require.Nil(t, ind.Err)
@@ -113,7 +113,7 @@ func testDistributed(t *testing.T, dirName string, batch bool) {
 			node := nodes[rand.Intn(len(nodes))]
 
 			batchObjs := dataAsBatchWithProps(refData, []string{"description"})
-			res, err := node.repo.BatchPutObjects(context.Background(), batchObjs, nil)
+			res, err := node.repo.BatchPutObjects(context.Background(), batchObjs, nil, "")
 			require.Nil(t, err)
 			for _, ind := range res {
 				require.Nil(t, ind.Err)
@@ -652,7 +652,7 @@ func testDistributed(t *testing.T, dirName string, batch bool) {
 		beforeDelete := len(res)
 		require.True(t, beforeDelete > 0)
 		// dryRun == false, perform actual delete
-		batchDeleteRes, err := node.repo.BatchDeleteObjects(context.Background(), getParams(distributedClass, false), nil)
+		batchDeleteRes, err := node.repo.BatchDeleteObjects(context.Background(), getParams(distributedClass, false), nil, "")
 		require.Nil(t, err)
 		require.Equal(t, int64(beforeDelete), batchDeleteRes.Matches)
 		require.Equal(t, beforeDelete, len(batchDeleteRes.Objects))
diff --git a/adapters/repos/db/crud_integration_test.go b/adapters/repos/db/crud_integration_test.go
index 87a78461c4..ec261b6084 100644
--- a/adapters/repos/db/crud_integration_test.go
+++ b/adapters/repos/db/crud_integration_test.go
@@ -1101,7 +1101,7 @@ func TestCRUD(t *testing.T) {
 				}
 				createdActionIDs[i] = newID
 			}
-			batchObjResp, err := repo.BatchPutObjects(context.Background(), actionBatch, nil)
+			batchObjResp, err := repo.BatchPutObjects(context.Background(), actionBatch, nil, "")
 			require.Len(t, batchObjResp, numThings)
 			require.Nil(t, err)
 			for _, r := range batchObjResp {
@@ -1125,7 +1125,7 @@ func TestCRUD(t *testing.T) {
 				}
 				createdThingIDs[i] = newID
 			}
-			batchObjResp, err := repo.BatchPutObjects(context.Background(), thingBatch, nil)
+			batchObjResp, err := repo.BatchPutObjects(context.Background(), thingBatch, nil, "")
 			require.Len(t, batchObjResp, numThings)
 			require.Nil(t, err)
 			for _, r := range batchObjResp {
@@ -1599,7 +1599,7 @@ func Test_ImportWithoutVector_UpdateWithVectorLater(t *testing.T) {
 			}
 		}
 
-		res, err := repo.BatchPutObjects(context.Background(), batch, nil)
+		res, err := repo.BatchPutObjects(context.Background(), batch, nil, "")
 		require.Nil(t, err)
 
 		for _, obj := range res {
diff --git a/adapters/repos/db/crud_null_objects_integration_test.go b/adapters/repos/db/crud_null_objects_integration_test.go
index b80d8b6b09..2611d1b948 100644
--- a/adapters/repos/db/crud_null_objects_integration_test.go
+++ b/adapters/repos/db/crud_null_objects_integration_test.go
@@ -124,7 +124,7 @@ func TestNullArrayClass(t *testing.T) {
 				batch := make([]objects.BatchObject, 2)
 				batch[0] = objects.BatchObject{Object: arrayObjNil, UUID: arrayObjNil.ID}
 				batch[1] = objects.BatchObject{Object: arrayObjEmpty, UUID: arrayObjEmpty.ID}
-				_, err := repo.BatchPutObjects(context.Background(), batch, nil)
+				_, err := repo.BatchPutObjects(context.Background(), batch, nil, "")
 				assert.Nil(t, err)
 			}
 
diff --git a/adapters/repos/db/multi_shard_integration_test.go b/adapters/repos/db/multi_shard_integration_test.go
index e2e5ff5fd7..c31d93e4b4 100644
--- a/adapters/repos/db/multi_shard_integration_test.go
+++ b/adapters/repos/db/multi_shard_integration_test.go
@@ -104,7 +104,7 @@ func Test_MultiShardJourneys_BatchedImports(t *testing.T) {
 			}
 		}
 
-		_, err := repo.BatchPutObjects(context.Background(), batch, nil)
+		_, err := repo.BatchPutObjects(context.Background(), batch, nil, "")
 		require.Nil(t, err)
 	})
 
@@ -231,7 +231,7 @@ func Test_MultiShardJourneys_BM25_Search(t *testing.T) {
 			},
 		}
 
-		_, err := repo.BatchPutObjects(context.Background(), objs, nil)
+		_, err := repo.BatchPutObjects(context.Background(), objs, nil, "")
 		require.Nil(t, err)
 	})
 
@@ -709,7 +709,7 @@ func makeTestBatchDeleteAllObjects(repo *DB) func(t *testing.T) {
 			beforeDelete := len(res)
 			require.True(t, beforeDelete > 0)
 			// dryRun == true
-			batchDeleteRes, err := repo.BatchDeleteObjects(context.Background(), getParams(className, true), nil)
+			batchDeleteRes, err := repo.BatchDeleteObjects(context.Background(), getParams(className, true), nil, "")
 			require.Nil(t, err)
 			require.Equal(t, int64(beforeDelete), batchDeleteRes.Matches)
 			require.Equal(t, beforeDelete, len(batchDeleteRes.Objects))
@@ -721,7 +721,7 @@ func makeTestBatchDeleteAllObjects(repo *DB) func(t *testing.T) {
 			require.Nil(t, err)
 			require.Equal(t, beforeDelete, len(res))
 			// dryRun == false, perform actual delete
-			batchDeleteRes, err = repo.BatchDeleteObjects(context.Background(), getParams(className, false), nil)
+			batchDeleteRes, err = repo.BatchDeleteObjects(context.Background(), getParams(className, false), nil, "")
 			require.Nil(t, err)
 			require.Equal(t, int64(beforeDelete), batchDeleteRes.Matches)
 			require.Equal(t, beforeDelete, len(batchDeleteRes.Objects))
diff --git a/adapters/repos/db/nodes_integration_test.go b/adapters/repos/db/nodes_integration_test.go
index 2f7c4b5b6a..c3cc18120c 100644
--- a/adapters/repos/db/nodes_integration_test.go
+++ b/adapters/repos/db/nodes_integration_test.go
@@ -113,7 +113,7 @@ func TestNodesAPI_Journey(t *testing.T) {
 			UUID: "86a380e9-cb60-4b2a-bc48-51f52acd72d6",
 		},
 	}
-	batchRes, err := repo.BatchPutObjects(context.Background(), batch, nil)
+	batchRes, err := repo.BatchPutObjects(context.Background(), batch, nil, "")
 	require.Nil(t, err)
 
 	assert.Nil(t, batchRes[0].Err)
diff --git a/modules/text2vec-contextionary/classification/fakes_for_test.go b/modules/text2vec-contextionary/classification/fakes_for_test.go
index 014d2e01bb..a2ba16c1ed 100644
--- a/modules/text2vec-contextionary/classification/fakes_for_test.go
+++ b/modules/text2vec-contextionary/classification/fakes_for_test.go
@@ -187,9 +187,7 @@ func (f *fakeVectorRepoKNN) VectorClassSearch(ctx context.Context,
 	return nil, fmt.Errorf("vector class search not implemented in fake")
 }
 
-func (f *fakeVectorRepoKNN) BatchPutObjects(ctx context.Context,
-	objects objects.BatchObjects, _ *additional.ReplicationProperties,
-) (objects.BatchObjects, error) {
+func (f *fakeVectorRepoKNN) BatchPutObjects(ctx context.Context, objects objects.BatchObjects, repl *additional.ReplicationProperties, tenantKey string) (objects.BatchObjects, error) {
 	f.Lock()
 	defer f.Unlock()
 
@@ -262,9 +260,7 @@ func (f *fakeVectorRepoContextual) ZeroShotSearch(ctx context.Context, vector []
 	panic("not implemented")
 }
 
-func (f *fakeVectorRepoContextual) BatchPutObjects(ctx context.Context,
-	objects objects.BatchObjects, _ *additional.ReplicationProperties,
-) (objects.BatchObjects, error) {
+func (f *fakeVectorRepoContextual) BatchPutObjects(ctx context.Context, objects objects.BatchObjects, repl *additional.ReplicationProperties, tenantKey string) (objects.BatchObjects, error) {
 	f.Lock()
 	defer f.Unlock()
 	for _, batchObject := range objects {
diff --git a/test/acceptance/multi_tenancy/add_tenant_objects_test.go b/test/acceptance/multi_tenancy/add_tenant_objects_test.go
index 94f289f698..9e554919e3 100644
--- a/test/acceptance/multi_tenancy/add_tenant_objects_test.go
+++ b/test/acceptance/multi_tenancy/add_tenant_objects_test.go
@@ -12,10 +12,13 @@
 package test
 
 import (
+	"fmt"
+	"net/http"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"github.com/weaviate/weaviate/client/objects"
 	"github.com/weaviate/weaviate/entities/models"
 	"github.com/weaviate/weaviate/test/helper"
 )
@@ -94,3 +97,56 @@ func TestAddTenantObjects(t *testing.T) {
 		}
 	})
 }
+
+func TestAddTenantObjects_MissingTenantKey(t *testing.T) {
+	className := "MultiTenantClass"
+	tenantKey := "tenantName"
+	tenantName := "Tenant1"
+	testClass := models.Class{
+		Class: className,
+		MultiTenancyConfig: &models.MultiTenancyConfig{
+			Enabled:   true,
+			TenantKey: tenantKey,
+		},
+		Properties: []*models.Property{
+			{
+				Name:     tenantKey,
+				DataType: []string{"string"},
+			},
+		},
+	}
+	tenantObject := models.Object{
+		ID:    "0927a1e0-398e-4e76-91fb-04a7a8f0405c",
+		Class: className,
+	}
+
+	defer func() {
+		helper.DeleteClass(t, className)
+	}()
+
+	t.Run("create class with multi-tenancy enabled", func(t *testing.T) {
+		helper.CreateClass(t, &testClass)
+	})
+
+	t.Run("create tenants", func(t *testing.T) {
+		helper.CreateTenants(t, className, []*models.Tenant{{tenantName}})
+	})
+
+	t.Run("add tenant object", func(t *testing.T) {
+		params := objects.NewObjectsCreateParams().
+			WithBody(&tenantObject).WithTenantKey(&tenantName)
+		resp, err := helper.Client(t).Objects.ObjectsCreate(params, nil)
+		require.Nil(t, resp)
+		require.NotNil(t, err)
+		parsedErr, ok := err.(*objects.ObjectsCreateUnprocessableEntity)
+		require.True(t, ok)
+		expected := "put object: import into index multitenantclass: " +
+			"tenant_key query param value \"Tenant1\" conflicts with object " +
+			"body value \"\" for class \"MultiTenantClass\" tenant key \"tenantName\""
+		require.NotNil(t, parsedErr.Payload)
+		require.NotNil(t, parsedErr.Payload.Error)
+		require.Len(t, parsedErr.Payload.Error, 1)
+		assert.Contains(t, err.Error(), fmt.Sprint(http.StatusUnprocessableEntity))
+		assert.Equal(t, expected, parsedErr.Payload.Error[0].Message)
+	})
+}
diff --git a/test/acceptance/multi_tenancy/batch_add_tenant_objects_test.go b/test/acceptance/multi_tenancy/batch_add_tenant_objects_test.go
index 20c4b8c49e..37d07819c7 100644
--- a/test/acceptance/multi_tenancy/batch_add_tenant_objects_test.go
+++ b/test/acceptance/multi_tenancy/batch_add_tenant_objects_test.go
@@ -12,6 +12,7 @@
 package test
 
 import (
+	"fmt"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
@@ -21,6 +22,84 @@ import (
 )
 
 func TestBatchAddTenantObjects(t *testing.T) {
+	tenantKey := "tenantName"
+	testClass := models.Class{
+		Class: "MultiTenantClass",
+		MultiTenancyConfig: &models.MultiTenancyConfig{
+			Enabled:   true,
+			TenantKey: tenantKey,
+		},
+		Properties: []*models.Property{
+			{
+				Name:     tenantKey,
+				DataType: []string{"string"},
+			},
+		},
+	}
+	tenantName := "Tenant1"
+	tenantObjects := []*models.Object{
+		{
+			ID:    "0927a1e0-398e-4e76-91fb-04a7a8f0405c",
+			Class: testClass.Class,
+			Properties: map[string]interface{}{
+				tenantKey: tenantName,
+			},
+		},
+		{
+			ID:    "831ae1d0-f441-44b1-bb2a-46548048e26f",
+			Class: testClass.Class,
+			Properties: map[string]interface{}{
+				tenantKey: tenantName,
+			},
+		},
+		{
+			ID:    "6f3363e0-c0a0-4618-bf1f-b6cad9cdff59",
+			Class: testClass.Class,
+			Properties: map[string]interface{}{
+				tenantKey: tenantName,
+			},
+		},
+	}
+
+	defer func() {
+		helper.DeleteClass(t, testClass.Class)
+	}()
+
+	t.Run("create class with multi-tenancy enabled", func(t *testing.T) {
+		helper.CreateClass(t, &testClass)
+	})
+
+	t.Run("create tenants", func(t *testing.T) {
+		tenants := make([]*models.Tenant, len(tenantObjects))
+		for i := range tenants {
+			tenants[i] = &models.Tenant{tenantName}
+		}
+		helper.CreateTenants(t, testClass.Class, tenants)
+	})
+
+	t.Run("add tenant objects", func(t *testing.T) {
+		resp, err := helper.CreateTenantObjectsBatch(t, tenantObjects, tenantName)
+		require.Nil(t, err)
+		for _, elem := range resp {
+			if !assert.Nil(t, elem.Result.Errors) {
+				t.Logf("expected nil, got: %v",
+					elem.Result.Errors.Error[0].Message)
+			}
+		}
+	})
+
+	t.Run("get tenant objects", func(t *testing.T) {
+		for _, obj := range tenantObjects {
+			resp, err := helper.TenantObject(t, obj.Class, obj.ID, tenantName)
+			require.Nil(t, err)
+			assert.Equal(t, obj.ID, resp.ID)
+			assert.Equal(t, obj.Class, resp.Class)
+			assert.Equal(t, obj.Properties, resp.Properties)
+		}
+	})
+}
+
+func TestBatchAddTenantObjects_MixedTenants(t *testing.T) {
 	tenantKey := "tenantName"
 	testClass := models.Class{
 		Class: "MultiTenantClass",
@@ -79,16 +158,86 @@ func TestBatchAddTenantObjects(t *testing.T) {
 	})
 
 	t.Run("add tenant objects", func(t *testing.T) {
-		helper.CreateObjectsBatch(t, tenantObjects)
+		otherTenant := "SomeOtherKey"
+		resp, err := helper.CreateTenantObjectsBatch(t, tenantObjects, otherTenant)
+		require.Nil(t, err)
+		for _, obj := range resp {
+			require.NotNil(t, obj.Result)
+			require.NotNil(t, obj.Result.Errors)
+			require.NotNil(t, obj.Result.Errors)
+			require.Len(t, obj.Result.Errors.Error, 1)
+			assert.Contains(t, obj.Result.Errors.Error[0].Message,
+				fmt.Sprintf("object does not belong to tenant %q", otherTenant))
+		}
 	})
+}
 
-	t.Run("get tenant objects", func(t *testing.T) {
-		for i, obj := range tenantObjects {
-			resp, err := helper.TenantObject(t, obj.Class, obj.ID, tenantNames[i])
-			require.Nil(t, err)
-			assert.Equal(t, obj.ID, resp.ID)
-			assert.Equal(t, obj.Class, resp.Class)
-			assert.Equal(t, obj.Properties, resp.Properties)
+func TestBatchAddTenantObjects_MissingTenantKey(t *testing.T) {
+	tenantKey := "tenantName"
+	testClass := models.Class{
+		Class: "MultiTenantClass",
+		MultiTenancyConfig: &models.MultiTenancyConfig{
+			Enabled:   true,
+			TenantKey: tenantKey,
+		},
+		Properties: []*models.Property{
+			{
+				Name:     tenantKey,
+				DataType: []string{"string"},
+			},
+		},
+	}
+	tenantName := "Tenant1"
+	tenantObjects := []*models.Object{
+		{
+			ID:    "0927a1e0-398e-4e76-91fb-04a7a8f0405c",
+			Class: testClass.Class,
+			Properties: map[string]interface{}{
+				tenantKey: tenantName,
+			},
+		},
+		{
+			ID:    "831ae1d0-f441-44b1-bb2a-46548048e26f",
+			Class: testClass.Class,
+			Properties: map[string]interface{}{
+				tenantKey: tenantName,
+			},
+		},
+		{
+			ID:    "6f3363e0-c0a0-4618-bf1f-b6cad9cdff59",
+			Class: testClass.Class,
+			Properties: map[string]interface{}{
+				tenantKey: tenantName,
+			},
+		},
+	}
+
+	defer func() {
+		helper.DeleteClass(t, testClass.Class)
+	}()
+
+	t.Run("create class with multi-tenancy enabled", func(t *testing.T) {
+		helper.CreateClass(t, &testClass)
+	})
+
+	t.Run("create tenants", func(t *testing.T) {
+		tenants := make([]*models.Tenant, len(tenantObjects))
+		for i := range tenants {
+			tenants[i] = &models.Tenant{tenantName}
+		}
+		helper.CreateTenants(t, testClass.Class, tenants)
+	})
+
+	t.Run("add tenant objects", func(t *testing.T) {
+		resp, err := helper.CreateTenantObjectsBatch(t, tenantObjects, "")
+		require.Nil(t, err)
+		for _, obj := range resp {
+			require.NotNil(t, obj.Result)
+			require.NotNil(t, obj.Result.Errors)
+			require.NotNil(t, obj.Result.Errors)
+			require.Len(t, obj.Result.Errors.Error, 1)
+			assert.Contains(t, obj.Result.Errors.Error[0].Message,
+				`class "MultiTenantClass" has multi-tenancy enabled, tenant_key "tenantName" required`)
 		}
 	})
 }
diff --git a/test/acceptance/multi_tenancy/batch_delete_tenant_objects_test.go b/test/acceptance/multi_tenancy/batch_delete_tenant_objects_test.go
index 93f0018f03..62bf195787 100644
--- a/test/acceptance/multi_tenancy/batch_delete_tenant_objects_test.go
+++ b/test/acceptance/multi_tenancy/batch_delete_tenant_objects_test.go
@@ -16,6 +16,7 @@ import (
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"github.com/weaviate/weaviate/client/batch"
 	"github.com/weaviate/weaviate/client/objects"
 	"github.com/weaviate/weaviate/entities/filters"
 	"github.com/weaviate/weaviate/entities/models"
@@ -38,29 +39,27 @@ func TestBatchDeleteTenantObjects(t *testing.T) {
 			},
 		},
 	}
-	tenantNames := []string{
-		"Tenant1", "Tenant2", "Tenant3",
-	}
+	tenantName := "Tenant1"
 	tenantObjects := []*models.Object{
 		{
 			ID:    "0927a1e0-398e-4e76-91fb-04a7a8f0405c",
-			Class: className,
+			Class: testClass.Class,
 			Properties: map[string]interface{}{
-				tenantKey: tenantNames[0],
+				tenantKey: tenantName,
 			},
 		},
 		{
 			ID:    "831ae1d0-f441-44b1-bb2a-46548048e26f",
-			Class: className,
+			Class: testClass.Class,
 			Properties: map[string]interface{}{
-				tenantKey: tenantNames[1],
+				tenantKey: tenantName,
 			},
 		},
 		{
 			ID:    "6f3363e0-c0a0-4618-bf1f-b6cad9cdff59",
-			Class: className,
+			Class: testClass.Class,
 			Properties: map[string]interface{}{
-				tenantKey: tenantNames[2],
+				tenantKey: tenantName,
 			},
 		},
 	}
@@ -74,19 +73,26 @@ func TestBatchDeleteTenantObjects(t *testing.T) {
 	})
 
 	t.Run("create tenants", func(t *testing.T) {
-		tenants := make([]*models.Tenant, len(tenantNames))
+		tenants := make([]*models.Tenant, len(tenantObjects))
 		for i := range tenants {
-			tenants[i] = &models.Tenant{tenantNames[i]}
+			tenants[i] = &models.Tenant{tenantName}
 		}
 		helper.CreateTenants(t, className, tenants)
 	})
 
 	t.Run("add tenant objects", func(t *testing.T) {
-		helper.CreateObjectsBatch(t, tenantObjects)
+		resp, err := helper.CreateTenantObjectsBatch(t, tenantObjects, tenantName)
+		require.Nil(t, err)
+		for _, elem := range resp {
+			if !assert.Nil(t, elem.Result.Errors) {
+				t.Logf("expected nil, got: %v",
+					elem.Result.Errors.Error[0].Message)
+			}
+		}
 
 		t.Run("verify tenant objects", func(t *testing.T) {
-			for i, obj := range tenantObjects {
-				resp, err := helper.TenantObject(t, obj.Class, obj.ID, tenantNames[i])
+			for _, obj := range tenantObjects {
+				resp, err := helper.TenantObject(t, obj.Class, obj.ID, tenantName)
 				require.Nil(t, err)
 				require.Equal(t, obj.ID, resp.ID)
 				require.Equal(t, obj.Class, resp.Class)
@@ -107,11 +113,17 @@ func TestBatchDeleteTenantObjects(t *testing.T) {
 			Where: &where,
 		}
 		batch := models.BatchDelete{Match: &match}
-		helper.DeleteObjectsBatch(t, &batch)
+		resp, err := helper.DeleteTenantObjectsBatch(t, &batch, tenantName)
+		require.Nil(t, err)
+		require.NotNil(t, resp)
+		require.NotNil(t, resp.Results)
+		assert.Nil(t, resp.Results.Objects)
+		assert.Equal(t, int64(3), resp.Results.Successful)
+		assert.Equal(t, int64(0), resp.Results.Failed)
 
 		t.Run("verify tenant object deletion", func(t *testing.T) {
-			for i, obj := range tenantObjects {
-				resp, err := helper.TenantObject(t, obj.Class, obj.ID, tenantNames[i])
+			for _, obj := range tenantObjects {
+				resp, err := helper.TenantObject(t, obj.Class, obj.ID, tenantName)
 				assert.Nil(t, resp)
 				assert.NotNil(t, err)
 				assert.EqualError(t, objects.NewObjectsClassGetNotFound(), err.Error())
@@ -119,3 +131,106 @@ func TestBatchDeleteTenantObjects(t *testing.T) {
 		})
 	})
 }
+
+func TestBatchDeleteTenantObjects_MissingTenantKey(t *testing.T) {
+	className := "MultiTenantClass"
+	tenantKey := "tenantName"
+	testClass := models.Class{
+		Class: className,
+		MultiTenancyConfig: &models.MultiTenancyConfig{
+			Enabled:   true,
+			TenantKey: tenantKey,
+		},
+		Properties: []*models.Property{
+			{
+				Name:     tenantKey,
+				DataType: []string{"string"},
+			},
+		},
+	}
+	tenantName := "Tenant1"
+	tenantObjects := []*models.Object{
+		{
+			ID:    "0927a1e0-398e-4e76-91fb-04a7a8f0405c",
+			Class: testClass.Class,
+			Properties: map[string]interface{}{
+				tenantKey: tenantName,
+			},
+		},
+		{
+			ID:    "831ae1d0-f441-44b1-bb2a-46548048e26f",
+			Class: testClass.Class,
+			Properties: map[string]interface{}{
+				tenantKey: tenantName,
+			},
+		},
+		{
+			ID:    "6f3363e0-c0a0-4618-bf1f-b6cad9cdff59",
+			Class: testClass.Class,
+			Properties: map[string]interface{}{
+				tenantKey: tenantName,
+			},
+		},
+	}
+
+	defer func() {
+		helper.DeleteClass(t, className)
+	}()
+
+	t.Run("create class with multi-tenancy enabled", func(t *testing.T) {
+		helper.CreateClass(t, &testClass)
+	})
+
+	t.Run("create tenants", func(t *testing.T) {
+		tenants := make([]*models.Tenant, len(tenantObjects))
+		for i := range tenants {
+			tenants[i] = &models.Tenant{tenantName}
+		}
+		helper.CreateTenants(t, className, tenants)
+	})
+
+	t.Run("add tenant objects", func(t *testing.T) {
+		resp, err := helper.CreateTenantObjectsBatch(t, tenantObjects, tenantName)
+		require.Nil(t, err)
+		for _, elem := range resp {
+			if !assert.Nil(t, elem.Result.Errors) {
+				t.Logf("expected nil, got: %v",
+					elem.Result.Errors.Error[0].Message)
+			}
+		}
+
+		t.Run("verify tenant objects", func(t *testing.T) {
+			for _, obj := range tenantObjects {
+				resp, err := helper.TenantObject(t, obj.Class, obj.ID, tenantName)
+				require.Nil(t, err)
+				require.Equal(t, obj.ID, resp.ID)
+				require.Equal(t, obj.Class, resp.Class)
+				require.Equal(t, obj.Properties, resp.Properties)
+			}
+		})
+	})
+
+	t.Run("batch delete tenant objects", func(t *testing.T) {
+		glob := "*"
+		where := models.WhereFilter{
+			Operator:    filters.OperatorLike.Name(),
+			Path:        []string{"id"},
+			ValueString: &glob,
+		}
+		match := models.BatchDeleteMatch{
+			Class: className,
+			Where: &where,
+		}
+		params := batch.NewBatchObjectsDeleteParams().WithBody(&models.BatchDelete{Match: &match})
+		resp, err := helper.Client(t).Batch.BatchObjectsDelete(params, nil)
+		badReqErr, ok := err.(*batch.BatchObjectsDeleteUnprocessableEntity)
+		require.Nil(t, resp)
+		require.NotNil(t, err)
+		require.True(t, ok, "expected *batch.BatchObjectsDeleteUnprocessableEntity, got: %T", err)
+		require.NotNil(t, badReqErr.Payload)
+		require.NotNil(t, badReqErr.Payload.Error)
+		require.Len(t, badReqErr.Payload.Error, 1)
+		assert.Contains(t, badReqErr.Payload.Error[0].Message,
+			`class "MultiTenantClass" has multi-tenancy enabled, tenant_key "tenantName" required`)
+	})
+}
diff --git a/test/acceptance/multi_tenancy/patch_tenant_objects_test.go b/test/acceptance/multi_tenancy/patch_tenant_objects_test.go
index c11b510d00..a6b8d9378e 100644
--- a/test/acceptance/multi_tenancy/patch_tenant_objects_test.go
+++ b/test/acceptance/multi_tenancy/patch_tenant_objects_test.go
@@ -13,9 +13,12 @@ package test
 
 import (
 	"fmt"
+	"net/http"
 	"testing"
 
+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"github.com/weaviate/weaviate/client/objects"
 	"github.com/weaviate/weaviate/entities/models"
 	"github.com/weaviate/weaviate/test/helper"
 )
@@ -128,3 +131,67 @@ func TestPatchTenantObjects(t *testing.T) {
 		})
 	})
 }
+
+func TestPatchTenantObjects_ChangeTenantKey(t *testing.T) {
+	className := "MultiTenantClass"
+	tenantKey := "tenantName"
+	tenantName := "Tenant1"
+	testClass := models.Class{
+		Class: className,
+		MultiTenancyConfig: &models.MultiTenancyConfig{
+			Enabled:   true,
+			TenantKey: tenantKey,
+		},
+		Properties: []*models.Property{
+			{
+				Name:     tenantKey,
+				DataType: []string{"string"},
+			},
+		},
+	}
+	tenantObject := models.Object{
+		ID:    "0927a1e0-398e-4e76-91fb-04a7a8f0405c",
+		Class: className,
+		Properties: map[string]interface{}{
+			tenantKey: tenantName,
+		},
+	}
+
+	defer func() {
+		helper.DeleteClass(t, className)
+	}()
+
+	t.Run("create class with multi-tenancy enabled", func(t *testing.T) {
+		helper.CreateClass(t, &testClass)
+		helper.CreateTenants(t, className, []*models.Tenant{{tenantName}})
+	})
+
+	t.Run("add tenant object", func(t *testing.T) {
+		params := objects.NewObjectsCreateParams().
+			WithBody(&tenantObject).WithTenantKey(&tenantName)
+		_, err := helper.Client(t).Objects.ObjectsCreate(params, nil)
+		require.Nil(t, err)
+	})
+
+	t.Run("patch tenant object", func(t *testing.T) {
+		toUpdate := models.Object{
+			Class: testClass.Class,
+			ID:    tenantObject.ID,
+			Properties: map[string]interface{}{
+				tenantKey: "updatedTenantName",
+			},
+		}
+		params := objects.NewObjectsClassPatchParams().WithClassName(toUpdate.Class).
+			WithID(toUpdate.ID).WithBody(&toUpdate).WithTenantKey(&tenantName)
+		resp, err := helper.Client(t).Objects.ObjectsClassPatch(params, nil)
+		require.Nil(t, resp)
+		require.NotNil(t, err)
+		parsedErr, ok := err.(*objects.ObjectsClassPatchUnprocessableEntity)
+		require.True(t, ok)
+		require.NotNil(t, parsedErr.Payload.Error)
+		require.Len(t, parsedErr.Payload.Error, 1)
+		assert.Contains(t, err.Error(), fmt.Sprint(http.StatusUnprocessableEntity))
+		expected := "tenant key \"tenantName\" is immutable"
+		assert.Contains(t, parsedErr.Payload.Error[0].Message, expected)
+	})
+}
diff --git a/test/acceptance/multi_tenancy/update_tenant_objects_test.go b/test/acceptance/multi_tenancy/update_tenant_objects_test.go
index 73b0903de6..ff2f4a8094 100644
--- a/test/acceptance/multi_tenancy/update_tenant_objects_test.go
+++ b/test/acceptance/multi_tenancy/update_tenant_objects_test.go
@@ -13,9 +13,12 @@ package test
 
 import (
 	"fmt"
+	"net/http"
 	"testing"
 
+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"github.com/weaviate/weaviate/client/objects"
 	"github.com/weaviate/weaviate/entities/models"
 	"github.com/weaviate/weaviate/test/helper"
 )
@@ -128,3 +131,129 @@ func TestUpdateTenantObjects(t *testing.T) {
 		})
 	})
 }
+
+func TestUpdateTenantObjects_MissingTenantKey(t *testing.T) {
+	className := "MultiTenantClass"
+	tenantKey := "tenantName"
+	tenantName := "Tenant1"
+	testClass := models.Class{
+		Class: className,
+		MultiTenancyConfig: &models.MultiTenancyConfig{
+			Enabled:   true,
+			TenantKey: tenantKey,
+		},
+		Properties: []*models.Property{
+			{
+				Name:     tenantKey,
+				DataType: []string{"string"},
+			},
+		},
+	}
+	tenantObject := models.Object{
+		ID:    "0927a1e0-398e-4e76-91fb-04a7a8f0405c",
+		Class: className,
+		Properties: map[string]interface{}{
+			tenantKey: tenantName,
+		},
+	}
+
+	defer func() {
+		helper.DeleteClass(t, className)
+	}()
+
+	t.Run("create class with multi-tenancy enabled", func(t *testing.T) {
+		helper.CreateClass(t, &testClass)
+		helper.CreateTenants(t, className, []*models.Tenant{{tenantName}})
+	})
+
+	t.Run("add tenant object", func(t *testing.T) {
+		params := objects.NewObjectsCreateParams().
+			WithBody(&tenantObject).WithTenantKey(&tenantName)
+		_, err := helper.Client(t).Objects.ObjectsCreate(params, nil)
+		require.Nil(t, err)
+	})
+
+	t.Run("update tenant object", func(t *testing.T) {
+		toUpdate := models.Object{
+			Class: testClass.Class,
+			ID:    tenantObject.ID,
+		}
+		params := objects.NewObjectsClassPutParams().WithClassName(toUpdate.Class).
+			WithID(toUpdate.ID).WithBody(&toUpdate).WithTenantKey(&tenantName)
+		resp, err := helper.Client(t).Objects.ObjectsClassPut(params, nil)
+		require.Nil(t, resp)
+		require.NotNil(t, err)
+		parsedErr, ok := err.(*objects.ObjectsClassPutUnprocessableEntity)
+		require.True(t, ok)
+		require.NotNil(t, parsedErr.Payload.Error)
+		require.Len(t, parsedErr.Payload.Error, 1)
+		assert.Contains(t, err.Error(), fmt.Sprint(http.StatusUnprocessableEntity))
+		expected := "tenant_key query param value \"Tenant1\" conflicts with object " +
+			"body value \"\" for class \"MultiTenantClass\" tenant key \"tenantName\""
+		assert.Contains(t, parsedErr.Payload.Error[0].Message, expected)
+	})
+}
+
+func TestUpdateTenantObjects_UpdateTenantKey(t *testing.T) {
+	className := "MultiTenantClass"
+	tenantKey := "tenantName"
+	tenantName := "Tenant1"
+	testClass := models.Class{
+		Class: className,
+		MultiTenancyConfig: &models.MultiTenancyConfig{
+			Enabled:   true,
+			TenantKey: tenantKey,
+		},
+		Properties: []*models.Property{
+			{
+				Name:     tenantKey,
+				DataType: []string{"string"},
+			},
+		},
+	}
+	tenantObject := models.Object{
+		ID:    "0927a1e0-398e-4e76-91fb-04a7a8f0405c",
+		Class: className,
+		Properties: map[string]interface{}{
+			tenantKey: tenantName,
+		},
+	}
+
+	defer func() {
+		helper.DeleteClass(t, className)
+	}()
+
+	t.Run("create class with multi-tenancy enabled", func(t *testing.T) {
+		helper.CreateClass(t, &testClass)
+		helper.CreateTenants(t, className, []*models.Tenant{{tenantName}})
+	})
+
+	t.Run("add tenant object", func(t *testing.T) {
+		params := objects.NewObjectsCreateParams().
+			WithBody(&tenantObject).WithTenantKey(&tenantName)
+		_, err := helper.Client(t).Objects.ObjectsCreate(params, nil)
+		require.Nil(t, err)
+	})
+
+	t.Run("update tenant object", func(t *testing.T) {
+		toUpdate := models.Object{
+			Class: testClass.Class,
+			ID:    tenantObject.ID,
+			Properties: map[string]interface{}{
+				tenantKey: "updatedTenantName",
+			},
+		}
+		params := objects.NewObjectsClassPutParams().WithClassName(toUpdate.Class).
+			WithID(toUpdate.ID).WithBody(&toUpdate).WithTenantKey(&tenantName)
+		resp, err := helper.Client(t).Objects.ObjectsClassPut(params, nil)
+		require.Nil(t, resp)
+		require.NotNil(t, err)
+		parsedErr, ok := err.(*objects.ObjectsClassPutUnprocessableEntity)
+		require.True(t, ok)
+		require.NotNil(t, parsedErr.Payload.Error)
+		require.Len(t, parsedErr.Payload.Error, 1)
+		assert.Contains(t, err.Error(), fmt.Sprint(http.StatusUnprocessableEntity))
+		expected := "tenant key \"tenantName\" is immutable"
+		assert.Contains(t, parsedErr.Payload.Error[0].Message, expected)
+	})
+}
diff --git a/test/helper/journey/backup_journey.go b/test/helper/journey/backup_journey.go
index 960cf0aa1d..bed0851ce0 100644
--- a/test/helper/journey/backup_journey.go
+++ b/test/helper/journey/backup_journey.go
@@ -12,6 +12,7 @@
 package journey
 
 import (
+	"fmt"
 	"math/rand"
 	"testing"
 	"time"
@@ -32,6 +33,11 @@ const (
 	clusterJourney
 )
 
+const (
+	singleTenant = ""
+	multiTenant  = "tenantID"
+)
+
 func backupJourney(t *testing.T, className, backend, backupID string, journeyType journeyType) {
 	if journeyType == clusterJourney && backend == "filesystem" {
 		t.Run("should fail backup/restore with local filesystem backend", func(t *testing.T) {
@@ -123,7 +129,7 @@ func backupJourney(t *testing.T, className, backend, backupID string, journeyTyp
 	assert.Equal(t, int64(500), count)
 }
 
-func addTestClass(t *testing.T, className string) {
+func addTestClass(t *testing.T, className string, tenantKey string) {
 	class := &models.Class{
 		Class: className,
 		ModuleConfig: map[string]interface{}{
@@ -140,10 +146,21 @@ func addTestClass(t *testing.T, className string) {
 		},
 	}
 
+	if tenantKey != singleTenant {
+		class.Properties = append(class.Properties, &models.Property{
+			Name:     multiTenant,
+			DataType: []string{"string"},
+		})
+		class.MultiTenancyConfig = &models.MultiTenancyConfig{
+			Enabled:   true,
+			TenantKey: multiTenant,
+		}
+	}
+
 	helper.CreateClass(t, class)
 }
 
-func addTestObjects(t *testing.T, className string) {
+func addTestObjects(t *testing.T, className string, tenantKey string) {
 	const (
 		noteLengthMin = 4
 		noteLengthMax = 1024
@@ -160,10 +177,14 @@ func addTestObjects(t *testing.T, className string) {
 			contentsLength := noteLengthMin + seededRand.Intn(noteLengthMax-noteLengthMin+1)
 			contents := helper.GetRandomString(contentsLength)
 
-			batch[j] = &models.Object{
+			obj := models.Object{
 				Class:      className,
 				Properties: map[string]interface{}{"contents": contents},
 			}
+			if tenantKey != singleTenant {
+				obj.Properties.(map[string]interface{})[multiTenant] = fmt.Sprintf("Tenant%d", i)
+			}
+			batch[j] = &obj
 		}
 
 		helper.CreateObjectsBatch(t, batch)
diff --git a/test/helper/journey/backup_journey_tests.go b/test/helper/journey/backup_journey_tests.go
index 66f0478c85..9d6b042959 100644
--- a/test/helper/journey/backup_journey_tests.go
+++ b/test/helper/journey/backup_journey_tests.go
@@ -16,28 +16,49 @@ import (
 )
 
 // BackupJourneyTests_SingleNode this method gathers all backup related e2e tests
-func BackupJourneyTests_SingleNode(t *testing.T, weaviateEndpoint, backend, className, backupID string) {
-	// This is a simple test which covers almost the same scenario as singleNodeBackupJourneyTest
-	// but is left here to be expanded in the future with a more complex example
-	// like adding there a new reference property and trying to run the test with 2 classes which
-	// one of those classes is a class with a reference property
-	t.Run("backup and restore Books", func(t *testing.T) {
-		backupAndRestoreJourneyTest(t, weaviateEndpoint, backend)
-	})
+func BackupJourneyTests_SingleNode(t *testing.T, weaviateEndpoint, backend, className, backupID string, tenantNames []string) {
+	if len(tenantNames) > 0 {
+		t.Run("multi-tenant single node backup", func(t *testing.T) {
+			singleNodeBackupJourneyTest(t, weaviateEndpoint, backend, className, backupID, tenantNames)
+		})
+	} else {
+		// This is a simple test which covers almost the same scenario as singleNodeBackupJourneyTest
+		// but is left here to be expanded in the future with a more complex example
+		// like adding there a new reference property and trying to run the test with 2 classes which
+		// one of those classes is a class with a reference property
+		t.Run("backup and restore Books", func(t *testing.T) {
+			backupAndRestoreJourneyTest(t, weaviateEndpoint, backend)
+		})
 
-	t.Run("single node backup", func(t *testing.T) {
-		singleNodeBackupJourneyTest(t, weaviateEndpoint, backend, className, backupID)
-	})
+		t.Run("single-tenant single node backup", func(t *testing.T) {
+			singleNodeBackupJourneyTest(t, weaviateEndpoint, backend, className, backupID, nil)
+		})
+	}
 }
 
 // BackupJourneyTests_Cluster this method gathers all backup related e2e tests to be run on a cluster
-func BackupJourneyTests_Cluster(t *testing.T, backend, className, backupID string, weaviateEndpoints ...string) {
-	t.Run("cluster backup", func(t *testing.T) {
-		if len(weaviateEndpoints) <= 1 {
-			t.Fatal("must provide more than one node for cluster backup test")
-		}
+func BackupJourneyTests_Cluster(t *testing.T, backend, className, backupID string,
+	tenantNames []string, weaviateEndpoints ...string,
+) {
+	if len(tenantNames) > 0 {
+		t.Run("multi-tenant cluster backup", func(t *testing.T) {
+			if len(weaviateEndpoints) <= 1 {
+				t.Fatal("must provide more than one node for cluster backup test")
+			}
 
-		coordinator := weaviateEndpoints[0]
-		clusterBackupJourneyTest(t, backend, className, backupID, coordinator, weaviateEndpoints[1:]...)
-	})
+			coordinator := weaviateEndpoints[0]
+			clusterBackupJourneyTest(t, backend, className, backupID,
+				coordinator, tenantNames, weaviateEndpoints[1:]...)
+		})
+	} else {
+		t.Run("single-tenant cluster backup", func(t *testing.T) {
+			if len(weaviateEndpoints) <= 1 {
+				t.Fatal("must provide more than one node for cluster backup test")
+			}
+
+			coordinator := weaviateEndpoints[0]
+			clusterBackupJourneyTest(t, backend, className, backupID,
+				coordinator, nil, weaviateEndpoints[1:]...)
+		})
+	}
 }
diff --git a/test/helper/journey/cluster_backup_journey.go b/test/helper/journey/cluster_backup_journey.go
index 94dfce8b3c..d15dc87646 100644
--- a/test/helper/journey/cluster_backup_journey.go
+++ b/test/helper/journey/cluster_backup_journey.go
@@ -16,19 +16,35 @@ import (
 	"math/rand"
 	"testing"
 
+	"github.com/weaviate/weaviate/entities/models"
 	"github.com/weaviate/weaviate/test/helper"
 )
 
-func clusterBackupJourneyTest(t *testing.T, backend, className, backupID, coordinatorEndpoint string, nodeEndpoints ...string) {
+func clusterBackupJourneyTest(t *testing.T, backend, className, backupID,
+	coordinatorEndpoint string, tenantNames []string, nodeEndpoints ...string,
+) {
 	uploaderEndpoint := nodeEndpoints[rand.Intn(len(nodeEndpoints))]
 	helper.SetupClient(uploaderEndpoint)
 	t.Logf("uploader selected -> %s:%s", helper.ServerHost, helper.ServerPort)
 
-	// upload data to a node other than the coordinator
-	t.Run(fmt.Sprintf("add test data to endpoint: %s", uploaderEndpoint), func(t *testing.T) {
-		addTestClass(t, className)
-		addTestObjects(t, className)
-	})
+	if len(tenantNames) > 0 {
+		// upload data to a node other than the coordinator
+		t.Run(fmt.Sprintf("add test data to endpoint: %s", uploaderEndpoint), func(t *testing.T) {
+			addTestClass(t, className, multiTenant)
+			tenants := make([]*models.Tenant, len(tenantNames))
+			for i := range tenantNames {
+				tenants[i] = &models.Tenant{Name: tenantNames[i]}
+			}
+			helper.CreateTenants(t, className, tenants)
+			addTestObjects(t, className, multiTenant)
+		})
+	} else {
+		// upload data to a node other than the coordinator
+		t.Run(fmt.Sprintf("add test data to endpoint: %s", uploaderEndpoint), func(t *testing.T) {
+			addTestClass(t, className, singleTenant)
+			addTestObjects(t, className, singleTenant)
+		})
+	}
 
 	helper.SetupClient(coordinatorEndpoint)
 	t.Logf("coordinator selected -> %s:%s", helper.ServerHost, helper.ServerPort)
diff --git a/test/helper/journey/single_node_backup_journey.go b/test/helper/journey/single_node_backup_journey.go
index 21d1a140b4..f8f589cb52 100644
--- a/test/helper/journey/single_node_backup_journey.go
+++ b/test/helper/journey/single_node_backup_journey.go
@@ -14,18 +14,34 @@ package journey
 import (
 	"testing"
 
+	"github.com/weaviate/weaviate/entities/models"
 	"github.com/weaviate/weaviate/test/helper"
 )
 
-func singleNodeBackupJourneyTest(t *testing.T, weaviateEndpoint, backend, className, backupID string) {
+func singleNodeBackupJourneyTest(t *testing.T,
+	weaviateEndpoint, backend, className, backupID string,
+	tenantNames []string,
+) {
 	if weaviateEndpoint != "" {
 		helper.SetupClient(weaviateEndpoint)
 	}
 
-	t.Run("add test data", func(t *testing.T) {
-		addTestClass(t, className)
-		addTestObjects(t, className)
-	})
+	if len(tenantNames) > 0 {
+		t.Run("add test data", func(t *testing.T) {
+			addTestClass(t, className, multiTenant)
+			tenants := make([]*models.Tenant, len(tenantNames))
+			for i := range tenantNames {
+				tenants[i] = &models.Tenant{Name: tenantNames[i]}
+			}
+			helper.CreateTenants(t, className, tenants)
+			addTestObjects(t, className, multiTenant)
+		})
+	} else {
+		t.Run("add test data", func(t *testing.T) {
+			addTestClass(t, className, singleTenant)
+			addTestObjects(t, className, singleTenant)
+		})
+	}
 
 	t.Run("single node backup", func(t *testing.T) {
 		backupJourney(t, className, backend, backupID, singleNodeJourney)
diff --git a/test/helper/objects.go b/test/helper/objects.go
index a11a095c6d..b90dc514db 100644
--- a/test/helper/objects.go
+++ b/test/helper/objects.go
@@ -90,6 +90,18 @@ func CreateObjectsBatch(t *testing.T, objects []*models.Object) {
 	}
 }
 
+func CreateTenantObjectsBatch(t *testing.T, objects []*models.Object,
+	tenantKey string,
+) ([]*models.ObjectsGetResponse, error) {
+	params := batch.NewBatchObjectsCreateParams().
+		WithBody(batch.BatchObjectsCreateBody{
+			Objects: objects,
+		}).WithTenantKey(&tenantKey)
+	resp, err := Client(t).Batch.BatchObjectsCreate(params, nil)
+	AssertRequestOk(t, resp, err, nil)
+	return resp.Payload, err
+}
+
 func UpdateObject(t *testing.T, object *models.Object) {
 	params := objects.NewObjectsUpdateParams().WithID(object.ID).WithBody(object)
 	resp, err := Client(t).Objects.ObjectsUpdate(params, nil)
@@ -143,6 +155,15 @@ func DeleteObjectsBatch(t *testing.T, body *models.BatchDelete) {
 	AssertRequestOk(t, resp, err, nil)
 }
 
+func DeleteTenantObjectsBatch(t *testing.T, body *models.BatchDelete,
+	tenantKey string,
+) (*models.BatchDeleteResponse, error) {
+	params := batch.NewBatchObjectsDeleteParams().
+		WithBody(body).WithTenantKey(&tenantKey)
+	resp, err := Client(t).Batch.BatchObjectsDelete(params, nil)
+	return resp.Payload, err
+}
+
 func AddReferences(t *testing.T, refs []*models.BatchReference) {
 	params := batch.NewBatchReferencesCreateParams().WithBody(refs)
 	resp, err := Client(t).Batch.BatchReferencesCreate(params, nil)
diff --git a/test/modules/backup-azure/backup_journey_test.go b/test/modules/backup-azure/backup_journey_test.go
index ec163f8e62..b473f7a5f5 100644
--- a/test/modules/backup-azure/backup_journey_test.go
+++ b/test/modules/backup-azure/backup_journey_test.go
@@ -60,7 +60,7 @@ func Test_BackupJourney(t *testing.T) {
 
 		t.Run("backup-azure", func(t *testing.T) {
 			journey.BackupJourneyTests_SingleNode(t, compose.GetWeaviate().URI(),
-				"azure", azureBackupJourneyClassName, azureBackupJourneyBackupIDSingleNode)
+				"azure", azureBackupJourneyClassName, azureBackupJourneyBackupIDSingleNode, nil)
 		})
 
 		require.Nil(t, compose.Terminate(ctx))
@@ -86,7 +86,7 @@ func Test_BackupJourney(t *testing.T) {
 
 		t.Run("backup-azure", func(t *testing.T) {
 			journey.BackupJourneyTests_Cluster(t, "azure", azureBackupJourneyClassName,
-				azureBackupJourneyBackupIDCluster, compose.GetWeaviate().URI(), compose.GetWeaviateNode2().URI())
+				azureBackupJourneyBackupIDCluster, nil, compose.GetWeaviate().URI(), compose.GetWeaviateNode2().URI())
 		})
 
 		require.Nil(t, compose.Terminate(ctx))
diff --git a/test/modules/backup-azure/multi_tenant_backup_test.go b/test/modules/backup-azure/multi_tenant_backup_test.go
new file mode 100644
index 0000000000..284557cce8
--- /dev/null
+++ b/test/modules/backup-azure/multi_tenant_backup_test.go
@@ -0,0 +1,89 @@
+//                           _       _
+// __      _____  __ ___   ___  __ _| |_ ___
+// \ \ /\ / / _ \/ _` \ \ / / |/ _` | __/ _ \
+//  \ V  V /  __/ (_| |\ V /| | (_| | ||  __/
+//   \_/\_/ \___|\__,_| \_/ |_|\__,_|\__\___|
+//
+//  Copyright © 2016 - 2023 Weaviate B.V. All rights reserved.
+//
+//  CONTACT: hello@weaviate.io
+//
+
+package test
+
+import (
+	"context"
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+	"github.com/weaviate/weaviate/test/docker"
+	"github.com/weaviate/weaviate/test/helper"
+	"github.com/weaviate/weaviate/test/helper/journey"
+	moduleshelper "github.com/weaviate/weaviate/test/helper/modules"
+)
+
+const (
+	numTenants = 50
+)
+
+func Test_MultiTenantBackupJourney(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Minute)
+	defer cancel()
+
+	tenantNames := make([]string, numTenants)
+	for i := range tenantNames {
+		tenantNames[i] = fmt.Sprintf("Tenant%d", i)
+	}
+
+	t.Run("single node", func(t *testing.T) {
+		t.Log("pre-instance env setup")
+		t.Setenv(envAzureContainer, azureBackupJourneyContainerName)
+
+		compose, err := docker.New().
+			WithBackendAzure(azureBackupJourneyContainerName).
+			WithText2VecContextionary().
+			WithWeaviate().
+			Start(ctx)
+		require.Nil(t, err)
+
+		t.Log("post-instance env setup")
+		azuriteEndpoint := compose.GetAzurite().URI()
+		t.Setenv(envAzureEndpoint, azuriteEndpoint)
+		moduleshelper.CreateAzureContainer(ctx, t, azuriteEndpoint, azureBackupJourneyContainerName)
+		helper.SetupClient(compose.GetWeaviate().URI())
+
+		t.Run("backup-azure", func(t *testing.T) {
+			journey.BackupJourneyTests_SingleNode(t, compose.GetWeaviate().URI(),
+				"azure", azureBackupJourneyClassName, azureBackupJourneyBackupIDSingleNode, tenantNames)
+		})
+
+		require.Nil(t, compose.Terminate(ctx))
+	})
+
+	t.Run("multiple node", func(t *testing.T) {
+		t.Log("pre-instance env setup")
+		t.Setenv(envAzureContainer, azureBackupJourneyContainerName)
+
+		compose, err := docker.New().
+			WithBackendAzure(azureBackupJourneyContainerName).
+			WithText2VecContextionary().
+			WithWeaviateCluster().
+			Start(ctx)
+		require.Nil(t, err)
+
+		t.Log("post-instance env setup")
+		azuriteEndpoint := compose.GetAzurite().URI()
+		t.Setenv(envAzureEndpoint, azuriteEndpoint)
+		moduleshelper.CreateAzureContainer(ctx, t, azuriteEndpoint, azureBackupJourneyContainerName)
+		helper.SetupClient(compose.GetWeaviate().URI())
+
+		t.Run("backup-azure", func(t *testing.T) {
+			journey.BackupJourneyTests_Cluster(t, "azure", azureBackupJourneyClassName,
+				azureBackupJourneyBackupIDCluster, tenantNames, compose.GetWeaviate().URI(), compose.GetWeaviateNode2().URI())
+		})
+
+		require.Nil(t, compose.Terminate(ctx))
+	})
+}
diff --git a/test/modules/backup-filesystem/backup_journey_test.go b/test/modules/backup-filesystem/backup_journey_test.go
index 2c430aaed8..35044196e8 100644
--- a/test/modules/backup-filesystem/backup_journey_test.go
+++ b/test/modules/backup-filesystem/backup_journey_test.go
@@ -47,7 +47,7 @@ func Test_BackupJourney(t *testing.T) {
 
 		t.Run("backup-filesystem", func(t *testing.T) {
 			journey.BackupJourneyTests_SingleNode(t, compose.GetWeaviate().URI(),
-				"filesystem", fsBackupJourneyClassName, fsBackupJourneyBackupIDSingleNode)
+				"filesystem", fsBackupJourneyClassName, fsBackupJourneyBackupIDSingleNode, nil)
 		})
 	})
 
@@ -67,7 +67,7 @@ func Test_BackupJourney(t *testing.T) {
 
 		t.Run("backup-filesystem", func(t *testing.T) {
 			journey.BackupJourneyTests_Cluster(t, "filesystem", fsBackupJourneyClassName,
-				fsBackupJourneyBackupIDCluster, compose.GetWeaviate().URI(), compose.GetWeaviateNode2().URI())
+				fsBackupJourneyBackupIDCluster, nil, compose.GetWeaviate().URI(), compose.GetWeaviateNode2().URI())
 		})
 	})
 }
diff --git a/test/modules/backup-filesystem/multi_tenant_backup_test.go b/test/modules/backup-filesystem/multi_tenant_backup_test.go
new file mode 100644
index 0000000000..1333e10136
--- /dev/null
+++ b/test/modules/backup-filesystem/multi_tenant_backup_test.go
@@ -0,0 +1,56 @@
+//                           _       _
+// __      _____  __ ___   ___  __ _| |_ ___
+// \ \ /\ / / _ \/ _` \ \ / / |/ _` | __/ _ \
+//  \ V  V /  __/ (_| |\ V /| | (_| | ||  __/
+//   \_/\_/ \___|\__,_| \_/ |_|\__,_|\__\___|
+//
+//  Copyright © 2016 - 2023 Weaviate B.V. All rights reserved.
+//
+//  CONTACT: hello@weaviate.io
+//
+
+package test
+
+import (
+	"context"
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+	"github.com/weaviate/weaviate/test/docker"
+	"github.com/weaviate/weaviate/test/helper/journey"
+)
+
+const numTenants = 50
+
+func Test_MultiTenantBackup(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Minute)
+	defer cancel()
+
+	t.Run("single node", func(t *testing.T) {
+		compose, err := docker.New().
+			WithBackendFilesystem().
+			WithText2VecContextionary().
+			WithWeaviate().
+			Start(ctx)
+		require.Nil(t, err)
+
+		defer func() {
+			if err := compose.Terminate(ctx); err != nil {
+				t.Fatalf("failed to terminte test containers: %s", err.Error())
+			}
+		}()
+
+		t.Run("backup-filesystem", func(t *testing.T) {
+			tenantNames := make([]string, numTenants)
+			for i := range tenantNames {
+				tenantNames[i] = fmt.Sprintf("Tenant%d", i)
+			}
+
+			journey.BackupJourneyTests_SingleNode(t,
+				compose.GetWeaviate().URI(), "filesystem", fsBackupJourneyClassName,
+				fsBackupJourneyBackupIDSingleNode, tenantNames)
+		})
+	})
+}
diff --git a/test/modules/backup-gcs/backup_journey_test.go b/test/modules/backup-gcs/backup_journey_test.go
index b55cf9274b..8695e97d7e 100644
--- a/test/modules/backup-gcs/backup_journey_test.go
+++ b/test/modules/backup-gcs/backup_journey_test.go
@@ -68,7 +68,7 @@ func Test_BackupJourney(t *testing.T) {
 
 		t.Run("backup-gcs", func(t *testing.T) {
 			journey.BackupJourneyTests_SingleNode(t, compose.GetWeaviate().URI(),
-				"gcs", gcsBackupJourneyClassName, gcsBackupJourneyBackupIDSingleNode)
+				"gcs", gcsBackupJourneyClassName, gcsBackupJourneyBackupIDSingleNode, nil)
 		})
 	})
 
@@ -98,7 +98,7 @@ func Test_BackupJourney(t *testing.T) {
 
 		t.Run("backup-gcs", func(t *testing.T) {
 			journey.BackupJourneyTests_Cluster(t, "gcs", gcsBackupJourneyClassName,
-				gcsBackupJourneyBackupIDCluster, compose.GetWeaviate().URI(), compose.GetWeaviateNode2().URI())
+				gcsBackupJourneyBackupIDCluster, nil, compose.GetWeaviate().URI(), compose.GetWeaviateNode2().URI())
 		})
 	})
 }
diff --git a/test/modules/backup-gcs/multi_tenant_backup_test.go b/test/modules/backup-gcs/multi_tenant_backup_test.go
new file mode 100644
index 0000000000..9c808492d7
--- /dev/null
+++ b/test/modules/backup-gcs/multi_tenant_backup_test.go
@@ -0,0 +1,99 @@
+//                           _       _
+// __      _____  __ ___   ___  __ _| |_ ___
+// \ \ /\ / / _ \/ _` \ \ / / |/ _` | __/ _ \
+//  \ V  V /  __/ (_| |\ V /| | (_| | ||  __/
+//   \_/\_/ \___|\__,_| \_/ |_|\__,_|\__\___|
+//
+//  Copyright © 2016 - 2023 Weaviate B.V. All rights reserved.
+//
+//  CONTACT: hello@weaviate.io
+//
+
+package test
+
+import (
+	"context"
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+	"github.com/weaviate/weaviate/test/docker"
+	"github.com/weaviate/weaviate/test/helper"
+	"github.com/weaviate/weaviate/test/helper/journey"
+	moduleshelper "github.com/weaviate/weaviate/test/helper/modules"
+)
+
+const numTenants = 50
+
+func Test_MultiTenantBackupJourney(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Minute)
+	defer cancel()
+
+	tenantNames := make([]string, numTenants)
+	for i := range tenantNames {
+		tenantNames[i] = fmt.Sprintf("Tenant%d", i)
+	}
+
+	t.Run("single node", func(t *testing.T) {
+		t.Log("pre-instance env setup")
+		t.Setenv(envGCSCredentials, "")
+		t.Setenv(envGCSProjectID, gcsBackupJourneyProjectID)
+		t.Setenv(envGCSBucket, gcsBackupJourneyBucketName)
+
+		compose, err := docker.New().
+			WithBackendGCS(gcsBackupJourneyBucketName).
+			WithText2VecContextionary().
+			WithWeaviate().
+			Start(ctx)
+		require.Nil(t, err)
+		defer func() {
+			if err := compose.Terminate(ctx); err != nil {
+				t.Fatalf("failed to terminte test containers: %s", err.Error())
+			}
+		}()
+
+		t.Log("post-instance env setup")
+		t.Setenv(envGCSEndpoint, compose.GetGCS().URI())
+		t.Setenv(envGCSStorageEmulatorHost, compose.GetGCS().URI())
+		moduleshelper.CreateGCSBucket(ctx, t, gcsBackupJourneyProjectID, gcsBackupJourneyBucketName)
+		helper.SetupClient(compose.GetWeaviate().URI())
+
+		t.Run("backup-gcs", func(t *testing.T) {
+			journey.BackupJourneyTests_SingleNode(t, compose.GetWeaviate().URI(),
+				"gcs", gcsBackupJourneyClassName,
+				gcsBackupJourneyBackupIDSingleNode, tenantNames)
+		})
+	})
+
+	t.Run("multiple node", func(t *testing.T) {
+		t.Log("pre-instance env setup")
+		t.Setenv(envGCSCredentials, "")
+		t.Setenv(envGCSProjectID, gcsBackupJourneyProjectID)
+		t.Setenv(envGCSBucket, gcsBackupJourneyBucketName)
+
+		compose, err := docker.New().
+			WithBackendGCS(gcsBackupJourneyBucketName).
+			WithText2VecContextionary().
+			WithWeaviateCluster().
+			Start(ctx)
+		require.Nil(t, err)
+		defer func() {
+			if err := compose.Terminate(ctx); err != nil {
+				t.Fatalf("failed to terminte test containers: %s", err.Error())
+			}
+		}()
+
+		t.Log("post-instance env setup")
+		t.Setenv(envGCSEndpoint, compose.GetGCS().URI())
+		t.Setenv(envGCSStorageEmulatorHost, compose.GetGCS().URI())
+		moduleshelper.CreateGCSBucket(ctx, t, gcsBackupJourneyProjectID, gcsBackupJourneyBucketName)
+		helper.SetupClient(compose.GetWeaviate().URI())
+
+		t.Run("backup-gcs", func(t *testing.T) {
+			journey.BackupJourneyTests_Cluster(t, "gcs", gcsBackupJourneyClassName,
+				gcsBackupJourneyBackupIDCluster, tenantNames,
+				compose.GetWeaviate().URI(), compose.GetWeaviateNode2().URI())
+		})
+	})
+}
diff --git a/test/modules/backup-s3/backup_journey_test.go b/test/modules/backup-s3/backup_journey_test.go
index 2a793936f6..2068b1381c 100644
--- a/test/modules/backup-s3/backup_journey_test.go
+++ b/test/modules/backup-s3/backup_journey_test.go
@@ -69,7 +69,7 @@ func Test_BackupJourney(t *testing.T) {
 
 		t.Run("backup-s3", func(t *testing.T) {
 			journey.BackupJourneyTests_SingleNode(t, compose.GetWeaviate().URI(),
-				"s3", s3BackupJourneyClassName, s3BackupJourneyBackupIDSingleNode)
+				"s3", s3BackupJourneyClassName, s3BackupJourneyBackupIDSingleNode, nil)
 		})
 	})
 
@@ -101,7 +101,8 @@ func Test_BackupJourney(t *testing.T) {
 
 		t.Run("backup-s3", func(t *testing.T) {
 			journey.BackupJourneyTests_Cluster(t, "s3", s3BackupJourneyClassName,
-				s3BackupJourneyBackupIDCluster, compose.GetWeaviate().URI(), compose.GetWeaviateNode2().URI())
+				s3BackupJourneyBackupIDCluster, nil,
+				compose.GetWeaviate().URI(), compose.GetWeaviateNode2().URI())
 		})
 	})
 }
diff --git a/test/modules/backup-s3/multi_tenant_backup_test.go b/test/modules/backup-s3/multi_tenant_backup_test.go
new file mode 100644
index 0000000000..3d9538645d
--- /dev/null
+++ b/test/modules/backup-s3/multi_tenant_backup_test.go
@@ -0,0 +1,98 @@
+//                           _       _
+// __      _____  __ ___   ___  __ _| |_ ___
+// \ \ /\ / / _ \/ _` \ \ / / |/ _` | __/ _ \
+//  \ V  V /  __/ (_| |\ V /| | (_| | ||  __/
+//   \_/\_/ \___|\__,_| \_/ |_|\__,_|\__\___|
+//
+//  Copyright © 2016 - 2023 Weaviate B.V. All rights reserved.
+//
+//  CONTACT: hello@weaviate.io
+//
+
+package test
+
+import (
+	"context"
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+	"github.com/weaviate/weaviate/test/docker"
+	"github.com/weaviate/weaviate/test/helper"
+	"github.com/weaviate/weaviate/test/helper/journey"
+)
+
+const numTenants = 50
+
+func Test_MultiTenantBackupJourney(t *testing.T) {
+	tenantNames := make([]string, numTenants)
+	for i := range tenantNames {
+		tenantNames[i] = fmt.Sprintf("Tenant%d", i)
+	}
+
+	t.Run("single node", func(t *testing.T) {
+		ctx, cancel := context.WithTimeout(context.Background(), 30*time.Minute)
+		defer cancel()
+
+		t.Log("pre-instance env setup")
+		t.Setenv(envS3AccessKey, s3BackupJourneyAccessKey)
+		t.Setenv(envS3SecretKey, s3BackupJourneySecretKey)
+		t.Setenv(envS3Bucket, s3BackupJourneyBucketName)
+
+		compose, err := docker.New().
+			WithBackendS3(s3BackupJourneyBucketName).
+			WithText2VecContextionary().
+			WithWeaviate().
+			Start(ctx)
+		require.Nil(t, err)
+		defer func() {
+			if err := compose.Terminate(ctx); err != nil {
+				t.Fatalf("failed to terminte test containers: %s", err.Error())
+			}
+		}()
+
+		t.Run("post-instance env setup", func(t *testing.T) {
+			createBucket(ctx, t, compose.GetMinIO().URI(), s3BackupJourneyRegion, s3BackupJourneyBucketName)
+			helper.SetupClient(compose.GetWeaviate().URI())
+		})
+
+		t.Run("backup-s3", func(t *testing.T) {
+			journey.BackupJourneyTests_SingleNode(t, compose.GetWeaviate().URI(),
+				"s3", s3BackupJourneyClassName, s3BackupJourneyBackupIDSingleNode, tenantNames)
+		})
+	})
+
+	t.Run("multiple node", func(t *testing.T) {
+		ctx, cancel := context.WithTimeout(context.Background(), 30*time.Minute)
+		defer cancel()
+
+		t.Log("pre-instance env setup")
+		t.Setenv(envS3AccessKey, s3BackupJourneyAccessKey)
+		t.Setenv(envS3SecretKey, s3BackupJourneySecretKey)
+		t.Setenv(envS3Bucket, s3BackupJourneyBucketName)
+
+		compose, err := docker.New().
+			WithBackendS3(s3BackupJourneyBucketName).
+			WithText2VecContextionary().
+			WithWeaviateCluster().
+			Start(ctx)
+		require.Nil(t, err)
+		defer func() {
+			if err := compose.Terminate(ctx); err != nil {
+				t.Fatalf("failed to terminte test containers: %s", err.Error())
+			}
+		}()
+
+		t.Run("post-instance env setup", func(t *testing.T) {
+			createBucket(ctx, t, compose.GetMinIO().URI(), s3BackupJourneyRegion, s3BackupJourneyBucketName)
+			helper.SetupClient(compose.GetWeaviate().URI())
+		})
+
+		t.Run("backup-s3", func(t *testing.T) {
+			journey.BackupJourneyTests_Cluster(t, "s3", s3BackupJourneyClassName,
+				s3BackupJourneyBackupIDCluster, tenantNames,
+				compose.GetWeaviate().URI(), compose.GetWeaviateNode2().URI())
+		})
+	})
+}
diff --git a/usecases/classification/fakes_for_test.go b/usecases/classification/fakes_for_test.go
index 5bf167de9d..9b0bab1714 100644
--- a/usecases/classification/fakes_for_test.go
+++ b/usecases/classification/fakes_for_test.go
@@ -188,9 +188,7 @@ func (f *fakeVectorRepoKNN) VectorClassSearch(ctx context.Context,
 	return nil, fmt.Errorf("vector class search not implemented in fake")
 }
 
-func (f *fakeVectorRepoKNN) BatchPutObjects(ctx context.Context,
-	objects objects.BatchObjects, _ *additional.ReplicationProperties,
-) (objects.BatchObjects, error) {
+func (f *fakeVectorRepoKNN) BatchPutObjects(ctx context.Context, objects objects.BatchObjects, repl *additional.ReplicationProperties, tenantKey string) (objects.BatchObjects, error) {
 	f.Lock()
 	defer f.Unlock()
 
@@ -263,9 +261,7 @@ func (f *fakeVectorRepoContextual) ZeroShotSearch(ctx context.Context, vector []
 	panic("not implemented")
 }
 
-func (f *fakeVectorRepoContextual) BatchPutObjects(ctx context.Context,
-	objects objects.BatchObjects, _ *additional.ReplicationProperties,
-) (objects.BatchObjects, error) {
+func (f *fakeVectorRepoContextual) BatchPutObjects(ctx context.Context, objects objects.BatchObjects, repl *additional.ReplicationProperties, tenantKey string) (objects.BatchObjects, error) {
 	f.Lock()
 	defer f.Unlock()
 	for _, batchObject := range objects {
diff --git a/usecases/classification/integrationtest/classifier_integration_test.go b/usecases/classification/integrationtest/classifier_integration_test.go
index 8b3b08e1af..7ed6431304 100644
--- a/usecases/classification/integrationtest/classifier_integration_test.go
+++ b/usecases/classification/integrationtest/classifier_integration_test.go
@@ -81,7 +81,7 @@ func Test_Classifier_KNN_SaveConsistency(t *testing.T) {
 				}
 			}
 
-			res, err := vrepo.BatchPutObjects(context.Background(), bt, nil)
+			res, err := vrepo.BatchPutObjects(context.Background(), bt, nil, "")
 			require.Nil(t, err)
 			for _, elem := range res {
 				require.Nil(t, elem.Err)
@@ -98,7 +98,7 @@ func Test_Classifier_KNN_SaveConsistency(t *testing.T) {
 					Object:        elem.Object(),
 				}
 			}
-			res, err := vrepo.BatchPutObjects(context.Background(), bt, nil)
+			res, err := vrepo.BatchPutObjects(context.Background(), bt, nil, "")
 			require.Nil(t, err)
 			for _, elem := range res {
 				require.Nil(t, elem.Err)
@@ -217,7 +217,7 @@ func Test_Classifier_ZeroShot_SaveConsistency(t *testing.T) {
 				}
 			}
 
-			res, err := vrepo.BatchPutObjects(context.Background(), bt, nil)
+			res, err := vrepo.BatchPutObjects(context.Background(), bt, nil, "")
 			require.Nil(t, err)
 			for _, elem := range res {
 				require.Nil(t, elem.Err)
diff --git a/usecases/objects/authorization_test.go b/usecases/objects/authorization_test.go
index d3b10be117..cba0481bc5 100644
--- a/usecases/objects/authorization_test.go
+++ b/usecases/objects/authorization_test.go
@@ -200,19 +200,13 @@ func Test_BatchKinds_Authorization(t *testing.T) {
 	}
 
 	tests := []testCase{
-		// testCase{
-		// 	methodName:       "AddActions",
-		// 	additionalArgs:   []interface{}{[]*models.Object{}, []*string{}},
-		// 	expectedVerb:     "create",
-		// 	expectedResource: "batch/actions",
-		// },
-
 		{
 			methodName: "AddObjects",
 			additionalArgs: []interface{}{
 				[]*models.Object{},
 				[]*string{},
 				&additional.ReplicationProperties{},
+				"",
 			},
 			expectedVerb:     "create",
 			expectedResource: "batch/objects",
@@ -236,6 +230,7 @@ func Test_BatchKinds_Authorization(t *testing.T) {
 				(*bool)(nil),
 				(*string)(nil),
 				&additional.ReplicationProperties{},
+				"",
 			},
 			expectedVerb:     "delete",
 			expectedResource: "batch/objects",
diff --git a/usecases/objects/batch_add_test.go b/usecases/objects/batch_add_test.go
index f4a60a1343..8c2a5ab7da 100644
--- a/usecases/objects/batch_add_test.go
+++ b/usecases/objects/batch_add_test.go
@@ -84,7 +84,7 @@ func Test_BatchManager_AddObjects_WithNoVectorizerModule(t *testing.T) {
 		expectedErr := NewErrInvalidUserInput("invalid param 'objects': cannot be empty, need at least" +
 			" one object for batching")
 
-		_, err := manager.AddObjects(ctx, nil, []*models.Object{}, []*string{}, nil)
+		_, err := manager.AddObjects(ctx, nil, []*models.Object{}, []*string{}, nil, "")
 
 		assert.Equal(t, expectedErr, err)
 	})
@@ -108,7 +108,7 @@ func Test_BatchManager_AddObjects_WithNoVectorizerModule(t *testing.T) {
 				Return(nil, nil)
 		}
 
-		_, err := manager.AddObjects(ctx, nil, objects, []*string{}, nil)
+		_, err := manager.AddObjects(ctx, nil, objects, []*string{}, nil, "")
 		repoCalledWithObjects := vectorRepo.Calls[0].Arguments[0].(BatchObjects)
 
 		assert.Nil(t, err)
@@ -144,7 +144,7 @@ func Test_BatchManager_AddObjects_WithNoVectorizerModule(t *testing.T) {
 				Return(nil, nil)
 		}
 
-		_, err := manager.AddObjects(ctx, nil, objects, []*string{}, nil)
+		_, err := manager.AddObjects(ctx, nil, objects, []*string{}, nil, "")
 		repoCalledWithObjects := vectorRepo.Calls[0].Arguments[0].(BatchObjects)
 
 		assert.Nil(t, err)
@@ -184,7 +184,7 @@ func Test_BatchManager_AddObjects_WithNoVectorizerModule(t *testing.T) {
 				Return(nil, nil)
 		}
 
-		_, err := manager.AddObjects(ctx, nil, objects, []*string{}, nil)
+		_, err := manager.AddObjects(ctx, nil, objects, []*string{}, nil, "")
 		repoCalledWithObjects := vectorRepo.Calls[0].Arguments[0].(BatchObjects)
 
 		assert.Nil(t, err)
@@ -222,7 +222,7 @@ func Test_BatchManager_AddObjects_WithNoVectorizerModule(t *testing.T) {
 				Return(nil, nil)
 		}
 
-		_, err := manager.AddObjects(ctx, nil, objects, []*string{}, nil)
+		_, err := manager.AddObjects(ctx, nil, objects, []*string{}, nil, "")
 		repoCalledWithObjects := vectorRepo.Calls[0].Arguments[0].(BatchObjects)
 
 		assert.Nil(t, err)
@@ -255,7 +255,7 @@ func Test_BatchManager_AddObjects_WithNoVectorizerModule(t *testing.T) {
 				Return(nil, nil)
 		}
 
-		_, err := manager.AddObjects(ctx, nil, objects, []*string{}, nil)
+		_, err := manager.AddObjects(ctx, nil, objects, []*string{}, nil, "")
 		repoCalledWithObjects := vectorRepo.Calls[0].Arguments[0].(BatchObjects)
 
 		assert.Nil(t, err)
@@ -305,7 +305,7 @@ func Test_BatchManager_AddObjects_WithExternalVectorizerModule(t *testing.T) {
 		expectedErr := NewErrInvalidUserInput("invalid param 'objects': cannot be empty, need at least" +
 			" one object for batching")
 
-		_, err := manager.AddObjects(ctx, nil, []*models.Object{}, []*string{}, nil)
+		_, err := manager.AddObjects(ctx, nil, []*models.Object{}, []*string{}, nil, "")
 
 		assert.Equal(t, expectedErr, err)
 	})
@@ -328,7 +328,7 @@ func Test_BatchManager_AddObjects_WithExternalVectorizerModule(t *testing.T) {
 				Return(expectedVector, nil)
 		}
 
-		_, err := manager.AddObjects(ctx, nil, objects, []*string{}, nil)
+		_, err := manager.AddObjects(ctx, nil, objects, []*string{}, nil, "")
 		repoCalledWithObjects := vectorRepo.Calls[0].Arguments[0].(BatchObjects)
 
 		assert.Nil(t, err)
@@ -364,7 +364,7 @@ func Test_BatchManager_AddObjects_WithExternalVectorizerModule(t *testing.T) {
 				Return(nil, nil)
 		}
 
-		_, err := manager.AddObjects(ctx, nil, objects, []*string{}, nil)
+		_, err := manager.AddObjects(ctx, nil, objects, []*string{}, nil, "")
 		repoCalledWithObjects := vectorRepo.Calls[0].Arguments[0].(BatchObjects)
 
 		assert.Nil(t, err)
@@ -394,7 +394,7 @@ func Test_BatchManager_AddObjects_WithExternalVectorizerModule(t *testing.T) {
 				Return(nil, nil)
 		}
 
-		_, err := manager.AddObjects(ctx, nil, objects, []*string{}, nil)
+		_, err := manager.AddObjects(ctx, nil, objects, []*string{}, nil, "")
 		repoCalledWithObjects := vectorRepo.Calls[0].Arguments[0].(BatchObjects)
 
 		assert.Nil(t, err)
@@ -464,7 +464,7 @@ func Test_BatchManager_AddObjectsEmptyProperties(t *testing.T) {
 		modulesProvider.On("UpdateVector", mock.Anything, mock.AnythingOfType(FindObjectFn)).
 			Return(nil, nil)
 	}
-	addedObjects, err := manager.AddObjects(ctx, nil, objects, []*string{}, nil)
+	addedObjects, err := manager.AddObjects(ctx, nil, objects, []*string{}, nil, "")
 	assert.Nil(t, err)
 	require.Len(t, addedObjects, 2)
 	require.NotNil(t, addedObjects[0].Object.Properties)
diff --git a/usecases/objects/batch_delete_test.go b/usecases/objects/batch_delete_test.go
index 381ed75f5e..7297e1e085 100644
--- a/usecases/objects/batch_delete_test.go
+++ b/usecases/objects/batch_delete_test.go
@@ -171,7 +171,7 @@ func Test_BatchDelete_RequestValidation(t *testing.T) {
 		}
 
 		for _, test := range tests {
-			_, err := manager.DeleteObjects(ctx, nil, test.input.Match, test.input.DryRun, test.input.Output, nil)
+			_, err := manager.DeleteObjects(ctx, nil, test.input.Match, test.input.DryRun, test.input.Output, nil, "")
 			assert.Equal(t, test.expectedError, err.Error())
 		}
 	})
diff --git a/usecases/objects/fakes_for_test.go b/usecases/objects/fakes_for_test.go
index c634780835..6146bcf84e 100644
--- a/usecases/objects/fakes_for_test.go
+++ b/usecases/objects/fakes_for_test.go
@@ -186,7 +186,7 @@ func (f *fakeVectorRepo) PutObject(ctx context.Context, concept *models.Object,
 }
 
 func (f *fakeVectorRepo) BatchPutObjects(ctx context.Context, batch BatchObjects,
-	repl *additional.ReplicationProperties,
+	repl *additional.ReplicationProperties, tenantKey string,
 ) (BatchObjects, error) {
 	args := f.Called(batch)
 	return batch, args.Error(0)
@@ -200,7 +200,7 @@ func (f *fakeVectorRepo) AddBatchReferences(ctx context.Context, batch BatchRefe
 }
 
 func (f *fakeVectorRepo) BatchDeleteObjects(ctx context.Context, params BatchDeleteParams,
-	repl *additional.ReplicationProperties,
+	repl *additional.ReplicationProperties, tenantKey string,
 ) (BatchDeleteResult, error) {
 	args := f.Called(params)
 	return args.Get(0).(BatchDeleteResult), args.Error(1)
diff --git a/usecases/objects/merge_test.go b/usecases/objects/merge_test.go
index ba6f4e4499..554a8367d1 100644
--- a/usecases/objects/merge_test.go
+++ b/usecases/objects/merge_test.go
@@ -132,7 +132,7 @@ func Test_MergeObject(t *testing.T) {
 					},
 				},
 			},
-			wantCode:  StatusBadRequest,
+			wantCode:  StatusNotFound,
 			errExists: errAny,
 			stage:     stageAuthorization,
 		},
@@ -438,7 +438,7 @@ func Test_MergeObject(t *testing.T) {
 						ClassName: tc.previous.Class,
 						Vector:    tc.previous.Vector,
 					}, nil)
-			} else if tc.stage >= stageUpdateValidation {
+			} else if tc.stage >= stageAuthorization {
 				m.repo.On("Object", cls, uuid, search.SelectProperties(nil), additional.Properties{}).
 					Return((*search.Result)(nil), tc.errGetObject)
 			}
diff --git a/usecases/objects/validation/phone_numbers_test.go b/usecases/objects/validation/phone_numbers_test.go
index 01f484df57..b8cb7ac013 100644
--- a/usecases/objects/validation/phone_numbers_test.go
+++ b/usecases/objects/validation/phone_numbers_test.go
@@ -168,7 +168,7 @@ func TestPropertyOfTypePhoneNumberValidation(t *testing.T) {
 				},
 			}
 			schema := testSchema()
-			err := validator.properties(context.Background(), obj, schema.Objects.Classes[0])
+			err := validator.properties(context.Background(), schema.Objects.Classes[0], obj, nil)
 			assert.Equal(t, test.expectedErr, err)
 			if err != nil {
 				return